Flink剖析|No.4 Kafka分区数据Skew导致Watermark放赖怎么办?

抛出疑无路?

有一种非常..非常…常见的痛苦是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。
众人拾柴
期待你典型问题的抛出… 我将知无不言…言无不尽… 我在又一村等你…
关注我的订阅号,我们并肩续航…