1.概述
轉載：Flink 中的木桶效應：單個 subtask 卡死導緻整個任務卡死
工作或者面試中一般都要求面試者有較強的獨立解決問題的能力，解決問題的前提是：我們對相應元件的原理非常清楚。本文先講述原理，再結合實戰分析一個線上任務的異常案例。
本文分以下幾個部分：
第一部分直接給出結論
第二部分會分析原理：Flink 中單個 subtask 卡死，為什麼會導緻整個任務卡死？
第三部分：線上業務如果出現類似問題如何定位？換言之，線上出現哪些現象可以說明是單個 subtask 導緻整個任務卡住了。會通過案例結合 Metric jstack 等定位問題的根本原因。
第四部分對 Flink 現有實體分區政策的思考
第五部分總結
1、結論
keyBy 或 rebalance 下遊的算子，如果單個 subtask 完全卡死，會把整個 Flink 任務卡死
通過反壓可以确定哪個 Task 出現性能瓶頸
通過 inPoolUsage 名額可以确定下遊 Task 的哪個 Subtask 出現性能瓶頸
Flink 現有的實體分區政策全是靜态的負載均衡政策，沒有動态根據負載能力進行負載均衡的政策
2、原理分析
2.1 分析一個簡單的 Flink 任務
如下圖所示，任務由 Source → map → KeyBy → Sink 四個算子組成。其中 keyBy 和 Sink 算子之間存在 shuffle，圖中相同顔色的箭頭表示到達 Sink 中相同的 subtask。其中 subtask A0 和 A1 都要給 TaskB 的 3 個 subtask 發送資料。
任務執行圖
2.2 任務運作過程中，具體的資料傳輸過程
如下圖所示，上遊每個 Subtask 中會有 3 個 resultSubPartition，連接配接下遊算子的 3 個 subtask。下遊每個 subtask 會有 2 個 InputChannel，連接配接上遊算子的 2 個 subtask。在正常運作過程中如果沒有反壓，所有的 buffer pool 是用不完的。就像下圖一樣，所有的 InputChannel 并沒有占滿，公共的 buffer pool 中也幾乎沒有資料。
正常的資料傳輸
2.3 Subtask B0 卡死後資料傳輸發生的現象
假設由于某些原因 Subtask B0 長時間地處理非常慢甚至卡死，其他的 Subtask 都正常，會出現下圖中的現象。
下遊其中一個 subtask 反壓嚴重
2.3.1 現象描述
1、Subtask B0 内的 A0 和 A1 兩個 InputChannel 會被占滿
2、Subtask B0 公共的 BufferPool 中可申請到的空間也被占滿
3、Subtask A0 和 A1 的 B0 ResultSubPartition 被占滿
4、Subtask A0 和 A1 公共的 BufferPool 中可申請到的空間也被占滿
5、Subtask B1 和 B2 的所有 InputChannel 和 BufferPool 都是空的
6、Subtask A0 和 A1 的 B1、B2 ResultSubPartition 都是空的
2.3.2 現象解釋 ☆☆☆☆☆
Subtask B0 卡死了，不再處理資料或者處理的超級慢。上遊如果一直給 Subtask B0 發送資料，必然會導緻 Subtask B0 的所有 InputChannel 占滿，最後導緻公共的 BufferPool 中可申請到的空間也被占滿。也就是現象中的 1、2 兩點。
雖然 Subtask B0 的所有 Buffer 占滿後，Subtask A0 和 A1 仍然在生産資料，此時必然不能發送資料到 B0，是以就會把 Subtask A0 和 A1 中 Subtask B0 對應的 buffer 給占滿（也就是 Flink 中反壓傳遞的過程），最後再把 Subtask A0 和 A1 公共的 BufferPool 中可申請到的空間也占滿。也就是現象中的 3、4 兩點。
其中 1、2、3、4 這四點比較容易了解，關鍵是 5、6 兩點，即：Subtask B0 卡死會什麼導緻 Subtask B1 和 B2 完全沒有資料了？
Subtask B1 和 B2 在整個上下遊的 buffer 都是空的，理論來講隻要有空餘的 buffer，就可以用來傳輸資料。但實際上并沒有将 Subtask A0 和 A1 的資料傳輸給 Subtask B1 和 B2。
「這裡的根本原因是：
Subtask A0 和 A1 的主線程完全卡死壓根不會生産資料了。
」
既然不會生産資料了，那麼 Subtask A0 和 A1 的下遊就算 buffer 空着，也是沒有意義的。是以就出現 5、6 的現象。
重點解釋：為什麼 Subtask A0 和 A1 的主線程會卡死？A0 和 A1 是一樣的，下面單獨分析 A0。Subtask A0 處理資料流程圖如下所示：
’
Subtask A0 處理資料流程
Subtask A0 的主線程會從上遊讀取資料消費，按照資料的 KeyBy 規則，将資料發送到 B0、B1、B2 三個 outputBuffer 中。現在我們可以看到 B0 對應的 buffer 占滿了，且 B0 在公共的 BufferPool 中可申請到的空間也被占滿。現在主線程在處理資料，假設這條資料根據 KeyBy 分區規則後，應該配置設定給 Subtask B0 處理，那麼主線程必須把資料放到 B0 這個 buffer 中。但是現在 buffer 沒有空間了，是以主線程就會卡在申請 buffer 上，直到可以再申請到 buffer（這也是 Flink 反壓的實作原理）。
同理 Subtask A1 也會出現這樣的問題，如果 Task A 的并行度是 1000，那麼 Subtask B0 也會将上遊 1000 個 Subtask A 全部卡住。最後導緻整個任務全部卡住。
原理弄懂了，下一階段要搞懂線上出現哪些現象可以說明是單個 subtask 導緻整個任務卡住了。線上業務如果出現類似問題如何定位？
2.4 小結
其實不隻是 keyBy 場景會出現上述問題，rebalance 場景也會出現上述問題。rebalance 分區政策表示，上遊 subtask 以輪詢的政策向下遊所有 subtask 發送資料，即：subtask A0 會先給 subtask B0 發一條，下一條發給 B1，下一條再發給 B2，再發給 B0 依次類推：B0、B1、B2、B0、B1、B2、B0、B1、B2。。。
一旦 B0 卡死，最終主線程肯定因為 B0 把 Subtask A 内的 buffer 用完了，導緻主線程卡住。
「是以總結成一句話就是：
keyBy 或 rebalance 下遊的算子，如果單個 subtask 完全卡死，會把整個 Flink 任務卡死
。」
3、問題定位過程
3.1 業務場景
業務回報一個寫 ES 的任務跑一會就沒輸出了，完全卡死，一條輸出都沒有。80 并發完全正常，可以正常輸出，調大并發到 100 以後，運作一會就沒有輸出了。
DAG 圖如下所示，上下遊算子之間的資料分區政策是 rebalance。
任務 DAG
3.2 思考及定位過程
聽到業務方的回報，看到作業 DAG 圖，開始定位問題，筆者并沒有想到第二部分那麼多的原理分析，因為大部分的任務卡住并不是因為單個 Subtask 卡住導緻整個任務卡住。是以下面的定位過程完全是以一個旁觀者的角度觸發，也是筆者當時定位問題的一個完整過程。筆者作為平台方，也是完全不清楚業務邏輯的，隻是從 Flink 的角度來定位問題。
3.2.1 從 DAG 上來看任務有兩個 Task，到底是上遊 Task 有問題還是下遊 Task 有問題
如何定位上遊 Task 還是下遊有問題很簡單：看一下上遊 Task 是否有反壓，如果下遊 Task 卡死或者消費慢，上遊 Task 肯定反壓比較嚴重。是以判定依據：
- 如果上遊 Task 反壓嚴重，則表示下遊 Task 有問題
- 如果上遊 Task 沒有反壓，大機率是上遊 Task 有問題
檢視後，發現上遊 Task 的所有 Subtask 反壓都非常嚴重，是以斷定下遊 Task 有問題。
3.2.2 下遊 Task 發生了什麼？在幹嘛？
要想知道下遊 Task （ES Sink）在幹嘛，很簡單：檢視現場，随便選一個 Subtask 打個 jstack，看看目前程序在做什麼。
下遊 Task 總共 100 個并行度，随便找了一個 Subtask 打 jstack，發現目前 Subtask 處理資料的主線程卡在 poll 資料。即：ES Sink 的目前 Subtask 不輸出資料竟然是因為上遊不發送資料了。為了确認目前 Subtask 接收不到上遊算子發送的資料，又看了目前 Subtask 的 Metric：inPoolUsage。inPoolUsage 表示目前算子輸入緩沖區的使用率，inPoolUsage 持續為 0 證明了目前 Subtask 确實接收不到上遊發送的資料。
讀者在這裡是不是開始懷疑了，是不是上遊出問題導緻整個任務的下遊都接收不到資料？
答：不可能。如果上遊 Task 出問題，所有下遊 Subtask 都是正常的，都在接收上遊發送資料，那麼上遊算子的 buffer 肯定是空的，怎麼可能出現反壓。是以上遊算子反壓嚴重必然是下遊算子處理性能不行。
到這裡，經過上述一步步推導，才開始想本文第二部分那些原理分析：是不是下遊 Task 有某幾個 Subtask 卡住了，導緻整個任務卡住了。問題來了：怎麼找出下遊 Task 那幾個可能卡住的 Subtask？
3.2.3 怎麼找出下遊 Task 那幾個可能卡住的 Subtask？
本文第二部分分析過，如果下遊 Task 某幾個 Subtask 卡死，那麼這幾個 Subtask 的 inputBuffer 會被占滿，且其他的 Subtask inputBuffer 全為空。是以我們隻需要找出下遊哪幾個 Subtask 的 inputBuffer 占滿了，也就是出現卡頓的 Subtask。
此時需要 Flink 強大的 Metric，Flink 的 Metric 可以看到下遊 Task 所有 Subtask 的 inPoolUsage，Flink 的 Web UI 可以看到 Metric 項。下遊 Task 有 100 個并行度，即：對應 100 個 Subtask。最笨的方式是分别檢視 100 個 Subtask 的 inPoolUsage 名額。
高效的方式是建構好整個 Flink 的 Metric 系統，通過 Metric Report 将各種名額收集到外部存儲系統，用 Grafana 或其他可視化工具展示。此時根據 Top N 的方式去查詢即可。按照 subtask_id 分組，對 inPoolUsage 排降序，找出 inPoolUsage 最高的幾個 subtask_id。
利用上述方案發現隻有一個 subtask 的 inPoolUsage 為 1，其餘的 subtask 的 inPoolUsage 都為 0。此時就可以得出結論了：确實是因為單個 Subtask 導緻整個任務卡住了。
3.2.4 解決方案
定位到具體的 Subtask，jstack 發現該 Subtask 的主線程卡在了 ES Sink 的某一處代碼。具體 ES 的問題這裡不分析了，是 ES 用戶端的 bug 導緻卡死。檢視了 ES 的相關 issue 将代碼合入後問題最終解決。
3.2.5 小結
第三部分主要是從 Flink 的角度出發，使用一種通用的方法論來定位到任務被卡死的真正原因。定位問題需要有全面的理論支撐結合強大的 Metric 系統輔助定位問題。
4、 關于 Flink 實體分區政策的思考
Flink 支援的實體分區政策
Flink 的實體分區政策支援多種，包括：partitionCustom、shuffle、rebalance、rescale、broadcast。具體參考：https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/#physical-partitioning。
其實前四種分區政策都可以認為是一種負載均衡政策，上遊算子 n 個并行度，下遊算子 m 個并行度，如何将上遊 n 個 Subtask 的資料打散到下遊 m 個 Subtask 呢？
partitionCustom 表示自定義分區政策，根據使用者自定義的分區政策發送資料
shuffle 表示随機的政策實作負載均衡。
rebalance 表示輪詢政策。
rescale 是對 rebalance 政策的優化。引用官網 rescale 圖示，相比 rebalance 而言，使用 rescale 政策時，上遊 Subtask 隻會給下遊某幾個 Subtask 發送資料。大大減少資料傳輸時邊的個數。
- 如下圖所示，Source 有兩個 Subtask，Map 有 6 個 Subtask，則一個 Source 的 Subtask 固定給 3 個 Map Subtask 發送資料。
- 如果是 rebalance，每個 Source 都會給所有的 Map Subtask 發送資料。
Flink 欠缺的一種負載均衡政策
上述幾種實體分區政策都是靜态的，而不是動态的。如下圖所示是 rebalance shuffle 圖示，上遊 Task A 的所有 Subtask 要發送資料給下遊 Task B 的所有 Subtask。假設 Subtask B0 沒有卡死，但是由于資源競争等原因，Subtask B0 的吞吐比 B1 和 B2 要差。但是 rebalance 是嚴格的輪詢政策，是以上遊給 Subtask B0、B1、B2 發送的資料量完全一緻。最後 B0 就會拖慢整個任務的吞吐量，B1 和 B2 也不能發揮出自己真正的性能。
rebalance shuffle
對于這種問題，常用的負載均衡政策并不是使用随機或者輪詢政策，而是上遊發送資料時會檢測下遊的負載能力，根據不同的負載能力，給下遊發送不同的資料量。假設下遊 Subtask B1 和 B2 吞吐量高于 B0，那麼上遊 Subtask A 會多給 B1 和 B2 發送一些資料，少給 B0 發送一些資料。
該政策可以解決 rebalance 政策導緻的木桶效應。但該政策不能解決 KeyBy 的場景，因為 KeyBy 政策嚴格決定了每條資料要發送到下遊哪個 Subtask。
5、總結
再次回顧第一部分的結論：
keyBy 或 rebalance 下遊的算子，如果單個 subtask 完全卡死，會把整個 Flink 任務卡死
通過反壓可以确定哪個 Task 出現性能瓶頸
通過 inPoolUsage 名額可以确定下遊 Task 的哪個 Subtask 出現性能瓶頸
Flink 現有的實體分區政策全是靜态的負載均衡政策，沒有動态根據負載能力進行負載均衡的政策
基于 Apache Flink 的實時監控告警系統關于資料中台的深度思考與總結（幹幹貨）日志收集Agent，陰暗潮濕的地底世界
2020 繼續踏踏實實的做好自己