[[429430]]

本文轉(zhuǎn)載自微信公眾號「大數(shù)據(jù)左右手」，作者左右。轉(zhuǎn)載本文請聯(lián)系大數(shù)據(jù)左右手公眾號。

了解背壓

什么是背壓

在流式處理系統(tǒng)中，如果出現(xiàn)下游消費的速度跟不上上游生產(chǎn)數(shù)據(jù)的速度，就種現(xiàn)象就叫做背壓(backpressure,有人叫反壓，不糾結(jié)，本篇叫背壓)。本篇主要以Flink作為流式計算框架來簡單背壓機制，為了更好理解，只做簡單分享。

背壓產(chǎn)生的原因

下游消費的速度跟不上上游生產(chǎn)數(shù)據(jù)的速度，可能出現(xiàn)的原因如下：

(1)節(jié)點有性能瓶頸，可能是該節(jié)點所在的機器有網(wǎng)絡(luò)、磁盤等等故障，機器的網(wǎng)絡(luò)延遲和磁盤不足、頻繁GC、數(shù)據(jù)熱點等原因。

(2)數(shù)據(jù)源生產(chǎn)數(shù)據(jù)的速度過快，計算框架處理不及時。比如消息中間件kafka，生產(chǎn)者生產(chǎn)數(shù)據(jù)過快，下游flink消費計算不及時。

(3)flink算子間并行度不同，下游算子相比上游算子過小。

背壓導(dǎo)致的影響

首先，背壓不會直接導(dǎo)致系統(tǒng)的崩盤，只是處在一個不健康的運行狀態(tài)。

(1)背壓會導(dǎo)致流處理作業(yè)數(shù)據(jù)延遲的增加。

(2)影響到Checkpoint，導(dǎo)致失敗，導(dǎo)致狀態(tài)數(shù)據(jù)保存不了，如果上游是kafka數(shù)據(jù)源，在一致性的要求下，可能會導(dǎo)致offset的提交不上。

原理: 由于Flink的Checkpoint機制需要進行Barrier對齊，如果此時某個Task出現(xiàn)了背壓，Barrier流動的速度就會變慢，導(dǎo)致Checkpoint整體時間變長，如果背壓很嚴重，還有可能導(dǎo)致Checkpoint超時失敗。

(3)影響state的大小，還是因為checkpoint barrier對齊要求。導(dǎo)致state變大。

原理：接受到較快的輸入管道的barrier后，它后面數(shù)據(jù)會被緩存起來但不處理，直到較慢的輸入管道的barrier也到達。這些被緩存的數(shù)據(jù)會被放到state 里面，導(dǎo)致state變大。

如何查找定位背壓

(1)在web頁面發(fā)現(xiàn)fink的checkpoint生成超時, 失敗。

(2)查看jobmanager日志

2021-10-17 19：43：19,235 org.apache.flink.runtime.checkpoint.CheckpointCoordinator    
 
 
-Checkpoint 236663 of job d521558603f6ef25dfd053c665d6afbe expired before completing 

(3)在BackPressure界面直接可以看到。背壓狀態(tài)可以大致鎖定背壓可能存在的算子，但具體背壓是由于當(dāng)前Task自身處理速度慢還是由于下游Task處理慢導(dǎo)致的，需要通過metric監(jiān)控進一步判斷。

原理：BackPressure界面會周期性的對Task線程棧信息采樣，通過線程被阻塞在請求Buffer的頻率來判斷節(jié)點是否處于背壓狀態(tài)。計算緩沖區(qū)阻塞線程數(shù)與總線程數(shù)的比值 rate。其中，rate < 0.1 為 OK，0.1 <= rate <= 0.5 為 LOW，rate > 0.5 為 HIGH。

(4)Metrics 監(jiān)控背壓。緩沖區(qū)的數(shù)據(jù)處理不過來，barrier流動慢，導(dǎo)致checkpoint生成時間長, 出現(xiàn)超時的現(xiàn)象。input 和 output緩沖區(qū)都占滿。

outPoolUsage 與 inPoolUsage

指標(biāo)可能出現(xiàn)以下情況：

(1)outPoolUsage與inPoolUsage 都低，代表當(dāng)前Subtask正常。

