[[429430]]

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

了解背壓

什么是背壓

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

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

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

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

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

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

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

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

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

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

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

(3)影響state的大小，還是因?yàn)閏heckpoint barrier對齊要求。導(dǎo)致state變大。

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

如何查找定位背壓

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

(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)控進(jìn)一步判斷。

原理：BackPressure界面會(huì)周期性的對Task線程棧信息采樣，通過線程被阻塞在請求Buffer的頻率來判斷節(jié)點(diǎn)是否處于背壓狀態(tài)。計(jì)算緩沖區(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òng)慢，導(dǎo)致checkpoint生成時(shí)間長, 出現(xiàn)超時(shí)的現(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高，則表明它有可能是背壓的根源。因?yàn)橥ǔ１硥簳?huì)傳導(dǎo)至其上游，導(dǎo)致上游某些 Subtask 的 outPoolUsage 為高。

inputFloatingBuffersUsage 與 inputExclusiveBuffersUsage

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

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

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

背壓的原理

基于 Credit-based Flow Control的背壓機(jī)制

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

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

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

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

圖集流程上面流程

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

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

(3)SubTask b 會(huì)向上游 SubTask a 反饋 Channel Credit = 2。然后上游下一次最多只給下游發(fā)送 2 個(gè) 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 背壓機(jī)制，這里不再闡述。

解決背壓

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

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

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

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