我的目标是使用Kafka作为源和Flink作为流处理引擎来设置高吞吐量的集群。以下是我所做的事情。 我已经在主节点和工作节点上设置了2个节点集群，配置如下： 主节点 flink-conf.yaml： jobmanager.rpc.address: <MASTER_IP_ADDR&g...

如果有人给出用例示例来解释以下每个带有Apache flink的Watermark API之间的区别将会很有帮助： 定期水印 - AssignerWithPeriodicWatermarks[T] 断点水印 - AssignerWithPunctuatedWatermarks[T]

关于生成代理键，第一步是获取唯一的元素，然后为每个元组构建一个增量键。 所以我使用Java Set获取唯一元素，但它占用了堆空间。 然后，我使用Flink的distinct（）方法，完全解决了这个问题。 请问是什么导致了这种不同？ 另一个相关的问题是，Flink能否在mapper中生成代...

我想通过flatMap对DataSet.groupBy生成的每个组应用一个函数。尝试调用flatMap时，我收到了编译器错误： error: value flatMap is not a member of org.apache.flink.api.scala.GroupedDataSet ...

在下面的代码示例中，我正在尝试获取员工记录的流{ Country, Employer, Name, Salary, Age }并转储每个国家最高薪水的员工。不幸的是，Multiple KEY By无法使用。 仅KeyBy（Employer）反映出来，因此我没有得到正确的结果。我错过了什么？ ...

这是在亚马逊的Kinesis数据分析Flink环境中使用Flink 1.13.2运行的。 此应用程序正在运行Kafka主题。当主题具有较小的流量容量时，此应用程序可以正常运行，但当流量更大时，我会遇到此错误。如何进行故障排除、调整和修复？ 我看到了一些类似的SO问题，比如这个，但那显然是较...

当我执行Flink应用程序时，它会给我一个NullPointerException: 的错误提示。 2017-08-08 13:21:57,690 INFO com.datastax.driver.core.Cluster - New Cassandra host /127.0.0.1...

我正在尝试将Sentry附加到我们的Flink集群以跟踪作业执行。Sentry充当记录器，捕获消息并将其发送到中央服务器。默认情况下，它会捕获所有级别为WARN或更高的消息。 为了让Sentry捕获所有问题，我需要在运算符引发未捕获异常时编写一个WARN或ERROR日志消息。如果重启策略失败...

Flink在何时进行操作状态的序列化/反序列化？是在每次get/update时还是基于检查点？状态后端是否有影响？ 我怀疑，在具有多样化键（数百万）和每个键每秒数千个事件的键控流情况下，序列化/反序列化可能是一个大问题。我对吗？

join和coGroup转换可以读取2个输入数据集并输出一个结果数据集（“Y”流）（如果我理解错了，请纠正）。 我想要处理和更新2个数据集。为此，我计划使用2个coGroup转换。 但是，出于性能考虑，这两个转换是否可以在单个（“H”流）转换中完成？ 此外，由于数据集正在更新，我希望对它...