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

  `包下面，與你的業務程式碼一起打包：

  http://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：

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

把這個專案拉到本地，按照README.md 體驗一把：

http://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。

作者介紹

孫金城，51CTO社群編輯，Apache Flink PMC 成員，Apache Beam Committer，Apache IoTDB PMC 成員，ALC Beijing 成員，Apache ShenYu 導師，Apache 軟體基金會成員。關注技術領域流計算和時序資料儲存。