整理：楊濤（Flink 社區(qū)志愿者）

Checkpoint 機(jī)制

1.什么是 checkpoint

簡(jiǎn)單地說(shuō)就是 Flink 為了達(dá)到容錯(cuò)和 exactly-once 語(yǔ)義的功能，定期把 state 持久化下來(lái)，而這一持久化的過(guò)程就叫做 checkpoint ，它是 Flink Job 在某一時(shí)刻全局狀態(tài)的快照。

當(dāng)我們要對(duì)分布式系統(tǒng)實(shí)現(xiàn)一個(gè)全局狀態(tài)保留的功能時(shí)，傳統(tǒng)方案會(huì)引入一個(gè)統(tǒng)一時(shí)鐘，通過(guò)分布式系統(tǒng)中的 master 節(jié)點(diǎn)廣播出去給每一個(gè) slaves 節(jié)點(diǎn)，當(dāng)節(jié)點(diǎn)接收到這個(gè)統(tǒng)一時(shí)鐘時(shí)，它們就記錄下自己當(dāng)前的狀態(tài)即可。

但是統(tǒng)一時(shí)鐘的方式也存在一定的問(wèn)題，某一個(gè) node 進(jìn)行的 GC 時(shí)間比較長(zhǎng)，或者 master 與 slaves 的網(wǎng)絡(luò)在當(dāng)時(shí)存在波動(dòng)而造成時(shí)鐘的發(fā)送延遲或者發(fā)送失敗，都會(huì)造成此 slave 和其它的機(jī)器出現(xiàn)數(shù)據(jù)不一致而最終導(dǎo)致腦裂的情況。如果我們想要解決這個(gè)問(wèn)題，就需要對(duì) master 和 slaves 做一個(gè) HA（High Availability）。但是，一個(gè)系統(tǒng)越是復(fù)雜，就越不穩(wěn)定且維護(hù)成本越高。

Flink 是將 checkpoint 都放進(jìn)了一個(gè)名為 Barrier 的流。

上圖中就是一個(gè) Barrier 的例子，從上游的第一個(gè) Task 到下游的最后一個(gè) Task，每次當(dāng) Task 經(jīng)過(guò)圖中藍(lán)色的柵欄時(shí)，就會(huì)觸發(fā) save snapshot（快照）的功能。我們用一個(gè)例子來(lái)簡(jiǎn)單說(shuō)明。

2.實(shí)例分析

這是一個(gè)簡(jiǎn)單的 ETL 過(guò)程，首先我們把數(shù)據(jù)從 Kafka 中拿過(guò)來(lái)進(jìn)行一個(gè) trans 的轉(zhuǎn)換操作，然后再發(fā)送到一個(gè)下游的 Kafka

此時(shí)這個(gè)例子中沒(méi)有進(jìn)行 chaining 的調(diào)優(yōu)。所以此時(shí)采用的是 forward strategy ，也就是 “一個(gè) task 的輸出只發(fā)送給一個(gè) task 作為輸入”，這樣的方式，這樣做也有一個(gè)好處就是如果兩個(gè) task 都在一個(gè) JVM 中的話，那么就可以避免不必要的網(wǎng)絡(luò)開(kāi)銷

設(shè)置 Parallism 為 2，此時(shí)的 DAG 圖如下：

■ CK的分析過(guò)程

每一個(gè) Flink 作業(yè)都會(huì)有一個(gè) JobManager ，JobManager 里面又會(huì)有一個(gè) checkpoint coordinator 來(lái)管理整個(gè) checkpoint 的過(guò)程，我們可以設(shè)置一個(gè)時(shí)間間隔讓 checkpoint coordinator 將一個(gè) checkpoint 的事件發(fā)送給每一個(gè) Container 中的 source task，也就是第一個(gè)任務(wù)（對(duì)應(yīng)并行圖中的 task1，task2）。

當(dāng)某個(gè) Source 算子收到一個(gè) Barrier 時(shí)，它會(huì)暫停自身的數(shù)據(jù)處理，然后將自己的當(dāng)前 state 制作成 snapshot（快照），并保存到指定的持久化存儲(chǔ)中，最后向 CheckpointCoordinator 異步發(fā)送一個(gè) ack（Acknowledge character --- 確認(rèn)字符），同時(shí)向自身所有下游算子廣播該 Barrier 后恢復(fù)自身的數(shù)據(jù)處理。

