Flink可靠性的基石-checkpoint機制詳細解析

Checkpoint介紹

checkpoint機制是Flink可靠性的基石，可以保證Flink集群在某個算子因為某些原因(如 異常退出)出現故障時，能夠將整個應用流圖的狀態恢復到故障之前的某一狀態，保 證應用流圖狀態的一致性。Flink的checkpoint機制原理來自“Chandy-Lamport algorithm”算法。

每個需要checkpoint的應用在啟動時，Flink的JobManager為其創建一個 CheckpointCoordinator(檢查點協調器)，CheckpointCoordinator全權負責本應用的快照制作。

1. CheckpointCoordinator(檢查點協調器) 周期性的向該流應用的所有source算子發送 barrier(屏障)。

2. 當某個source算子收到一個barrier時，便暫停數據處理過程，然后將自己的當前狀態制作成快照，並保存到指定的持久化存儲中，最后向CheckpointCoordinator報告自己快照制作情況，同時向自身所有下游算子廣播該barrier，恢復數據處理

3. 下游算子收到barrier之后，會暫停自己的數據處理過程，然后將自身的相關狀態制作成快照，並保存到指定的持久化存儲中，最后向CheckpointCoordinator報告自身快照情況，同時向自身所有下游算子廣播該barrier，恢復數據處理。

4. 每個算子按照步驟3不斷制作快照並向下游廣播，直到最后barrier傳遞到sink算子，快照制作完成。

5. 當CheckpointCoordinator收到所有算子的報告之后，認為該周期的快照制作成功; 否則，如果在規定的時間內沒有收到所有算子的報告，則認為本周期快照制作失敗。

如果一個算子有兩個輸入源，則暫時阻塞先收到barrier的輸入源，等到第二個輸入源相 同編號的barrier到來時，再制作自身快照並向下游廣播該barrier。具體如下圖所示：

1. 假設算子C有A和B兩個輸入源

2. 在第i個快照周期中，由於某些原因(如處理時延、網絡時延等)輸入源A發出的 barrier 先到來，這時算子C暫時將輸入源A的輸入通道阻塞，僅收輸入源B的數據。

3. 當輸入源B發出的barrier到來時，算子C制作自身快照並向 CheckpointCoordinator 報告自身的快照制作情況，然后將兩個barrier合並為一個，向下游所有的算子廣播。

4. 當由於某些原因出現故障時，CheckpointCoordinator通知流圖上所有算子統一恢復到某個周期的checkpoint狀態，然后恢復數據流處理。分布式checkpoint機制保證了數據僅被處理一次(Exactly Once)。

持久化存儲

MemStateBackend

該持久化存儲主要將快照數據保存到JobManager的內存中，僅適合作為測試以及快照的數據量非常小時使用，並不推薦用作大規模商業部署。

MemoryStateBackend 的局限性：

默認情況下，每個狀態的大小限制為 5 MB。可以在MemoryStateBackend的構造函數中增加此值。

無論配置的最大狀態大小如何，狀態都不能大於akka幀的大小（請參閱配置）。

聚合狀態必須適合 JobManager 內存。

建議MemoryStateBackend 用於：

本地開發和調試。

狀態很少的作業，例如僅包含一次記錄功能的作業（Map，FlatMap，Filter，...），kafka的消費者需要很少的狀態。

FsStateBackend

該持久化存儲主要將快照數據保存到文件系統中，目前支持的文件系統主要是 HDFS和本地文件。如果使用HDFS，則初始化FsStateBackend時，需要傳入以 “hdfs://”開頭的路徑(即: new FsStateBackend("hdfs:///hacluster/checkpoint"))， 如果使用本地文件，則需要傳入以“file://”開頭的路徑(即:new FsStateBackend("file:///Data"))。在分布式情況下，不推薦使用本地文件。如果某 個算子在節點A上失敗，在節點B上恢復，使用本地文件時，在B上無法讀取節點 A上的數據，導致狀態恢復失敗。

建議FsStateBackend：

