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 軟體基金會成員。關注技術領域流計算和時序資料儲存。
- Spring中實現非同步呼叫的方式有哪些?
- 帶引數的全型別 Python 裝飾器
- 整理了幾個Python正則表示式,拿走就能用!
- 設計模式之狀態模式
- 如何實現資料庫讀一致性
- SOLID:開閉原則Go程式碼實戰
- React中如何引入CSS呢
- 慢查詢 MySQL 定位優化技巧,從10s優化到300ms
- 一個新視角:前端框架們都卷錯方向了?
- 編碼中的Adapter,不僅是一種設計模式,更是一種架構理念與解決方案
- 手寫程式語言-遞迴函式是如何實現的?
- 一文搞懂模糊匹配:定義、過程與技術
- 新來個阿里 P7,僅花 2 小時,做出一個多執行緒永動任務,看完直接跪了
- Puzzlescript,一種開發H5益智遊戲的引擎
- @Autowired和@Resource到底什麼區別,你明白了嗎?
- “四招”守護個人資訊保安
- CSS transition 小技巧!如何保留 hover 的狀態?
- React如此受歡迎離不開這4個主要原則
- 我是怎麼入行做風控的
- 重溫三十年前對於 NN 的批判:神經網路無法實現可解釋 AI