(2)outPoolUsage與inPoolUsage 都高，代表當(dāng)前Subtask下游背壓。

(3)outPoolUsage 高，通常是被下游 Task 所影響。

(4)inPoolUsage高，則表明它有可能是背壓的根源。因為通常背壓會傳導(dǎo)至其上游，導(dǎo)致上游某些 Subtask 的 outPoolUsage 為高。

inputFloatingBuffersUsage 與 inputExclusiveBuffersUsage

指標(biāo)可能出現(xiàn)以下情況：

(1)floatingBuffersUsage高，則表明背壓正在傳導(dǎo)至上游。

(2)floatingBuffersUsage 高、exclusiveBuffersUsage 低，則表明了背壓可能存在傾斜。

背壓的原理

基于 Credit-based Flow Control的背壓機制

Credit 的反饋策略，保證每次上游發(fā)送的數(shù)據(jù)都是下游 InputChannel 可以承受的數(shù)據(jù)量。具體原理是這樣的：

(1)上游 SubTask 給下游 SubTask 發(fā)送數(shù)據(jù)時，會把 Buffer 中要發(fā)送的數(shù)據(jù)和上游 ResultSubPartition堆積的數(shù)據(jù)量 Backlog size 發(fā)給下游，下游接收到上游發(fā)來的 Backlog size 后，會向上游反饋現(xiàn)在的 Credit值，Credit 值表示目前下游可以接收上游的 Buffer 量，1 個Buffer 等價于 1 個 Credit。上游接收到下游反饋的Credit 值后，上游下次最多只會發(fā)送 Credit 個數(shù)據(jù)到下游，保障不會有數(shù)據(jù)積壓在 Socket 這一層。

(2)當(dāng)下游 SubTask 反壓比較嚴重時，可能就會向上游反饋 Channel Credit = 0，此時上游就知道下游目前對應(yīng)的InputChannel 沒有可用空間了，所以就不向下游發(fā)送數(shù)據(jù)了。

(3)上游會定期向下游發(fā)送探測信號，檢測下游返回的 Credit 是否大于 0，當(dāng)下游返回的 Credit 大于 0 表示下游有可用的Buffer 空間，上游就可以開始向下游發(fā)送數(shù)據(jù)了。

圖集流程上面流程

(1)上游 SubTask a 發(fā)送完數(shù)據(jù)后，還有 4 個 Buffer 被積壓，那么會把發(fā)送數(shù)據(jù)和 Backlog size = 4 一塊發(fā)送給下游 SubTask b。

(2)下游接受到數(shù)據(jù)后，知道上游積壓了 4 個Buffer，于是向 Buffer Pool 申請 Buffer，由于容量有限，下游 InputChannel 目前僅有 2 個 Buffer 空間。

(3)SubTask b 會向上游 SubTask a 反饋 Channel Credit = 2。然后上游下一次最多只給下游發(fā)送 2 個 Buffer 的數(shù)據(jù)，這樣每次上游發(fā)送的數(shù)據(jù)都是下游 InputChannel 的 Buffer 可以承受的數(shù)據(jù)量。

建議

參考官網(wǎng)【https://flink.apache.org/2019/07/23/flink-network-stack-2.html】

自行了解老版本TCP-based 背壓機制，這里不再闡述。

解決背壓

Flink不需要一個特殊的機制來處理背壓，因為Flink中的數(shù)據(jù)傳輸相當(dāng)于已經(jīng)提供了應(yīng)對背壓的機制。所以只有從代碼上與資源上去做一些調(diào)整。

(1)背壓部分原因可能是由于數(shù)據(jù)傾斜造成的，我們可以通過 Web UI 各個 SubTask 的 指標(biāo)值來確認。Checkpoint detail 里不同 SubTask 的 State size 也是一個分析數(shù)據(jù)傾斜的有用指標(biāo)。解決方式把數(shù)據(jù)分組的 key 預(yù)聚合來消除數(shù)據(jù)傾斜。

(2)代碼的執(zhí)行效率問題，阻塞或者性能問題。

(3)TaskManager 的內(nèi)存大小導(dǎo)致背壓。