每個(gè)算子按照上面不斷制作 snapshot 并向下游廣播，直到最后 Barrier 傳遞到 sink 算子，此時(shí)快照便制作完成。這時(shí)候需要注意的是，上游算子可能是多個(gè)數(shù)據(jù)源，對(duì)應(yīng)多個(gè) Barrier 需要全部到齊才一次性觸發(fā) checkpoint ，所以在遇到 checkpoint 時(shí)間較長(zhǎng)的情況時(shí)，有可能是因?yàn)閿?shù)據(jù)對(duì)齊需要耗費(fèi)的時(shí)間比較長(zhǎng)所造成的。

■ Snapshot & Recover

如圖，這是我們的Container容器初始化的階段，e1 和 e2 是剛從 Kafka 消費(fèi)過(guò)來(lái)的數(shù)據(jù)，與此同時(shí)，CheckpointCoordinator 也往它發(fā)送了 Barrier。

此時(shí) Task1 完成了它的 checkpoint 過(guò)程，效果就是記錄下 offset 為2（e1，e2），然后把 Barrier 往下游的算子廣播，Task3 的輸入為 Task1 的輸出，現(xiàn)在假設(shè)我的這個(gè)程序的功能是統(tǒng)計(jì)數(shù)據(jù)的條數(shù)，此時(shí) Task3 的 checkpoint 效果就是就記錄數(shù)據(jù)數(shù)為2（因?yàn)閺?Task1 過(guò)來(lái)的數(shù)據(jù)就是 e1 和 e2 兩條），之后再將 Barrier 往下廣播，當(dāng)此 Barrier 傳遞到 sink 算子，snapshot 就算是制作完成了。

此時(shí) source 中還會(huì)源源不斷的產(chǎn)生數(shù)據(jù)，并產(chǎn)生新的 checkpoint ，但是此時(shí)如果 Container 宕機(jī)重啟就需要進(jìn)行數(shù)據(jù)的恢復(fù)了。剛剛完成的 checkpoint 中 offset為2，count為2，那我們就按照這個(gè) state 進(jìn)行恢復(fù)。此時(shí) Task1 會(huì)從 e3 開(kāi)始消費(fèi)，這就是 Recover 操作。

■ checkpoint 的注意事項(xiàng)

下面列舉的3個(gè)注意要點(diǎn)都會(huì)影響到系統(tǒng)的吞吐，在實(shí)際開(kāi)發(fā)過(guò)程中需要注意：

3.背壓的產(chǎn)生及 Flink 的反壓處理

在分布式系統(tǒng)中經(jīng)常會(huì)出現(xiàn)多個(gè) Task 多個(gè) JVM 之間可能需要做數(shù)據(jù)的交換，我們使用生產(chǎn)者和消費(fèi)者來(lái)說(shuō)明這個(gè)事情。

假設(shè)我現(xiàn)在的 Producer 是使用了無(wú)界 buffer 來(lái)進(jìn)行存儲(chǔ)，當(dāng)我們的生產(chǎn)者生產(chǎn)速度遠(yuǎn)大于消費(fèi)者消費(fèi)的速度時(shí)，生產(chǎn)端的數(shù)據(jù)會(huì)因?yàn)橄M(fèi)端的消費(fèi)能力低下而導(dǎo)致數(shù)據(jù)積壓，最終導(dǎo)致 OOM 的產(chǎn)生。

而就算使用了有界 buffer，同樣消費(fèi)者端的消費(fèi)能力低下，當(dāng) buffer 被積滿時(shí)生產(chǎn)者就會(huì)停止生產(chǎn)，這樣還不能完全地解決我們的問(wèn)題，所以就需要根據(jù)不同的情況進(jìn)行調(diào)整。

Flink 也是通過(guò)有界 buffer 來(lái)進(jìn)行不同 TaskManager 的數(shù)據(jù)交換。而且做法分為了靜態(tài)控流和動(dòng)態(tài)控流兩種方式。

簡(jiǎn)單來(lái)說(shuō)就是當(dāng)生產(chǎn)者比消費(fèi)者的 TPS 多時(shí)，我們采用溢寫(xiě)的方式，使用 batch 來(lái)封裝好我們的數(shù)據(jù)，然后分批發(fā)送出去，每次發(fā)送完成后再 sleep 一段時(shí)間，這個(gè)時(shí)間的計(jì)算方式是 left（剩余的數(shù)據(jù)）/ tps，但是這個(gè)做法是很難去預(yù)估系統(tǒng)的情況的。

