我的目标是使用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”流)转换中完成? 此外,由于数据集正在更新,我希望对它...