问题标签 [flink-streaming]

我需要 Flink Streaming 方面的帮助。我在下面生成了一个简单的 Hello-world 类型的代码。这会从 RabbitMQ 流式传输 Avro 消息并将其持久化到 HDFS。我希望有人可以查看代码，也许它可以帮助其他人。

我为 Flink 流式传输找到的大多数示例都将结果发送到标准输出。我实际上想将数据保存到 Hadoop。我读到，理论上，你可以使用 Flink 流式传输到任何你喜欢的地方。实际上，我还没有找到任何将数据保存到 HDFS 的示例。但是，根据我找到的示例以及试验和错误，我提供了以下代码。

这里的数据源是 RabbitMQ。我使用客户端应用程序将“MyAvroObjects”发送到 RabbitMQ。MyAvroObject.java - 不包括 - 从 avro IDL 生成...可以是任何 avro 消息。

下面的代码使用 RabbitMQ 消息，并将其保存到 HDFS，作为 avro 文件......嗯，这就是我希望的。

如果您更喜欢 RabbitMQ 以外的其他来源，那么使用其他来源也可以正常工作。例如使用 Kafka 消费者：

问题：

1. 请查阅。这是将数据保存到 HDFS 的好习惯吗？

2. 如果流式传输过程引起问题怎么办，比如在序列化期间。它生成和异常，代码就退出了。Spark 流式传输依赖于 Yarn 自动重启应用。这也是使用 Flink 时的好习惯吗？

3. 我正在使用 FileSinkFunctionByMillis。我实际上希望使用 HdfsSinkFunction 之类的东西，但这并不存在。所以 FileSinkFunctionByMillis 是最接近这个的，这对我来说很有意义。同样，我发现的文档没有任何解释该怎么做，所以我只是猜测。

4. 当我在本地运行它时，我找到了一个类似“C:\hdfsroot_temporary\0_temporary\attempt__0000_r_000001_0”的目录结构，它是...... basare。这里有什么想法吗？

顺便说一句，当您想将数据保存到 Kafka 时，我可以使用...

提前谢谢了！！！！

在本地使用 flink 0.10.1 时，由于以下错误，我无法与 jobmanager 连接：

还有我的 pom.xml：

使用 flink 0.9.1 可以正常工作......我错过了什么？谢谢！

我通过以下方式跟踪了 Apache-flink 的实现：quick_start

我无法执行最后一项任务，即“分析结果”，因为 kmeans 文件夹中没有结果文件。

如果您查看上面 flink JobManager 的屏幕截图，您可以看到 KMeans Example 的 Status as FAILED。并且可能是由于此失败状态，kmeans 文件夹内没有结果文件。

现在单击 KMeans 示例，我得到以下可视化效果：

​</p>

下面是异常的截图：

​</p>

你能指导我我做错了什么吗？

假设我们要计算项目的总和和平均值，并且可以使用states或windows（时间）。

使用示例windows- https://ci.apache.org/projects/flink/flink-docs-release-0.10/apis/streaming_guide.html#example-program

使用示例states- https://github.com/dataArtisans/flink-training-exercises/blob/master/src/main/java/com/dataartisans/flinktraining/exercises/datastream_java/ride_speed/RideSpeed.java

我能问下做出决定的理由是什么吗？我是否可以推断，如果数据非常不规则地到达（50% 进入定义的窗口长度，而其他 50% 没有），窗口方法的结果更加有偏差（因为 50% 的事件被丢弃）？

另一方面，在处理状态时，我们是否会花费更多时间检查和更新状态？

我正在研究 Kafka Flink 集成实际上我已经完成了该集成，我使用 Flink API 在 Java 中编写了一个简单的字数统计程序，当我通过 java -jar myjarname 运行它时它运行良好，但是当我尝试使用 . /bin/flink 运行 myjarname 命令它给了我以下错误，

NoSuchMethodError:org.apache.flink.streaming.api.operators.isCheckpointingEnabled

受人尊敬的罐子在那里，但它仍然给了我以上错误。

我已经构建了一个自定义源来处理 Flink 中的日志流。该程序运行良好，并在处理记录后给了我想要的结果。

但是，当我检查 Web UI 时，我看不到计数。下面是截图： 记录/字节数

我在 Eclipse 中运行 flink，Maven 已经获取了必要的 jar。我的机器有一个具有八个内核的处理器，我必须编写的流应用程序从它的输入中读取行并计算一些统计数据。

当我在我的机器上运行程序时，我希望 flink 使用 CPU 的所有内核作为线程良好的代码。但是，当我观察核心时，我发现只使用了一个核心。我尝试了很多东西，最后一次尝试留下了以下代码，即设置环境的并行度。我还尝试单独为流设置它等等。

我使用 netcat 为程序提供了数据：

问题是如何使程序在本地扩展并使用所有可用的内核？

我正在尝试将 ContinuousTimeEventTrigger 与 timeWindow 一起使用：

但是我在使用以下stackTrace在flink流集群上执行它时遇到了java.lang.StackOverflowError：

有谁可以帮忙吗？

我有一个用 Flink 编写的流处理应用程序，我想使用来自状态后端的内部键值存储来计算流聚合。因为我要处理很多聚合，所以我想避免像当前提供的内存支持和文件支持的实现那样在 Flink 应用程序中将它们维护在堆上。相反，我想在 Apache Ignite 中维护状态缓存，这反过来可以使用直写和直读功能在 HBase 中提供更可靠的备份。

理想情况下，我会在每个物理节点上都有一个本地 Ignite 缓存，用于处理该节点上所有长时间运行的 Flink 操作符的状态。例如，每个节点在可用的 8 GB 容器中都有一个 Ignite 节点，无论它运行 1 个还是 10 个 Flink 操作符。

问题是我希望 Flink 和 Ignite 都在 YARN 上运行。通过一致性分区，我可以确保一般数据被发送到正确的缓存，并且在出现故障等情况下，可以从 HBase 重新填充。我面临的问题是 Ignite 似乎随机地从 YARN 请求容器，这意味着我不能保证实际上有可用的本地缓存，即使我将 Ignite 节点的数量设置为与物理节点的数量完全相同节点。

有关如何为每个物理节点设置一个 Ignite 节点的任何建议？

简而言之，我想从头开始对 Kafka 中的数据重新运行 Flink 管道。

Flink 0.10.2，卡夫卡 0.8.2。

我在 Kafka 中有一个保留 2 小时的推文主题，在 Flink 中有一个管道，它以每 10 秒 5 分钟的滑动窗口计数推文。

如果我中断管道并重新运行它，我希望它重新阅读较旧的推文，从而发出 5 分钟的推文计数。相反，它似乎从新到达的推文重新开始，因此需要 5 分钟才能计数“处于状态”。

我已经尝试了两个auto.offset.reset = smallest/earliest，并更改了group.id，但没有成功。我还尝试按照此处所述手动更改 Kafka 中的偏移量：https ://metabroadcast.com/blog/resetting-kafka-offsets

然后我假设该问题可能与 Flink 的检查点有关，但我不知道/找不到有关如何重置它的信息。

任何人都可以分享一些工作代码吗？谢谢，E。