问题标签 [apache-flink]

我想知道 Spark 或 Flink 执行引擎（主调度程序）如何为每个块找到合适的工作程序。

namenode 将能够告诉他们块的确切位置，但这项任务是由 Spark 和 Flink 的作业管理器完成的，还是 YARN 发挥作用的东西？

我正在努力寻找解决以下问题的方法，使用 Apache Flink：

我有一个向量流，由本地文件夹中的文件表示。使用 定位新文本文件后DataStream<String> text = env.readFileStream(...)，我将 (flatMap)、Input 转换为 a SingleOutputStreamOperator<Tuple2<String, Integer>, ?>，其中 Integer 是来自评分函数的分数。

我想保留一个包含前 k 个向量的全局 HashMap，使用它们的分数。我使用有状态的转换来解决这个问题。

1. 我遇到的第一个问题是 HashMap 保留每个接收器的数据（因此对于每个工作线程，一个 HashMap 数据）。我怎样才能使它成为一个全球集合

2. 使用 Apache Spark，我通过 JavaPairDStream<String, Integer> stateDstream = tuples.updateStateByKey(updateFunction);

然后在stateDstream. 有没有办法使用 FLink 获得相同的功能？

提前致谢！

使用 Eclipse IDE 将 Flink 提交到 Yarn 单节点集群我得到：org.apache.flink.client.program.ProgramInvocationException: Failed to resolve JobManager

Jar 将文件从 Hdfs 中的一个位置复制到另一个位置，并在集群上本地执行的工作文件，例如 sudo bin/flink run -c Test ./examples/MyJar.jar hdfs://localhost:9000/flink/in.txt hdfs ://localhost:9000/flink/out.txt

代码：

我的目标是使用 Kafka 作为源和 Flink 作为流处理引擎来设置一个高吞吐量的集群。这就是我所做的。

我已经在主节点和工作节点上设置了一个 2 节点集群，配置如下。

掌握 flink-conf.yaml

工人 flink-conf.yaml

Master 节点上的slaves文件如下所示：

两个节点上的 flink 设置位于同名文件夹中。我通过运行在主服务器上启动集群

这将启动 Worker 节点上的任务管理器。

我的输入源是 Kafka。这是片段。

这是我的接收器功能

这是我的 pom.xml 中的 Flink 依赖项。

然后我在master上用这个命令运行打包的jar

SinkFunction但是，当我将消息插入 Kafka 主题时，我能够仅在主节点上考虑来自我的 Kafka 主题的所有消息（通过我的实现的调用方法中的调试消息）。

在作业管理器 UI 中，我可以看到 2 个任务管理器，如下所示：

仪表板也看起来像这样： 问题：

1. 为什么工作节点没有收到任务？
2. 我错过了一些配置吗？

我对 Flink 有一个奇怪的问题：在工作中，我必须用预编译的资源文件初始化一个对象。并且在作业开始后，它在第一时间运行没有问题。但是当我再次像第一次一样使用 Web 界面访问它时，我的对象无法初始化，错误是：

java.nio.file.NoSuchFileException: hdfs:/.../.../my_file

这让我感到困惑，因为文件在那里并且对象第一次成功初始化。相关代码为：

如果资源在本地系统上并且我运行具有相同作业的本地服务器，我不会遇到任何问题。那么有人有任何线索吗？

编辑：完整跟踪

flink流中是否可以知道本地窗口属于哪个子任务？我想getRuntimeContext().getIndexOfThisSubtask()在实现中使用该方法TriggerPolicy。

关于构建 flink-avro 的问题。

我下载了源代码，我能够构建完整的源代码。

但是当我移动到文件夹时

并运行命令mvn package，因为我想编译和运行测试，它开始下载最新的 flink SNAPSHOT jars。然而，我从源代码构建了相同的罐子，一切都在本地回购中。

为什么它会下载本地构建的相同内容？

这是将 Flink fold 与 scala 匿名函数一起使用的无效尝试：

它编译得很好，但在执行时，我得到一个“类型擦除问题”（见下文）。在 Java 中这样做很好，但当然更冗长。我喜欢简洁明了的 lambda。我怎么能在scala中做到这一点？

我编写了一个小型 Scala 程序，它使用 Apache Flink Streaming API 来读取 Twitter 推文。

执行时遇到以下问题：

程序尝试重新建立连接。因此，这 4 行日志消息继续发出。

奇怪的是，当我运行 Apache Flink 项目中提供的示例时，一切正常（我从 GitHub 中提取了最新版本的 master）。我什至使用相同的属性文件。如果我将该示例类复制到我自己的项目中，也会出现上述问题状态。

我使用 Flink 原型来创建我自己的项目。我尝试了 0.9.1 和 0.10-SNAPSHOT 版本。依赖项、 和flink-scala用于flink-streaming-scala对应版本。flink-clientsflink-connector-twitter

有没有人遇到过类似的问题并且可以让我走上正轨？

我尝试联合一些 flink 数据集。它们包含在 Seq. 以下是产生问题的代码

我得到的是一个

线程“主”org.apache.flink.api.common.InvalidProgramException 中的异常：无法联合不同类型的输入。Input1=scala.Tuple2(_1: Integer, _2: Option[scala.Tuple4(_1: GenericType [java.time.LocalDateTime], _2: String, _3: Integer, _4: Boolean)]), input2=scala.Tuple2( _1：整数，_2：选项[scala.Tuple4（_1：GenericType[java.time.LocalDateTime]，_2：字符串，_3：整数，_4：布尔值）]）

我错过了什么？类型没有区别，是吗？工会运营商应该是便宜的，所以绕过这个问题似乎没有吸引力。我提供了前两行代码作为 DataSet 中数据类型相同的参数。我使用了 flink 版本 0.9.0 和 0.9.1