Flink Checkpoint 參數詳解

什么是 checkpoint

保存狀態

Checkpoint 參數詳解

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 每 60s 做一次 checkpoint env.enableCheckpointing(60000); // 高級配置： // checkpoint 語義設置為 EXACTLY_ONCE，這是默認語義 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 兩次 checkpoint 的間隔時間至少為 1 s，默認是 0，立即進行下一次 checkpoint env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000); // checkpoint 必須在 60s 內結束，否則被丟棄，默認是 10 分鍾 env.getCheckpointConfig().setCheckpointTimeout(60000); // 同一時間只能允許有一個 checkpoint env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 最多允許 checkpoint 失敗 3 次 env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); // 當 Flink 任務取消時，保留外部保存的 checkpoint 信息 env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 當有較新的 Savepoint 時，作業也會從 Checkpoint 處恢復 env.getCheckpointConfig().setPreferCheckpointForRecovery(true); // 允許實驗性的功能：非對齊的 checkpoint，以提升性能 env.getCheckpointConfig().enableUnalignedCheckpoints();

相關參數的文字描述：

1. env.enableCheckpointing(60000)，1 分鍾觸發一次 checkpoint；
2. setCheckpointTimeout，checkpoint 超時時間，默認是 10 分鍾超時，超過了超時時間就會被丟棄；
3. setCheckpointingMode，設置 checkpoint 語義，可以設置為 EXACTLY_ONCE，表示既不重復消費也不丟數據；AT_LEAST_ONCE，表示至少消費一次，可能會重復消費；
4. setMinPauseBetweenCheckpoints，兩次 checkpoint 之間的間隔時間。假如設置每分鍾進行一次 checkpoint，兩次 checkpoint 間隔時間為 30s。假設某一次 checkpoint 耗時 40s，那么理論上20s 后就要進行一次 checkpoint，但是設置了兩次 checkpoint 之間的間隔時間為 30s，所以是 30s 之后才會進行 checkpoint。另外，如果配置了該參數，那么同時進行的 checkpoint 數量只能為 1；
5. enableExternalizedCheckpoints，Flink 任務取消后，外部 checkpoint 信息是否被清理。

• DELETE_ON_CANCELLATION，任務取消后，所有的 checkpoint 都將會被清理。只有在任務失敗后，才會被保留；
• RETAIN_ON_CANCELLATION，任務取消后，所有的 checkpoint 都將會被保留，需要手工清理。

1. setPreferCheckpointForRecovery，恢復任務時，是否從最近一個比較新的 savepoint 處恢復，默認是 false；
2. enableUnalignedCheckpoints，是否開啟試驗性的非對齊的 checkpoint，可以在反壓情況下極大減少 checkpoint 的次數;

Flink 1.11 對 Checkpoint 的優化

在以前，在進行對齊的過程中，算子是不會再接着處理數據了，一定要等到對齊動作完成之后，才能繼續對齊

在 Flink 1.11 版本中，引入了一個 Unaligned Checkpointing 的模塊，主要功能是，在 barrier 到達之后，不必等待所有的輸入流的 barrier，而是繼續處理數據

 

轉載自：

Flink Checkpoint 參數詳解 - 寧君 - 博客園 (cnblogs.com)