Flink剖析|No.4 Kafka分区数据Skew导致Watermark放赖怎么办?
抛出疑无路?
有一种非常..非常…常见的痛苦是Kafka分区数据Skew,由于某一个分区数据缓慢导致整个作业无法事件驱动计算。
From @孙金城的知识星球用户,如下:
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。
众人拾柴
期待你典型问题的抛出… 我将知无不言…言无不尽… 我在又一村等你…
关注我的订阅号,我们并肩续航…