Flink 1.5 之前的流控是基于 TCP 的滑動(dòng)窗口實(shí)現(xiàn)的，在之前的課程中已經(jīng)有提到過(guò)了。而 Flink 在1.5之后已經(jīng)棄用了該機(jī)制，所以這里不展開(kāi)說(shuō)明。在此網(wǎng)絡(luò)模型中，數(shù)據(jù)生成節(jié)點(diǎn)只能通過(guò)檢查當(dāng)前的 channel 是否可寫(xiě)來(lái)決定自己是否要向消費(fèi)端發(fā)送數(shù)據(jù)，它對(duì)下游數(shù)據(jù)消費(fèi)端的真實(shí)容量情況一概不知。這就導(dǎo)致，當(dāng)生成節(jié)點(diǎn)發(fā)現(xiàn) channel 已經(jīng)不可寫(xiě)的時(shí)候，有可能下游消費(fèi)節(jié)點(diǎn)已經(jīng)積壓了很多數(shù)據(jù)。

Credit-Based 我們用下面的數(shù)據(jù)交換的例子說(shuō)明：

Flink 的數(shù)據(jù)交換大致分為三種，一種是同一個(gè) Task 的數(shù)據(jù)交換，另一種是 不同 Task 同 JVM 下的數(shù)據(jù)交換。第三種就是不同 Task 且不同 JVM 之間的交換。

同一個(gè) Task 的數(shù)據(jù)交換就是我們剛剛提到的 forward strategy 方式，主要就是避免了序列化和網(wǎng)絡(luò)的開(kāi)銷。

第二種數(shù)據(jù)交換的方式就是數(shù)據(jù)會(huì)先通過(guò)一個(gè) record Writer ，數(shù)據(jù)在里面進(jìn)行序列化之后再傳遞給 Result Partition ，之后數(shù)據(jù)會(huì)通過(guò) local channel 傳遞給另外一個(gè) Task 的 Input Gate 里面，再進(jìn)行反序列化，推送給 Record Reader 之后進(jìn)行操作。

因?yàn)榈谌N數(shù)據(jù)交換涉及到了不同的 JVM，所以會(huì)有一定的網(wǎng)絡(luò)開(kāi)銷，和第二種的區(qū)別就在于它先推給了 Netty ，通過(guò)netty把數(shù)據(jù)推送到遠(yuǎn)程端的 Task 上。

■ Credit-Based

此時(shí)我們可以看到 event1 已經(jīng)連帶一個(gè) backlog = 1 推送給了 TaskB，backlog 的作用其實(shí)只是為了讓消費(fèi)端感知到我們生產(chǎn)端的情況

此時(shí) event1 被 TaskB 接收后，TaskB會(huì)返回一個(gè) ack 給 TaskA，同時(shí)返回一個(gè)credit = 3，這個(gè)是告知 TaskA 它還能接收多少條數(shù)據(jù)，F(xiàn)link 就是通過(guò)這種互相告知的方式，來(lái)讓生產(chǎn)者和消費(fèi)者都能感知到對(duì)方的狀態(tài)。

此時(shí)經(jīng)過(guò)一段時(shí)間之后，TaskB中的有界 buffer 已經(jīng)滿了，此時(shí) TaskB回復(fù) credit = 0 給 TaskA，此時(shí) channel 通道將會(huì)停止工作，TaskA 不再將數(shù)據(jù)發(fā)往 TaskB。

此時(shí)再經(jīng)過(guò)一段時(shí)間，TaskA 中的有界 Buffer 也已經(jīng)出現(xiàn)了數(shù)據(jù)積壓，所以我們平時(shí)遇到的吞吐下降，處理延遲的問(wèn)題，就是因?yàn)榇藭r(shí)整個(gè)系統(tǒng)相當(dāng)于一個(gè)停滯的狀態(tài)，如圖二示，所有的過(guò)程都被打上 “X”，表示這些過(guò)程都已經(jīng)停止工作。

