启用检查点

开启自动保存快照 (默认：关闭) ：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每隔 1 秒启动一次检查点保存
env.enableCheckpointing(1000);

间隔调整 :

• 对性能的影响更小，就调大间隔时间
• 为了更好的容错性，就以调小间隔时间

检查点存储

检查点存储 (CheckpointStorage) : 持久化存储位置

• JobManager 的堆内存 (JobManagerCheckpointStorage) : 默认
• 文件系统 (FileSystemCheckpointStorage) : 常用 , (HDFS , S3)

// 配置存储检查点到 JobManager 堆内存
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

// 配置存储检查点到文件系统
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));

通用增量

Rocksdb 状态后端 : 启用增量 checkpoint

• Flink 1.15 后 , HashMap , Rocksdb 都能开启通用的增量 checkpoint

EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(true);

增量 checkpoint 过程 :

1. 带状态的算子任务 , 将状态更改 , 写入变更日志（记录状态）
2. 状态物化：状态表定期保存，独立于检查点
3. 状态物化完成后，状态变更日志 , 就截断到相应的点

注意点 :

• HDFS : 文件数变多
• 上传变更日志 : IO 宽带较大
• 序列化状态变更 : CPU 消耗较大
• 缓存状态变更 : TaskManager 内存消耗较大
• Checkpint 最大并发 = 1
• Flink 1.15 , Memory 测试阶段
• 不支持 NO_ClAIM 模式

配置文件 :

state.backend.changelog.enabled: true
state.backend.changelog.storage: filesystem 

# 存储 changelog 数据
dstl.dfs.base-path: hdfs://hadoop102:8020/changelog 
execution.checkpointing.max-concurrent-checkpoints: 1
execution.savepoint-restore-mode: CLAIM

代码配置 :

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-changelog</artifactId>
    <version>${flink.version}</version>
    <scope>runtime</scope>
</dependency>

// 开启changelog:
env.enableChangelogStateBackend(true);

最终检查点

当有界数据 , 部分Task 完成 , Flink 1.14 后 , 它们依然能进行检查点

禁用 (Flink 1.15 后, 默认启用) :

Configuration config = new Configuration();
// 禁用最终检查点
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);

配置建议

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 获取所有配置
CheckpointConfig checkpointConfig = env.getCheckpointConfig();

// 检查点模式 (CheckpointingMode) : 
//   精确一次 : exactly-once (默认)
//   至少一次 : at-least-once (效率更高)
checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

// 最大并发检查点数量（maxConcurrentCheckpoints）: 
//  检查点最多有多少个
checkpointConfig.setMaxConcurrentCheckpoints(1)

// 启用非对齐的检查点保存
// 限制: CheckpointingMode= exctly-once , 并发的检查点 = 1
checkpointConfig.enableUnalignedCheckpoints();

// 默认: 0: 用非对齐的检查点
// > 0: 用 对齐的检查点（barrier对齐)
// 当对齐时间 > 阈值, 为: 非对齐检查点（barrier非对齐）
checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));

// 超时时间 (checkpointTimeout) : 
//  检查点保存的超时时间，当超时就丢弃
//  单位 : 长整型毫秒数
checkpointConfig.setCheckpointTimeout(60000)

//最小间隔时间 (minPauseBetweenCheckpoints): 
//   上个 checkpoint 完成后, 最快多久触发另个 checkpoint
checkpointConfig.setMinPauseBetweenCheckpoints(500)

// 开启检查点的外部持久化
//   DELETE_ON_CANCELLATION： 作业取消时, 自动删除外部检查点，但作业失败退出，就保留检查点
//   RETAIN_ON_CANCELLATION：作业取消时, 也保留外部检查点
checkpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)


// 检查点异常时, 是否整个任务失败
//   true : 失败提出
//   false: 丢弃, 并继续运行
checkpointConfig.setFailOnCheckpointingErrors(true)