拋出疑無(wú)路？

有一種非常..非常...常見(jiàn)的痛苦是Kafka分區(qū)數(shù)據(jù)Skew，由于某一個(gè)分區(qū)數(shù)據(jù)緩慢導(dǎo)致整個(gè)作業(yè)無(wú)法事件驅(qū)動(dòng)計(jì)算。From @孫金城的知識(shí)星球用戶(hù)，如下：

示例說(shuō)明比如我們有一個(gè)Kafka的Topic，有2個(gè)分區(qū)，如下數(shù)據(jù)：

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 // 這條數(shù)據(jù)在第一個(gè)分區(qū)，其他數(shù)據(jù)在第二個(gè)分區(qū)。
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條數(shù)據(jù)到第一個(gè)分區(qū)，其他的在第二個(gè)分區(qū)。這時(shí)候，如果業(yè)務(wù)需求是一個(gè)5秒鐘的窗口。

那么，目前Flink-1.10默認(rèn)只能觸發(fā)4個(gè)窗口計(jì)算，也就是從22條數(shù)據(jù)到39條數(shù)據(jù)都不會(huì)觸發(fā)計(jì)算了。利用本篇提及的解決方案可以完成

7個(gè)窗口的觸發(fā)（全部窗口）。

不考慮Idle情況，計(jì)算結(jié)果 如下：

考慮Idle情況，計(jì)算結(jié)果 如下：

再現(xiàn)又一村！

【Flink 1.10 】這又是一個(gè)知道1秒鐘，不知道坐地哭的情況。問(wèn)題的本質(zhì)是目前生成Watermark的機(jī)制是min(partition1, partition2,..,partitionN), 所以就出現(xiàn)了木桶效應(yīng)，也就是用戶(hù)描述的情況，怎么辦呢？修改代碼.... 還是那句話，看這個(gè)系列的朋友都是來(lái)看怎么快速解決問(wèn)題的，所以咱們不啰嗦，直接看解決步驟：

• 仿照下面的代碼開(kāi)發(fā)一個(gè)`StreamSource`， 放到`org.apache.flink.streaming.api.operators`包下面，與你的業(yè)務(wù)代碼一起打包：
  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

注意上面添加了一個(gè)配置`idleTimeout`的配置項(xiàng)，這個(gè)配置下默認(rèn)`-1`，也就是不生效，那么只要你配置了這個(gè)數(shù)值，指定的時(shí)間不來(lái)數(shù)據(jù)Flink系統(tǒng)就認(rèn)為這個(gè)Partition沒(méi)數(shù)據(jù)了，那么計(jì)算Watermark的時(shí)候就不考慮他了，等他有數(shù)據(jù)再把他列入計(jì)算Watermark的范疇。

• 在寫(xiě)作業(yè)的時(shí)候配置`source.idle.timeout.ms`參數(shù)，如下：

OK，上面兩個(gè)步驟就解決了這個(gè)問(wèn)題。如你遇到classloader問(wèn)題，我說(shuō)的是如果，那么把下面默認(rèn)值進(jìn)行修改。

【說(shuō)明】如上解決方案適用 Flink 1.10 及之前版本 DataStream 和SQL flink planner開(kāi)發(fā)（我想以后也一樣，因?yàn)閒link planner 逐步被blink planner替代）。

對(duì) Flink blink planner SQL (1.9+) 可以添加`table.exec.source.idle-timeout`。 對(duì)于Flink 1.11及之后的DataStrem可以利用`WatermarkStrategy`進(jìn)行設(shè)置，最終參考1.11發(fā)布之后的文檔。

前進(jìn)一小步？

如果是已經(jīng)遇到這個(gè)問(wèn)題的朋友，那么按照上面兩步應(yīng)該可以解決問(wèn)題。如果你沒(méi)有遇到這個(gè)問(wèn)題，想自己體驗(yàn)一下，那么可以clone我的git：

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

把這個(gè)項(xiàng)目拉到本地，按照README.md 體驗(yàn)一把：

https://github.com/sunjincheng121/know_how_know_why/blob/master/QA/v110/discover-idle-sources/src/main/java/qa/README.md

如果你上面操作還遇到了困難，那也不用著急，關(guān)注我《Apache Flink知其然，知其所以然》視頻課程，里面會(huì)有視頻演示（這個(gè)系列文章保持簡(jiǎn)單，只說(shuō)How，不細(xì)說(shuō)Why）

Flink 的鍋？...

關(guān)于這個(gè)問(wèn)題社區(qū)也在不斷的做努力，感興趣的朋友可以參閱 FLIP-27&FLIP-126。當(dāng)然對(duì)于flink planner（old）目前看只能用本篇提到的方案進(jìn)行解決，這里也建議大家盡早升級(jí)到 blink planner。

作者介紹

孫金城，51CTO社區(qū)編輯，Apache Flink PMC 成員，Apache Beam Committer，Apache IoTDB PMC 成員，ALC Beijing 成員，Apache ShenYu 導(dǎo)師，Apache 軟件基金會(huì)成員。關(guān)注技術(shù)領(lǐng)域流計(jì)算和時(shí)序數(shù)據(jù)存儲(chǔ)。