具有大狀態，長窗口，大鍵 / 值狀態的作業。

所有高可用性設置。

RocksDBStateBackend

RocksDBStatBackend介於本地文件和HDFS之間，平時使用RocksDB的功能，將數 據持久化到本地文件中，當制作快照時，將本地數據制作成快照，並持久化到 FsStateBackend中(FsStateBackend不必用戶特別指明，只需在初始化時傳入HDFS 或本地路徑即可，如new RocksDBStateBackend("hdfs:///hacluster/checkpoint")或new RocksDBStateBackend("file:///Data"))。

如果用戶使用自定義窗口(window)，不推薦用戶使用RocksDBStateBackend。在自定義窗口中，狀態以ListState的形式保存在StatBackend中，如果一個key值中有多個value值，則RocksDB讀取該種ListState非常緩慢，影響性能。用戶可以根據應用的具體情況選擇FsStateBackend+HDFS或RocksStateBackend+HDFS。

語法

val env = StreamExecutionEnvironment.getExecutionEnvironment()
// start a checkpoint every 1000 ms
env.enableCheckpointing(1000)
// advanced options:
// 設置checkpoint的執行模式，最多執行一次或者至少執行一次
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// 設置checkpoint的超時時間
env.getCheckpointConfig.setCheckpointTimeout(60000)
// 如果在只做快照過程中出現錯誤，是否讓整體任務失敗：true是  false不是
env.getCheckpointConfig.setFailTasksOnCheckpointingErrors(false)
//設置同一時間有多少 個checkpoint可以同時執行 
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

修改State Backend的兩種方式

第一種：單任務調整

修改當前任務代碼

env.setStateBackend(new FsStateBackend("hdfs://namenode:9000/flink/checkpoints"));
或者new MemoryStateBackend()
或者new RocksDBStateBackend(filebackend, true);【需要添加第三方依賴】

第二種：全局調整

修改flink-conf.yaml

state.backend: filesystem
state.checkpoints.dir: hdfs://namenode:9000/flink/checkpoints

注意：state.backend的值可以是下面幾種：jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)

Checkpoint的高級選項

默認checkpoint功能是disabled的，想要使用的時候需要先啟用checkpoint開啟之后，默認的checkPointMode是Exactly-once

//配置一秒鍾開啟一個checkpoint
env.enableCheckpointing(1000)
//指定checkpoint的執行模式
//兩種可選：
//CheckpointingMode.EXACTLY_ONCE：默認值
//CheckpointingMode.AT_LEAST_ONCE

env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

一般情況下選擇CheckpointingMode.EXACTLY_ONCE，除非場景要求極低的延遲（幾毫秒）

注意：如果需要保證EXACTLY_ONCE，source和sink要求必須同時保證EXACTLY_ONCE

//如果程序被cancle，保留以前做的checkpoint
env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

默認情況下，檢查點不被保留，僅用於在故障中恢復作業，可以啟用外部持久化檢查點，同時指定保留策略:

ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:在作業取消時保留檢查點，注意，在這種情況下，您必須在取消后手動清理檢查點狀態

ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION：當作業在被cancel時，刪除檢查點，檢查點僅在作業失敗時可用

//設置checkpoint超時時間
env.getCheckpointConfig.setCheckpointTimeout(60000)
//Checkpointing的超時時間，超時時間內沒有完成則被終止

//Checkpointing最小時間間隔，用於指定上一個checkpoint完成之后
//最小等多久可以觸發另一個checkpoint，當指定這個參數時，maxConcurrentCheckpoints的值為1
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

//設置同一個時間是否可以有多個checkpoint執行
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
指定運行中的checkpoint最多可以有多少個

env.getCheckpointConfig.setFailOnCheckpointingErrors(true)
用於指定在checkpoint發生異常的時候，是否應該fail該task，默認是true，如果設置為false，則task會拒絕checkpoint然后繼續運行

Flink的重啟策略

