首先，我读过这篇关于同一问题的文章，并尝试遵循适用于他的相同解决方案（使用 mvn 创建一个新的快速入门并将代码迁移到那里），并且在开箱即用时无法正常工作IntelliJ。
这是我的 pom.xml 与其他 pom.xml 的依赖项混合在一起。我究竟做错了什么？
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable …我的要求是在一天内传输数百万条记录,并且它对外部配置参数有很大的依赖性.例如,用户可以随时在Web应用程序中更改所需的设置,并且在进行更改后,必须使用新的应用程序配置参数进行流式处理.这些是应用级配置,我们还有一些动态排除参数,每个数据都必须通过并过滤.
我看到flink没有全局状态,它在所有任务管理器和子任务中共享.拥有一个集中式缓存是一个选项,但对于每个参数,我必须从缓存中读取它,这将增加延迟.请告知更好的方法来处理这些场景以及其他应用程序如何处理它.谢谢.
我遇到了几个可序列化的异常，并且我在Flink的互联网和文档上进行了一些搜索；有一些著名的解决方案，如瞬态、扩展序列化等。每次异常的起源都非常清楚，但就我而言，我无法找到它到底在哪里没有序列化。
问：遇到这种异常应该如何调试？
A.斯卡拉：
class executor ( val sink: SinkFunction[List[String]] {
def exe(): Unit = {
xxx.....addSink(sinks)
}
}
B.scala：
class Main extends App {
def createSink: SinkFunction[List[String]] = new StringSink()
object StringSink {
// static
val stringList: List[String] = List()
}
// create a testing sink
class StringSink extends SinkFunction[List[String]] {
override def invoke(strs: List[String]): Unit = {
// add strs into the variable "stringList" of the compagin object StringSink
}
}
new executor(createSink()).exe()
// then do …我有一个场景,如果第二个事件在x秒内没有跟随第一个事件,我必须改变状态.例如,用户未在100分钟内注销,请认为他处于无效状态.如何使用当前的模式操作来设计它？
我设置了 Flink v1.2、3 个 JobManager、2 个 TaskManager。我想使用 hdfs 作为后端状态和检查点以及 Zookeeper storageDir
state.backend：文件系统
state.backend.fs.checkpointdir: hdfs:///[ip:port]/flink-checkpoints
state.checkpoints.dir: hdfs:///[ip:port]/external-checkpoints
高可用性：zookeeper
高可用性.zookeeper.storageDir: hdfs:///[ip:端口]/recovery
在 JobManager 中我记录了
2017-03-22 17:41:43,559 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: high-availability.zookeeper.client.acl, open
2017-03-22 17:41:43,680 ERROR org.apache.flink.runtime.jobmanager.JobManager - Error while starting up JobManager
java.io.IOException: The given HDFS file URI (hdfs:///ip:port/recovery/blob) did not describe the HDFS NameNode. The attempt to use a default HDFS configuration, as specified in the 'fs.hdfs.hdfsdefault' or 'fs.hdfs.hdfssite' config parameter failed due to the following problem: Either no …我们正处于构建物联网云平台项目的初期。有一些众所周知的部分可以实现完整的物联网平台解决方案。其中之一是实时规则处理/引擎系统，它需要了解流事件与最终用户以可读格式（SQL 或 Drools if/when/then 等）动态定义的任何规则相匹配。
我很困惑，因为互联网上有很多产品、项目（Storm、Spark、Flink、Drools、Espertech 等），所以考虑到我们有 3 人的开发团队（初级、中高级、高级），什么是最好的选择？
streaming rule-engine complex-event-processing iot flink-cep
我想在 flink 作业完成后执行一些任务，在 Intellij 中运行代码时没有任何问题，但是在 shell 文件中运行 Flink jar 时出现问题。我正在使用下面的行来确保 flink 程序的执行完成
//start the execution
JobExecutionResult jobExecutionResult = envrionment.execute(" Started the execution ");
is_job_finished = jobExecutionResult.isJobExecutionResult();
我不确定，上面的检查是否正确？
然后我在下面的方法中使用上面的变量来执行一些任务
if(print_mode && is_job_finished){
System.out.println(" \n \n -- System related variables -- \n");
System.out.println(" Stream_join Window length = " + WindowLength_join__ms + " milliseconds");
System.out.println(" Input rate for stream RR = " + input_rate_rr_S + " events/second");
System.out.println("Stream RR Runtime = " + Stream_RR_RunTime_S + " seconds");
System.out.println(" # raw events in stream …有没有一种方法可以确定从执行计划或其他方式运行作业所需的任务槽总数，而不必先实际启动作业。
根据此文档：https ://ci.apache.org/projects/flink/flink-docs-stable/concepts/runtime.html
“Flink 集群需要的任务槽数量与作业中使用的最高并行度完全相同。无需计算程序总共包含多少个任务（具有不同的并行度）。”
如果我从 StreamExecutionEnvironment 获取执行计划（设置后但没有实际执行作业）并从执行计划 json 中的节点列表中获取任何节点的最大并行度，这是否足以确定执行任务所需的任务槽数运行作业。
是否存在这种情况不再出现的情况？或者有什么注意事项需要牢记吗？
我正在寻找建议/最佳实践，以确定在 Flink 集群上部署流作业所需的最佳资源。
资源有
假设我有2种不同类型的数据流，一种提供天气数据，另一种提供车辆数据，我想使用Flink对数据进行复杂的事件处理。
Flink 1.3.x 中哪种方法是正确的使用方法？我看到了不同的方法，如 Union、Connect、Window Join。基本上我只是想尝试一个简单的 CEP，如下所示：
IF weather is wet AND vehicle speed > 60
WITHIN the last 10 seconds
THEN raise alert
谢谢！