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

【Flink 1.10 】这又是一个知道1秒钟，不知道坐地哭的情况。

抛出疑无路？

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

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

S001,1, 2020-06-13 09:58:00
S001,1, 2020-06-13 09:58:01
S001,2, 2020-06-13 09:58:02
S001,3, 2020-06-13 09:58:03
S001,4, 2020-06-13 09:58:04
S001,5, 2020-06-13 09:58:05
S001,6, 2020-06-13 09:58:06
S001,7, 2020-06-13 09:58:07
S001,8, 2020-06-13 09:58:08
S001,9, 2020-06-13 09:58:09
S001,10, 2020-06-13 09:58:10
S001,11, 2020-06-13 09:58:11
S001,12, 2020-06-13 09:58:12
S001,13, 2020-06-13 09:58:13
S001,14, 2020-06-13 09:58:14
S001,15, 2020-06-13 09:58:15
S001,16, 2020-06-13 09:58:16
S001,17, 2020-06-13 09:58:17
S001,18, 2020-06-13 09:58:18
S001,19, 2020-06-13 09:58:19
S001,20, 2020-06-13 09:58:20
S001,21, 2020-06-13 09:58:21 // 这条数据在第一个分区，其他数据在第二个分区。
S001,22, 2020-06-13 09:58:22
S001,23, 2020-06-13 09:58:23
S001,24, 2020-06-13 09:58:24
S001,25, 2020-06-13 09:58:25
S001,26, 2020-06-13 09:58:26
S001,27, 2020-06-13 09:58:27
S001,28, 2020-06-13 09:58:28
S001,29, 2020-06-13 09:58:29
S001,30, 2020-06-13 09:58:30
S001,31, 2020-06-13 09:58:31
S001,32, 2020-06-13 09:58:32
S001,33, 2020-06-13 09:58:33
S001,34, 2020-06-13 09:58:34
S001,35, 2020-06-13 09:58:35
S001,36, 2020-06-13 09:58:36
S001,37, 2020-06-13 09:58:37
S001,38, 2020-06-13 09:58:38
S001,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), 所以就出现了木桶效应，也就是用户描述的情况，怎么办呢？修改代码.... 还是那句话，看这个系列的朋友都是来看怎么快速解决问题的，所以咱们不啰嗦，直接看解决步骤：

• 仿照下面的代码开发一个`StreamSource`， 放到`org.apache.flink.streaming.api.operators`包下面，与你的业务代码一起打包：
  https://github.com/sunjincheng121/know_how_know_why/blob/master/QA/v110/discover-idle-sources/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java

注意上面添加了一个配置`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：

https://github.com/sunjincheng121/know_how_know_why/tree/master/QA/v110/discover-idle-sources

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

https://github.com/sunjincheng121/know_how_know_why/blob/master/QA/v110/discover-idle-sources/src/main/java/qa/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 软件基金会成员。关注技术领域流计算和时序数据存储。