Flink支持不同的重啟策略，這些重啟策略控制着job失敗后如何重啟。集群可以通過默認的重啟策略來重啟，這個默認的重啟策略通常在未指定重啟策略的情況下使用，而如果Job提交的時候指定了重啟策略，這個重啟策略就會覆蓋掉集群的默認重啟策略。

概覽

默認的重啟策略是通過Flink的 flink-conf.yaml 來指定的，這個配置參數 restart-strategy 定義了哪種策略會被采用。如果checkpoint未啟動，就會采用 no restart 策略，如果啟動了checkpoint機制，但是未指定重啟策略的話，就會采用 fixed-delay 策略，重試 Integer.MAX_VALUE 次。請參考下面的可用重啟策略來了解哪些值是支持的。

每個重啟策略都有自己的參數來控制它的行為，這些值也可以在配置文件中設置，每個重啟策略的描述都包含着各自的配置值信息。

重啟策略	重啟策略值
Fixed delay	fixed-delay
Failure rate	failure-rate
No restart	None

除了定義一個默認的重啟策略之外，你還可以為每一個Job指定它自己的重啟策略，這個重啟策略可以在 ExecutionEnvironment 中調用 setRestartStrategy() 方法來程序化地調用，注意這種方式同樣適用於 StreamExecutionEnvironment。

下面的例子展示了如何為Job設置一個固定延遲重啟策略，一旦有失敗，系統就會嘗試每10秒重啟一次，重啟3次。

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
  3, // 重啟次數
  Time.of(10, TimeUnit.SECONDS) // 延遲時間間隔
))

固定延遲重啟策略(Fixed Delay Restart Strategy)

固定延遲重啟策略會嘗試一個給定的次數來重啟Job，如果超過了最大的重啟次數，Job最終將失敗。在連續的兩次重啟嘗試之間，重啟策略會等待一個固定的時間。

重啟策略可以配置flink-conf.yaml的下面配置參數來啟用，作為默認的重啟策略:

restart-strategy: fixed-delay

配置參數	描述	默認值
restart-strategy.fixed-delay.attempts	在Job最終宣告失敗之前，Flink嘗試執行的次數	1，如果啟用checkpoint的話是Integer.MAX_VALUE
restart-strategy.fixed-delay.delay	延遲重啟意味着一個執行失敗之后，並不會立即重啟，而是要等待一段時間。	akka.ask.timeout,如果啟用checkpoint的話是1s

例子:

restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s

固定延遲重啟也可以在程序中設置:

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
  3, // 重啟次數
  Time.of(10, TimeUnit.SECONDS) // 重啟時間間隔
))

失敗率重啟策略

失敗率重啟策略在Job失敗后會重啟，但是超過失敗率后，Job會最終被認定失敗。在兩個連續的重啟嘗試之間，重啟策略會等待一個固定的時間。

失敗率重啟策略可以在flink-conf.yaml中設置下面的配置參數來啟用:

restart-strategy:failure-rate

配置參數	描述	默認值
restart-strategy.failure-rate.max-failures-per-interval	在一個Job認定為失敗之前，最大的重啟次數	1
restart-strategy.failure-rate.failure-rate-interval	計算失敗率的時間間隔	1分鍾
restart-strategy.failure-rate.delay	兩次連續重啟嘗試之間的時間間隔	akka.ask.timeout

例子:

restart-strategy.failure-rate.max-failures-per-interval: 3
restart-strategy.failure-rate.failure-rate-interval: 5 min
restart-strategy.failure-rate.delay: 10 s

失敗率重啟策略也可以在程序中設置:

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.failureRateRestart(
  3, // 每個測量時間間隔最大失敗次數
  Time.of(5, TimeUnit.MINUTES), //失敗率測量的時間間隔
  Time.of(10, TimeUnit.SECONDS) // 兩次連續重啟嘗試的時間間隔
))

無重啟策略

Job直接失敗，不會嘗試進行重啟

restart-strategy: none

無重啟策略也可以在程序中設置

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.noRestart())

搜索公眾號：五分鍾學大數據，發送 秘籍，即可獲取大數據學習秘籍大禮包，深入鑽研大數據技術！