Flink 從入門到精通?系列文章

工作或者面試中一般都要求面試者有較強的獨立解決問題的能力，解決問題的前提是：我們對相應組件的原理非常清楚。本文先講述原理，再結合實戰分析一個線上任務的異常案例。

本文分以下幾個部分：

• 第一部分直接給出結論

• 第二部分會分析原理：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 繼續踏踏實實的做好自己

公眾號(zhisheng)里回復?面經、ClickHouse、ES、Flink、?Spring、Java、Kafka、監控?等關鍵字可以查看更多關鍵字對應的文章。點個贊+在看，少個 bug?????

總結

以上是生活随笔為你收集整理的Flink 中的木桶效应：单个 subtask 卡死导致整个任务卡死的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。