Flink Checkpoint 问题排查实用指南
在 Flink 中，狀態(tài)可靠性保證由 Checkpoint 支持，當(dāng)作業(yè)出現(xiàn) failover 的情況下，Flink 會(huì)從最近成功的 Checkpoint 恢復(fù)。在實(shí)際情況中，我們可能會(huì)遇到 Checkpoint 失敗，或者 Checkpoint 慢的情況，本文會(huì)統(tǒng)一聊一聊 Flink 中 Checkpoint 異常的情況（包括失敗和慢），以及可能的原因和排查思路。
1. Checkpoint 流程簡(jiǎn)介
首先我們需要了解 Flink 中 Checkpoint 的整個(gè)流程是怎樣的，在了解整個(gè)流程之后，我們才能在出問題的時(shí)候，更好的進(jìn)行定位分析。
從上圖我們可以知道，Flink 的 Checkpoint 包括如下幾個(gè)部分：
- JM trigger checkpoint
- Source 收到 trigger checkpoint 的 PRC，自己開始做 snapshot，并往下游發(fā)送 barrier
- 下游接收 barrier（需要 barrier 都到齊才會(huì)開始做 checkpoint）
- Task 開始同步階段 snapshot
- Task 開始異步階段 snapshot
- Task snapshot 完成，匯報(bào)給 JM
上面的任何一個(gè)步驟不成功，整個(gè) checkpoint 都會(huì)失敗。
2 Checkpoint 異常情況排查
2.1 Checkpoint 失敗
可以在 Checkpoint 界面看到如下圖所示，下圖中 Checkpoint 10423 失敗了。
點(diǎn)擊 Checkpoint 10423 的詳情，我們可以看到類系下圖所示的表格（下圖中將 operator 名字截取掉了）。
上圖中我們看到三行，表示三個(gè) operator，其中每一列的含義分別如下：
- 其中?Acknowledged?一列表示有多少個(gè) subtask 對(duì)這個(gè) Checkpoint 進(jìn)行了 ack，從圖中我們可以知道第三個(gè) operator 總共有 5 個(gè) subtask，但是只有 4 個(gè)進(jìn)行了 ack；
- 第二列?Latest Acknowledgement?表示該 operator 的所有 subtask 最后 ack 的時(shí)間；
- End to End Duration?表示整個(gè) operator 的所有 subtask 中完成 snapshot 的最長(zhǎng)時(shí)間；
- State Size?表示當(dāng)前 Checkpoint 的 state 大小 -- 主要這里如果是增量 checkpoint 的話，則表示增量大小；
- Buffered During Alignment?表示在 barrier 對(duì)齊階段積攢了多少數(shù)據(jù)，如果這個(gè)數(shù)據(jù)過(guò)大也間接表示對(duì)齊比較慢）；
Checkpoint 失敗大致分為兩種情況：Checkpoint Decline 和 Checkpoint Expire。
2.1.1 Checkpoint Decline
我們能從?jobmanager.log?中看到類似下面的日志
Decline checkpoint 10423 by task 0b60f08bf8984085b59f8d9bc74ce2e1 of job 85d268e6fbc19411185f7e4868a44178.?其中
10423 是 checkpointID，0b60f08bf8984085b59f8d9bc74ce2e1?是 execution id，85d268e6fbc19411185f7e4868a44178?是 job id，我們可以在?jobmanager.log?中查找 execution id，找到被調(diào)度到哪個(gè) taskmanager 上，類似如下所示：
從上面的日志我們知道該 execution 被調(diào)度到?hostnameABCDE?的?container_e24_1566836790522_8088_04_013155_1?slot 上，接下來(lái)我們就可以到 container?container_e24_1566836790522_8088_04_013155?的 taskmanager.log 中查找 Checkpoint 失敗的具體原因了。
另外對(duì)于 Checkpoint Decline 的情況，有一種情況我們?cè)谶@里單獨(dú)抽取出來(lái)進(jìn)行介紹：Checkpoint Cancel。
當(dāng)前 Flink 中如果較小的 Checkpoint 還沒有對(duì)齊的情況下，收到了更大的 Checkpoint，則會(huì)把較小的 Checkpoint 給取消掉。我們可以看到類似下面的日志：
$taskNameWithSubTaskAndID: Received checkpoint barrier for checkpoint 20 before completing current checkpoint 19. Skipping current checkpoint.這個(gè)日志表示，當(dāng)前 Checkpoint 19 還在對(duì)齊階段，我們收到了 Checkpoint 20 的 barrier。然后會(huì)逐級(jí)通知到下游的 task checkpoint 19 被取消了，同時(shí)也會(huì)通知 JM 當(dāng)前 Checkpoint 被 decline 掉了。
在下游 task 收到被 cancelBarrier 的時(shí)候，會(huì)打印類似如下的日志：
DEBUG $taskNameWithSubTaskAndID: Checkpoint 19 canceled, aborting alignment.或者DEBUG $taskNameWithSubTaskAndID: Checkpoint 19 canceled, skipping alignment.或者WARN $taskNameWithSubTaskAndID: Received cancellation barrier for checkpoint 20 before completing current checkpoint 19. Skipping current checkpoint.上面三種日志都表示當(dāng)前 task 接收到上游發(fā)送過(guò)來(lái)的 barrierCancel 消息，從而取消了對(duì)應(yīng)的 Checkpoint。
2.1.2 Checkpoint Expire
如果 Checkpoint 做的非常慢，超過(guò)了 timeout 還沒有完成，則整個(gè) Checkpoint 也會(huì)失敗。當(dāng)一個(gè) Checkpoint 由于超時(shí)而失敗是，會(huì)在?jobmanager.log?中看到如下的日志：
Checkpoint 1 of job 85d268e6fbc19411185f7e4868a44178 expired before completing.表示 Chekpoint 1 由于超時(shí)而失敗，這個(gè)時(shí)候可以可以看這個(gè)日志后面是否有類似下面的日志：
Received late message for now expired checkpoint attempt 1 from 0b60f08bf8984085b59f8d9bc74ce2e1 of job 85d268e6fbc19411185f7e4868a44178.可以按照 2.1.1 中的方法找到對(duì)應(yīng)的 taskmanager.log 查看具體信息。
下面的日志如果是 DEBUG 的話，我們會(huì)在開始處標(biāo)記?DEBUG
我們按照下面的日志把 TM 端的 snapshot 分為三個(gè)階段，開始做 snapshot 前，同步階段，異步階段：
DEBUG Starting checkpoint (6751) CHECKPOINT on task taskNameWithSubtasks (4/4)這個(gè)日志表示 TM 端 barrier 對(duì)齊后，準(zhǔn)備開始做 Checkpoint。
DEBUG 2019-08-06 13:43:02,613 DEBUG org.apache.flink.runtime.state.AbstractSnapshotStrategy - DefaultOperatorStateBackend snapshot (FsCheckpointStorageLocation {fileSystem=org.apache.flink.core.fs.SafetyNetWrapperFileSystem@70442baf, checkpointDirectory=xxxxxxxx, sharedStateDirectory=xxxxxxxx, taskOwnedStateDirectory=xxxxxx, metadataFilePath=xxxxxx, reference=(default), fileStateSizeThreshold=1024}, synchronous part) in thread Thread[Async calls on Source: xxxxxx _source -> Filter (27/70),5,Flink Task Threads] took 0 ms.上面的日志表示當(dāng)前這個(gè) backend 的同步階段完成，共使用了 0 ms。
DEBUG DefaultOperatorStateBackend snapshot (FsCheckpointStorageLocation {fileSystem=org.apache.flink.core.fs.SafetyNetWrapperFileSystem@7908affe, checkpointDirectory=xxxxxx, sharedStateDirectory=xxxxx, taskOwnedStateDirectory=xxxxx, metadataFilePath=xxxxxx, reference=(default), fileStateSizeThreshold=1024}, asynchronous part) in thread Thread[pool-48-thread-14,5,Flink Task Threads] took 369 ms上面的日志表示異步階段完成，異步階段使用了 369 ms
在現(xiàn)有的日志情況下，我們通過(guò)上面三個(gè)日志，定位 snapshot 是開始晚，同步階段做的慢，還是異步階段做的慢。然后再按照情況繼續(xù)進(jìn)一步排查問題。
2.2 Checkpoint 慢
在 2.1 節(jié)中，我們介紹了 Checkpoint 失敗的排查思路，本節(jié)會(huì)分情況介紹 Checkpoint 慢的情況。
Checkpoint 慢的情況如下：比如 Checkpoint interval 1 分鐘，超時(shí) 10 分鐘，Checkpoint 經(jīng)常需要做 9 分鐘（我們希望 1 分鐘左右就能夠做完），而且我們預(yù)期 state size 不是非常大。
對(duì)于 Checkpoint 慢的情況，我們可以按照下面的順序逐一檢查。
2.2.0 Source Trigger Checkpoint 慢
這個(gè)一般發(fā)生較少，但是也有可能，因?yàn)?source 做 snapshot 并往下游發(fā)送 barrier 的時(shí)候，需要搶鎖（這個(gè)現(xiàn)在社區(qū)正在進(jìn)行用 mailBox 的方式替代當(dāng)前搶鎖的方式，詳情參考[1])。如果一直搶不到鎖的話，則可能導(dǎo)致 Checkpoint 一直得不到機(jī)會(huì)進(jìn)行。如果在 Source 所在的?taskmanager.log?中找不到開始做 Checkpoint 的 log，則可以考慮是否屬于這種情況，可以通過(guò)?jstack?進(jìn)行進(jìn)一步確認(rèn)鎖的持有情況。
2.2.1 使用增量 Checkpoint
現(xiàn)在 Flink 中 Checkpoint 有兩種模式，全量 Checkpoint 和 增量 Checkpoint，其中全量 Checkpoint 會(huì)把當(dāng)前的 state 全部備份一次到持久化存儲(chǔ)，而增量 Checkpoint，則只備份上一次 Checkpoint 中不存在的 state，因此增量 Checkpoint 每次上傳的內(nèi)容會(huì)相對(duì)更好，在速度上會(huì)有更大的優(yōu)勢(shì)。
現(xiàn)在 Flink 中僅在 RocksDBStateBackend 中支持增量 Checkpoint，如果你已經(jīng)使用 RocksDBStateBackend，可以通過(guò)開啟增量 Checkpoint 來(lái)加速，具體的可以參考 [2]。
2.2.2 作業(yè)存在反壓或者數(shù)據(jù)傾斜
我們知道 task 僅在接受到所有的 barrier 之后才會(huì)進(jìn)行 snapshot，如果作業(yè)存在反壓，或者有數(shù)據(jù)傾斜，則會(huì)導(dǎo)致全部的 channel 或者某些 channel 的 barrier 發(fā)送慢，從而整體影響 Checkpoint 的時(shí)間，這兩個(gè)可以通過(guò)如下的頁(yè)面進(jìn)行檢查：
上圖中我們選擇了一個(gè) task，查看所有 subtask 的反壓情況，發(fā)現(xiàn)都是 high，表示反壓情況嚴(yán)重，這種情況下會(huì)導(dǎo)致下游接收 barrier 比較晚。
上圖中我們選擇其中一個(gè) operator，點(diǎn)擊所有的 subtask，然后按照 Records Received/Bytes Received/TPS 從大到小進(jìn)行排序，能看到前面幾個(gè) subtask 會(huì)比其他的 subtask 要處理的數(shù)據(jù)多。
如果存在反壓或者數(shù)據(jù)傾斜的情況，我們需要首先解決反壓或者數(shù)據(jù)傾斜問題之后，再查看 Checkpoint 的時(shí)間是否符合預(yù)期。
2.2.2 Barrier 對(duì)齊慢
從前面我們知道 Checkpoint 在 task 端分為 barrier 對(duì)齊（收齊所有上游發(fā)送過(guò)來(lái)的 barrier），然后開始同步階段，再做異步階段。如果 barrier 一直對(duì)不齊的話，就不會(huì)開始做 snapshot。
barrier 對(duì)齊之后會(huì)有如下日志打印：
DEBUG Starting checkpoint (6751) CHECKPOINT on task taskNameWithSubtasks (4/4)如果?taskmanager.log?中沒有這個(gè)日志，則表示 barrier 一直沒有對(duì)齊，接下來(lái)我們需要了解哪些上游的 barrier 沒有發(fā)送下來(lái)，如果你使用 At Least Once 的話，可以觀察下面的日志：
DEBUG Received barrier for checkpoint 96508 from channel 5表示該 task 收到了 channel 5 來(lái)的 barrier，然后看對(duì)應(yīng) Checkpoint，再查看還剩哪些上游的 barrier 沒有接受到，對(duì)于 ExactlyOnce 暫時(shí)沒有類似的日志，可以考慮自己添加，或者 jmap 查看。
2.2.3 主線程太忙，導(dǎo)致沒機(jī)會(huì)做 snapshot
在 task 端，所有的處理都是單線程的，數(shù)據(jù)處理和 barrier 處理都由主線程處理，如果主線程在處理太慢（比如使用 RocksDBBackend，state 操作慢導(dǎo)致整體處理慢），導(dǎo)致 barrier 處理的慢，也會(huì)影響整體 Checkpoint 的進(jìn)度，在這一步我們需要能夠查看某個(gè) PID 對(duì)應(yīng) hotmethod，這里推薦兩個(gè)方法：
如果有其他更方便的方法當(dāng)然更好，也歡迎推薦。
2.2.4 同步階段做的慢
同步階段一般不會(huì)太慢，但是如果我們通過(guò)日志發(fā)現(xiàn)同步階段比較慢的話，對(duì)于非 RocksDBBackend 我們可以考慮查看是否開啟了異步 snapshot，如果開啟了異步 snapshot 還是慢，需要看整個(gè) JVM 在干嘛，也可以使用前一節(jié)中的工具。對(duì)于 RocksDBBackend 來(lái)說(shuō)，我們可以用?iostate?查看磁盤的壓力如何，另外可以查看 tm 端 RocksDB 的 log 的日志如何，查看其中 SNAPSHOT 的時(shí)間總共開銷多少。
RocksDB 開始 snapshot 的日志如下：
2019/09/10-14:22:55.734684 7fef66ffd700 [utilities/checkpoint/checkpoint_impl.cc:83] Started the snapshot process -- creating snapshot in directory /tmp/flink-io-87c360ce-0b98-48f4-9629-2cf0528d5d53/XXXXXXXXXXX/chk-92729snapshot 結(jié)束的日志如下：
2019/09/10-14:22:56.001275 7fef66ffd700 [utilities/checkpoint/checkpoint_impl.cc:145] Snapshot DONE. All is good2.2.6 異步階段做的慢
對(duì)于異步階段來(lái)說(shuō)，tm 端主要將 state 備份到持久化存儲(chǔ)上，對(duì)于非 RocksDBBackend 來(lái)說(shuō)，主要瓶頸來(lái)自于網(wǎng)絡(luò)，這個(gè)階段可以考慮觀察網(wǎng)絡(luò)的 metric，或者對(duì)應(yīng)機(jī)器上能夠觀察到網(wǎng)絡(luò)流量的情況（比如?iftop)。
對(duì)于 RocksDB 來(lái)說(shuō)，則需要從本地讀取文件，寫入到遠(yuǎn)程的持久化存儲(chǔ)上，所以不僅需要考慮網(wǎng)絡(luò)的瓶頸，還需要考慮本地磁盤的性能。另外對(duì)于 RocksDBBackend 來(lái)說(shuō)，如果覺得網(wǎng)絡(luò)流量不是瓶頸，但是上傳比較慢的話，還可以嘗試考慮開啟多線程上傳功能[3]。
3 總結(jié)
在第二部分內(nèi)容中，我們介紹了官方編譯的包的情況下排查一些 Checkpoint 異常情況的主要場(chǎng)景，以及相應(yīng)的排查方法，如果排查了上面所有的情況，還是沒有發(fā)現(xiàn)瓶頸所在，則可以考慮添加更詳細(xì)的日志，逐步將范圍縮小，然后最終定位原因。
上文提到的一些 DEBUG 日志，如果 flink dist 包是自己編譯的話，則建議將 Checkpoint 整個(gè)步驟內(nèi)的一些 DEBUG 改為 INFO，能夠通過(guò)日志了解整個(gè) Checkpoint 的整體階段，什么時(shí)候完成了什么階段，也在 Checkpoint 異常的時(shí)候，快速知道每個(gè)階段都消耗了多少時(shí)間。
原文鏈接
本文為云棲社區(qū)原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。
總結(jié)
以上是生活随笔為你收集整理的Flink Checkpoint 问题排查实用指南的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 阿里HBase高可用8年抗战回忆录
- 下一篇： 核心系统100%上云！全球最大流量洪峰，