Flink 中的 Savepoint 和 Checkpoint 有什么區別？
作者：甜獲師兄
Savepoint 和Checkpoint 都是 Flink 中用于狀態持久化和恢復的機 制，但它們在目的、觸發方式、生命周期管理和使用場景上有著明顯區別。
Savepoint(保存點)和Checkpoint(檢查點)都是Flink中用于狀態持久化和恢復的機制，但它們在目的、觸發方式、生命周期管理和使用場景上有著明顯區別。
一、基本概念
1. Checkpoint(檢查點)
檢查點是Flink自動觸發的狀態快照，用于故障恢復，主要特點：
- 自動創建：由Flink定期自動觸發
- 生命周期：通常在作業運行期間有限存在，舊的檢查點會被新的覆蓋
- 主要目的：實現容錯機制，在故障發生時能夠恢復到最近的一致狀態
- 存儲格式：針對性能優化，可能使用增量存儲機制
2. Savepoint(保存點）
保存點是用戶手動觸發的狀態快照，用于有計劃的作業升級或維護，主要特點：
- 手動創建：由用戶通過命令或API手動觸發
- 生命周期：長期存在，直到用戶明確刪除
- 主要目的：應用版本升級、集群遷移、A/B測試等有計劃的操作
- 存儲格式：更加完整和自包含，確保長期兼容性
二、詳細對比
三、在FlinkSQL 中配置和使用
1. Checkpoint 配置
1.SQL Client 模式
-- 啟用檢查點，每10秒觸發一次
SET 'execution.checkpointing.interval' = '10s';
-- 設置檢查點模式（EXACTLY_ONCE/AT_LEAST_ONCE）
SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
-- 設置檢查點超時時間
SET 'execution.checkpointing.timeout' = '5min';
-- 設置檢查點最小間隔
SET 'execution.checkpointing.min-pause' = '1s';
-- 設置同時進行的檢查點數量上限
SET 'execution.checkpointing.max-concurrent-checkpoints' = '1';
-- 設置檢查點存儲位置
SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';
-- 設置保留的檢查點數量
SET 'state.checkpoints.num-retained' = '5';
-- 作業取消時保留檢查點
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';2. Java API模式
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 每10秒觸發一次檢查點
env.enableCheckpointing(10000);
// 獲取檢查點配置
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
// 設置模式為EXACTLY_ONCE
checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 設置檢查點超時時間
checkpointConfig.setCheckpointTimeout(300000);
// 設置最小間隔時間
checkpointConfig.setMinPauseBetweenCheckpoints(1000);
// 設置最大并發檢查點數
checkpointConfig.setMaxConcurrentCheckpoints(1);
// 設置外部化檢查點的清理行為
checkpointConfig.enableExternalizedCheckpoints(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// 設置檢查點存儲
env.setStateBackend(new FsStateBackend("hdfs:///flink/checkpoints"));2. Savepoint 操作
1. 創建Savepoint
使用Flink CLI:
# 為作業創建保存點
flink savepoint :jobId [:targetDirectory]
# 例如
flink savepoint 1234567890abcdef hdfs:///flink/savepoints
使用REST API:
# 發送POST請求到作業管理器
curl -X POST "http://jobmanager:8081/jobs/:jobId/savepoints" \
-d '{"target-directory": "hdfs:///flink/savepoints", "cancel-job": false}'
2. 從Savepoint恢復
使用Flink CLI:
# 從保存點恢復作業
flink run -s :savepointPath [:runArgs]
# 例如
flink run -s hdfs:///flink/savepoints/savepoint-1234567-aabbccdd jarfile.jar
使用SQL Client:
-- 設置從保存點恢復
SET 'execution.savepoint.path' = 'hdfs:///flink/savepoints/savepoint-1234567-aabbccdd';
-- 執行SQL任務
INSERT INTO target_table SELECT * FROM source_table;
使用Java API:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 設置從保存點恢復
env.setRestartStrategy(RestartStrategies.noRestart());
env.setStateBackend(new FsStateBackend("hdfs:///flink/checkpoints"));
// 指定保存點路徑
String savepointPath = "hdfs:///flink/savepoints/savepoint-1234567-aabbccdd";
Configuration configuration = new Configuration();
configuration.setString(SavepointConfigOptions.SAVEPOINT_PATH, savepointPath);
env.configure(configuration);
// 構建作業并執行
// ...
env.execute("Job restored from savepoint");四、常見應用場景
1. Checkpoint 應用場景
- 故障自動恢復: – 節點崩潰時，作業自動從最近的檢查點恢復 – 網絡分區時，保持數據一致性
- 保證數據處理的一致性: – 實現exactly-once 或at-least-once 語義 – 確保在失敗時不丟失狀態
- 處理反壓和資源限制: – 當系統出現反壓時可以從檢查點恢復 – 資源緊張時重新平衡負載
2. Savepoint 應用場景
(1) 應用升級和版本遷移:
① 在升級前備份當前狀態 -- (通過Flink CLI創建保存點)
flink savepoint 1234567890abcdef hdfs:///flink/savepoints② 停止當前作業
flink cancel 1234567890abcdef③ 部署新版本代碼
④ 從保存點恢復到新版本
flink run -s hdfs:///flink/savepoints/savepoint-1234567-aabbccdd new-version.jar(2) 集群遷移或擴展:
① 在原集群創建保存點
flink savepoint 1234567890abcdef hdfs:///flink/savepoints② 在新集群使用保存點恢復作業
flink run -s hdfs:///flink/savepoints/savepoint-1234567-aabbccdd jarfile.jar(3) A/B 測試和算法切換:
# 創建保存點
flink savepoint 1234567890abcdef hdfs:///flink/savepoints# 使用保存點啟動算法A的實現
flink run -s hdfs:///flink/savepoints/savepoint-1234567-aabbccdd algorithm-a.jar
# 使用同一保存點啟動算法B的實現進行比較
flink run -s hdfs:///flink/savepoints/savepoint-1234567-aabbccdd algorithm-b.jar(4) 生產環境回滾:
# 當新版本出現問題時，使用之前的保存點回滾
flink run -s hdfs:///flink/savepoints/savepoint-previous-version old-version.jar責任編輯：趙寧寧
來源：
大數據技能圈
