流批模式
SET 'execution.runtime-mode' = 'streaming'; // or batch
基础 Checkpoint 配置
-- 启用 Checkpoint,间隔 5 分钟
SET 'execution.checkpointing.interval' = '5min';
-- Checkpoint 超时时间(10 分钟)
SET 'execution.checkpointing.timeout' = '10min';
-- 最小 Checkpoint 间隔(2.5 分钟)
SET 'execution.checkpointing.min-pause' = '150s';-- 最大并发 Checkpoint 数(默认 1)
SET 'execution.checkpointing.max-concurrent-checkpoints' = '1';
-- Checkpoint 存储路径(HDFS/S3 等) 配合backend state设置
SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';
-- 作业取消后保留 Checkpoint
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';
backend存储路径配置
-- 启用 RocksDB 状态后端
SET 'state.backend' = 'rocksdb'; // hashmap, filesystem-- RocksDBStateBackend
-- RocksDB 本地临时存储路径(多磁盘路径用逗号分隔)
SET 'state.backend.rocksdb.localdir' = '/mnt/ssd1/rocksdb,/mnt/ssd2/rocksdb';
-- 增量 Checkpoint(减少存储开销)
SET 'state.backend.rocksdb.incremental' = 'true';
以下是 Flink SQL Client 中与 Checkpoint 配置相关的常用参数及其详细说明,适用于优化容错机制与作业稳定性:
核心 Checkpoint 参数列表
参数名称 | 默认值 | 作用描述 | 配置示例 |
---|---|---|---|
execution.checkpointing.interval | 无 (需显式设置) | 连续两次 Checkpoint 触发的最小时间间隔(毫秒)。 | SET 'execution.checkpointing.interval' = '60s'; |
execution.checkpointing.timeout | 10 分钟 | Checkpoint 完成的超时时间,超时则触发失败。 | SET 'execution.checkpointing.timeout' = '5min'; |
execution.checkpointing.min-pause | 0 | 两次 Checkpoint 之间的最小暂停间隔(防止连续触发)。 | SET 'execution.checkpointing.min-pause' = '30s'; |
execution.checkpointing.max-concurrent | 1 | 最大并发 Checkpoint 数量(适用于高吞吐场景)。 | SET 'execution.checkpointing.max-concurrent' = '2'; |
execution.checkpointing.externalized-checkpoint-retention | 无 | Checkpoint 保留策略: - DELETE_ON_CANCELLATION (作业取消时删除)- RETAIN_ON_CANCELLATION (作业取消时保留) | SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION'; |
state.backend.incremental | false | 是否启用增量 Checkpoint(需使用 RocksDB 状态后端)。 | SET 'state.backend.incremental' = 'true'; |
state.checkpoints.dir | 无 | Checkpoint 存储目录(需配置为持久化路径,如 HDFS/S3)。 | SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints'; |
配置方式
1. 在 SQL Client 中直接设置
-- 启用 Checkpoint 并设置间隔为 1 分钟
SET 'execution.checkpointing.interval' = '60s';
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';
2. 通过配置文件 sql-client-defaults.yaml
全局配置
execution:checkpointing:interval: 60stimeout: 5minmin-pause: 30smax-concurrent: 2externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
state:backend: rocksdbcheckpoints.dir: hdfs:///flink/checkpointsbackend.incremental: true
参数调优场景建议
-
高吞吐低延迟场景
- 缩短间隔:
interval=10s
(需权衡 Checkpoint 开销)。 - 增大并发:
max-concurrent=2
(避免资源争抢)。 - 增量 Checkpoint:
state.backend.incremental=true
(减少全量快照开销)。
- 缩短间隔:
-
大状态作业(如窗口聚合)
- 延长超时:
timeout=15min
(防止大状态序列化超时)。 - 保留策略:
externalized-checkpoint-retention=RETAIN_ON_CANCELLATION
(便于手动恢复)。
- 延长超时:
-
避免 Checkpoint 风暴
- 设置
min-pause
:如min-pause=30s
(防止频繁触发导致吞吐下降)。 - 监控背压:Checkpoint 失败可能由数据反压引起,需优化算子并行度。
- 设置
Checkpoint 状态恢复
从指定 Checkpoint 重启作业(需保留 Checkpoint 元数据):
# 提交作业时指定恢复路径
./bin/flink run -s hdfs:///flink/checkpoints/<job-id>/chk-1234 ...
注意事项
- 存储路径权限:确保 Flink 进程对
state.checkpoints.dir
有读写权限(如 HDFS 路径需配置 Kerberos 认证)。 - RocksDB 调优:若使用 RocksDB,需额外配置内存参数(如
state.backend.rocksdb.memory.managed=true
)。 - 监控指标:通过 Flink Web UI 监控 Checkpoint 持续时间、大小及失败率,及时调整参数。
通过合理配置 Checkpoint 参数,可显著提升 Flink SQL 作业的容错能力与稳定性,具体数值需根据数据量、集群资源和业务 SLA 动态调整。