Flink在大規模狀態資料集下的checkpoint調優
文章目錄
前言
眾所周知，Flink內部為了實現它的高可用性，實現了一套強大的checkpoint機制，還能保證作用的Exactly Once的快速恢復。對此，圍繞checkpoint過程本身做了很多的工作。在官方文件中，也為使用者解釋了checkpoint的部分原理以及checkpoint在實際生產中（尤其是大規模狀態集下）的checkpoint調優引數。筆者結合官方文件，給大家做個總結，也算是對Flink checkpoint機理的一個學習。
Checkpoint快慢的效能指標
如果說我們想要對flink的checkpoint操作做調優，那麼我們首先得有個衡量指標來展現當前checkpoint是否快慢。在這裡，官方提供了以下2個metric指標：
- Checkpoint每次開始的時間。觀察每次checkpoint開始的時間是為了檢測在每次前後checkpoint中間是否存在空閒時間間隔。如果存在間隔時間，說明當前checkpoint都在合理時間內完成。
- 觀察資料buffered的量。這個buffered動作是為了等待其它較慢資料流的stream barriers而設計的。這個偏向於checkpoint原理化的相關內容了。
但大體上，使用者根據第一條就能夠監測出應用的checkout快慢了。
相鄰Checkpoint的間隔時間設定
我們假設一個使用場景，在極大規模狀態資料集下，應用每次的checkpoint時長都超過系統設定的最大時間（也就是checkpoint間隔時長），那麼會發生什麼樣的事情。
答案是應用會一直在做checkpoint，因為當應用發現它剛剛做完一次checkpoint後，又已經到了下次checkpoint的時間了，然後又開始新的checkpoint。最後就會造成一個很壞的結果：使用者應用本身都沒法跑了。。。
當然了，我們可能會說了，我們設定一下並行checkpoint數，或者說做增量checkpoint，不用每次做全量checkpoint。每次只checkpoint出對前一次checkpoint內的狀態資料的增量改動。然後恢復的時候做狀態改動的重放
但是這裡，我們可以採用一種更加直接有效的方法，設定連續checkpoint的時間間隔。形象地解釋，就是強行在checkpoint間塞入空閒時間，如下圖。
涉及的相關配置設定如下：
StreamExecutionEnvironment.getCheckpointConfig().setMinPauseBetweenCheckpoints(milliseconds)
外部State的儲存選擇
上小節的方法其實還並沒有從本質上解決大規模狀態集下checkpoint慢的問題，只是說它降低了這個慢的風險和造成的影響。在這裡我們反覆強調的是一個大規模狀態，我們理理思路，因為規模之大，所以我們才會慢。那如果我們能找到一種更快的儲存狀態的介質（或者策略），那麼這個過程也是能夠變快的。
所以在這裡，我們可以選擇更加高效的外部儲存介質來做State的儲存（比如RocksDB），而不是僅限於儲存於有限的記憶體空間裡，或完全落地到磁碟上。這是我們在State Backend上做的一個選擇。
Checkpoint的資源設定
當我們對越多的狀態資料集做checkpoint時，需要消耗越多的資源。因為Flink在checkpoint時是首先在每個task上做資料checkpoint，然後在外部儲存中做checkpoint持久化。在這裡的一個優化思路是：在總狀態資料固定的情況下，當每個task平均所checkpoint的資料越少，那麼相應地checkpoint的總時間也會變短。所以我們可以為每個task設定更多的並行度（即分配更多的資源）來加速checkpoint的執行過程。
Checkpoint的task本地性恢復
為了大家未來對checkpoint的優化，我們有必要在runtime級別的checkpoint過程。首先我們要明白一點，flink的checkpoint不是一個完全在master節點的過程，而是分散在每個task上執行，然後在做彙總持久化。這些task做的checkpoint資料在後面應用恢復時包括並行度擴增或減少時還能夠重新打散分佈。
為了快速的狀態恢復，每個task會同時寫checkpoint資料到本地磁碟和遠端分散式儲存，也就是說，這是一份雙拷貝。只要task本地的checkpoint資料沒有被破壞，系統在應用恢復時會首先載入本地的checkpoint資料，這樣就大大減少了遠端拉取狀態資料的過程。此過程如下圖所示：
引用
[1].https://ci.apache.org/projects/flink/flink-docs-master/ops/state/large_state_tuning.html