JVM 是一個(gè)非常復(fù)雜的系統(tǒng)，當(dāng)其內(nèi)存不足時(shí)會(huì)造成 OOM ，導(dǎo)致系統(tǒng)的崩潰。Flink 在拿到我們分配的內(nèi)存之后會(huì)先分配一個(gè) cutoff 預(yù)留內(nèi)存，保證系統(tǒng)的安全性。Netword buffers 其實(shí)就是對(duì)應(yīng)我們剛剛一直提到的有界 buffer，momery manager 是一個(gè)內(nèi)存池，這部分的內(nèi)存可以設(shè)置為堆內(nèi)或者堆外的內(nèi)存，當(dāng)然在流式作業(yè)中我們一般設(shè)置其為堆外內(nèi)存，而 Free 部分就是提供給用戶使用的內(nèi)存塊。

現(xiàn)在我們假設(shè)分配給此 TaskManager 的內(nèi)存是 8g。

1. 首先是要砍掉 cutoff 的部分，默認(rèn)是0.25，所以我們的可用內(nèi)存就是 8gx0.75
2. network buffers 占用可用內(nèi)存的 0.1 ，所以是 6144x0.1
3. 堆內(nèi)/堆外內(nèi)存為可用內(nèi)存減去 network buffers 的部分，再乘以 0.8
4. 給到用戶使用的內(nèi)存就是堆內(nèi)存剩下的 0.2 那部分

其實(shí)真實(shí)情況是 Flink 是先知道了 heap 內(nèi)存的大小然后逆推出其它內(nèi)存的大小。

Flink 作業(yè)的問(wèn)題定位

1.問(wèn)題定位口訣

“一壓二查三指標(biāo)，延遲吞吐是核心。

時(shí)刻關(guān)注資源量 , ?排查首先看GC。”

一壓是指背壓，遇到問(wèn)題先看背壓的情況，二查就是指 checkpoint ，對(duì)齊數(shù)據(jù)的時(shí)間是否很長(zhǎng)，state 是否很大，這些都是和系統(tǒng)吞吐密切相關(guān)的，三指標(biāo)就是指 Flink UI 那塊的一些展示，我們的主要關(guān)注點(diǎn)其實(shí)就是延遲和吞吐，系統(tǒng)資源，還有就是 GC logs。

• 看反壓：通常最后一個(gè)被壓高的 subTask 的下游就是 job 的瓶頸之一。

• 看 Checkpoint 時(shí)長(zhǎng)：Checkpoint 時(shí)長(zhǎng)能在一定程度影響 job 的整體吞吐。

• 看核心指標(biāo)：指標(biāo)是對(duì)一個(gè)任務(wù)性能精準(zhǔn)判斷的依據(jù)，延遲指標(biāo)和吞吐則是其中最為關(guān)鍵的指標(biāo)。

• 資源的使用率：提高資源的利用率是最終的目的。

■ 常見(jiàn)的性能問(wèn)題

簡(jiǎn)單解釋一下：

1. 在關(guān)注背壓的時(shí)候大家往往忽略了數(shù)據(jù)的序列化和反序列化過(guò)程所造成的性能問(wèn)題。
2. 一些數(shù)據(jù)結(jié)構(gòu)，比如 HashMap 和 HashSet 這種 key 需要經(jīng)過(guò) hash 計(jì)算的數(shù)據(jù)結(jié)構(gòu)，在數(shù)據(jù)量大的時(shí)候使用 keyby 進(jìn)行操作， 造成的性能影響是非常大的。
3. 數(shù)據(jù)傾斜是我們的經(jīng)典問(wèn)題，后面再進(jìn)行展開(kāi)。
4. 如果我們的下游是 MySQL，HBase 這種，我們都會(huì)進(jìn)行一個(gè)批處理的操作，就是讓數(shù)據(jù)存儲(chǔ)到一個(gè) buffer 里面，在達(dá)到某些條件的時(shí)候再進(jìn)行發(fā)送，這樣做的目的就是減少和外部系統(tǒng)的交互，降低網(wǎng)絡(luò)開(kāi)銷的成本。
5. 頻繁 GC ，無(wú)論是 CMS 也好，G1 也好，在進(jìn)行 GC 的時(shí)候，都會(huì)停止整個(gè)作業(yè)的運(yùn)行，GC 時(shí)間較長(zhǎng)還會(huì)導(dǎo)致 JobManager 和 TaskManager 沒(méi)有辦法準(zhǔn)時(shí)發(fā)送心跳，此時(shí) JobManager 就會(huì)認(rèn)為此 TaskManager 失聯(lián)，它就會(huì)另外開(kāi)啟一個(gè)新的 TaskManager
6. 窗口是一種可以把無(wú)限數(shù)據(jù)切割為有限數(shù)據(jù)塊的手段。比如我們知道，使用滑動(dòng)窗口的時(shí)候數(shù)據(jù)的重疊問(wèn)題，size = 5min 雖然不屬于大窗口的范疇，可是 step = 1s 代表1秒就要進(jìn)行一次數(shù)據(jù)的處理，這樣就會(huì)造成數(shù)據(jù)的重疊很高，數(shù)據(jù)量很大的問(wèn)題。

