Kafka分区数据Skew导致Watermark放赖怎么办？

Kafka分区数据Skew导致Watermark放赖怎么办？

抛出疑无路？

有一种非常..非常...常见的痛苦是Kafka分区数据Skew，由于某一个分区数据缓慢导致整个作业无法事件驱动计算。From @孙金城的知识星球用户，如下：

示例说明比如我们有一个Kafka的Topic，有2个分区，如下数据：

S001,1, 2020-06-13 09:58:00S001,1, 2020-06-13 09:58:01S001,2, 2020-06-13 09:58:02S001,3, 2020-06-13 09:58:03S001,4, 2020-06-13 09:58:04S001,5, 2020-06-13 09:58:05S001,6, 2020-06-13 09:58:06S001,7, 2020-06-13 09:58:07S001,8, 2020-06-13 09:58:08S001,9, 2020-06-13 09:58:09S001,10, 2020-06-13 09:58:10S001,11, 2020-06-13 09:58:11S001,12, 2020-06-13 09:58:12S001,13, 2020-06-13 09:58:13S001,14, 2020-06-13 09:58:14S001,15, 2020-06-13 09:58:15S001,16, 2020-06-13 09:58:16S001,17, 2020-06-13 09:58:17S001,18, 2020-06-13 09:58:18S001,19, 2020-06-13 09:58:19S001,20, 2020-06-13 09:58:20S001,21, 2020-06-13 09:58:21 // 这条数据在第一个分区，其他数据在第二个分区。S001,22, 2020-06-13 09:58:22S001,23, 2020-06-13 09:58:23S001,24, 2020-06-13 09:58:24S001,25, 2020-06-13 09:58:25S001,26, 2020-06-13 09:58:26S001,27, 2020-06-13 09:58:27S001,28, 2020-06-13 09:58:28S001,29, 2020-06-13 09:58:29S001,30, 2020-06-13 09:58:30S001,31, 2020-06-13 09:58:31S001,32, 2020-06-13 09:58:32S001,33, 2020-06-13 09:58:33S001,34, 2020-06-13 09:58:34S001,35, 2020-06-13 09:58:35S001,36, 2020-06-13 09:58:36S001,37, 2020-06-13 09:58:37S001,38, 2020-06-13 09:58:38S001,39, 2020-06-13 09:58:39

我们利用自定义Partitioner的方式，让第21条数据到第一个分区，其他的在第二个分区。这时候，如果业务需求是一个5秒钟的窗口。

那么，目前Flink-1.10默认只能触发4个窗口计算，也就是从22条数据到39条数据都不会触发计算了。利用本篇提及的解决方案可以完成

7个窗口的触发（全部窗口）。

不考虑Idle情况，计算结果 如下：

考虑Idle情况，计算结果 如下：

再现又一村！

【Flink 1.10 】这又是一个知道1秒钟，不知道坐地哭的情况。问题的本质是目前生成Watermark的机制是min(partition1, partition2,..,partitionN), 所以就出现了木桶效应，也就是用户描述的情况，怎么办呢？修改代码.... 还是那句话，看这个系列的朋友都是来看怎么快速解决问题的，所以咱们不啰嗦，直接看解决步骤：

注意上面添加了一个配置`idleTimeout`的配置项，这个配置下默认`-1`，也就是不生效，那么只要你配置了这个数值，指定的时间不来数据Flink系统就认为这个Partition没数据了，那么计算Watermark的时候就不考虑他了，等他有数据再把他列入计算Watermark的范畴。

在写作业的时候配置`source.idle.timeout.ms`参数，如下：

OK，上面两个步骤就解决了这个问题。如你遇到classloader问题，我说的是如果，那么把下面默认值进行修改。

【说明】如上解决方案适用 Flink 1.10 及之前版本 DataStream 和SQL flink planner开发（我想以后也一样，因为flink planner 逐步被blink planner替代）。

对 Flink blink planner SQL (1.9+) 可以添加`table.exec.source.idle-timeout`。 对于Flink 1.11及之后的DataStrem可以利用`WatermarkStrategy`进行设置，最终参考1.11发布之后的文档。

前进一小步？

如果是已经遇到这个问题的朋友，那么按照上面两步应该可以解决问题。如果你没有遇到这个问题，想自己体验一下，那么可以clone我的git：

把这个项目拉到本地，按照README.md 体验一把：

如果你上面操作还遇到了困难，那也不用着急，关注我《Apache Flink知其然，知其所以然》视频课程，里面会有视频演示（这个系列文章保持简单，只说How，不细说Why）

Flink 的锅？...

关于这个问题社区也在不断的做努力，感兴趣的朋友可以参阅 FLIP-27&FLIP-126。当然对于flink planner（old）目前看只能用本篇提到的方案进行解决，这里也建议大家尽早升级到 blink planner。

作者介绍

孙金城，社区编辑，Apache Flink PMC 成员，Apache Beam Committer，Apache IoTDB PMC 成员，ALC Beijing 成员，Apache ShenYu 导师，Apache 软件基金会成员。关注技术领域流计算和时序数据存储。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。