2.Flink 作業(yè)調(diào)優(yōu)

我們可以通過(guò)一些數(shù)據(jù)結(jié)構(gòu)，比如 Set 或者 Map 來(lái)結(jié)合 Flink state 進(jìn)行去重。但是這些去重方案會(huì)隨著數(shù)據(jù)量不斷增大，從而導(dǎo)致性能的急劇下降，比如剛剛我們分析過(guò)的 hash 沖突帶來(lái)的寫(xiě)入性能問(wèn)題，內(nèi)存過(guò)大導(dǎo)致的 GC 問(wèn)題，TaskManger 的失聯(lián)問(wèn)題。

方案二和方案三也都是通過(guò)一些數(shù)據(jù)結(jié)構(gòu)的手段去進(jìn)行去重，有興趣的同學(xué)可以自行下去了解，在這里不再展開(kāi)。

■ 數(shù)據(jù)傾斜

數(shù)據(jù)傾斜是大家都會(huì)遇到的高頻問(wèn)題，解決的方案也不少。

第一種場(chǎng)景是當(dāng)我們的并發(fā)度設(shè)置的比分區(qū)數(shù)要低時(shí)，就會(huì)造成上面所說(shuō)的消費(fèi)不均勻的情況。

第二種提到的就是 key 分布不均勻的情況，可以通過(guò)添加隨機(jī)前綴打散它們的分布，使得數(shù)據(jù)不會(huì)集中在幾個(gè) Task 中。

在每個(gè)節(jié)點(diǎn)本地對(duì)相同的 key 進(jìn)行一次聚合操作，類似于 MapReduce 中的本地 combiner。map-side 預(yù)聚合之后，每個(gè)節(jié)點(diǎn)本地就只會(huì)有一條相同的 key，因?yàn)槎鄺l相同的 key 都被聚合起來(lái)了。其他節(jié)點(diǎn)在拉取所有節(jié)點(diǎn)上的相同 key 時(shí)，就會(huì)大大減少需要拉取的數(shù)據(jù)數(shù)量，從而也就減少了磁盤(pán) IO 以及網(wǎng)絡(luò)傳輸開(kāi)銷。

■ 內(nèi)存調(diào)優(yōu)

Flink 的內(nèi)存結(jié)構(gòu)剛剛我們已經(jīng)提及到了，所以我們清楚，調(diào)優(yōu)的方面主要是針對(duì) 非堆內(nèi)存 Network buffer ，manager pool 和堆內(nèi)存的調(diào)優(yōu)，這些基本都是通過(guò)參數(shù)來(lái)進(jìn)行控制的。

這些參數(shù)我們都需要結(jié)合自身的情況去進(jìn)行調(diào)整，這里只給出一些建議。而且對(duì)于 ManagerBuffer 來(lái)說(shuō)，F(xiàn)link 的流式作業(yè)現(xiàn)在并沒(méi)有過(guò)多使用到這部分的內(nèi)存，所以我們都會(huì)設(shè)置得比較小，不超過(guò)0.3。

堆內(nèi)存的調(diào)優(yōu)是關(guān)于 JVM 方面的，主要就是將默認(rèn)使用的垃圾回收器改為 G1 ，因?yàn)槟J(rèn)使用的 Parallel Scavenge 對(duì)于老年代的 GC 存在一個(gè)串行化的問(wèn)題，它的 Full GC 耗時(shí)較長(zhǎng)，下面是關(guān)于 G1 的一些介紹，網(wǎng)上資料也非常多，這里就不展開(kāi)說(shuō)明了。

總 結(jié)

本文帶大家了解了 Flink 的 CheckPoint 機(jī)制，反壓機(jī)制及 Flink 的內(nèi)存模型和基于內(nèi)存模型分析了一些調(diào)優(yōu)的策略。希望能對(duì)大家有所幫助，原文分享的視頻回顧可移步下方鏈接：

https://ververica.cn/developers/flink-training-course-operation/