Flink 的 Checkpoint配置详解

07-03 1377阅读

先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里P7

深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年最新大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

如果你需要这些资料，可以添加V获取：vip204888 （备注大数据）

正文

3、启用检查点

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 启用 Checkpoint 每 5 秒 一次，模式为 EXACTLY_ONCE
env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

4、常用配置参数

1）最终检查点

// 最终检查点，1.15开始，默认是true
configuration.set(ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);

2）开启 Changelog

// 要求checkpoint的最大并发必须为1
env.enableChangelogStateBackend(true);

3）代码中用到HDFS，需要导入hadoop依赖、指定访问HDFS的用户名

System.setProperty("HADOOP_USER_NAME", "HADOOP");

4）开启非对齐检查点（barrier非对齐）

// 开启的要求： Checkpoint模式必须是精准一次，最大并发必须设为1
checkpointConfig.enableUnalignedCheckpoints();
// 开启非对齐检查点才生效： 默认0，表示一开始就直接用 非对齐的检查点
// 如果大于0，一开始用 对齐的检查点（barrier对齐），对齐的时间超过这个参数，自动切换成 非对齐检查点（barrier非对齐）
checkpointConfig.setAlignedCheckpointTimeout(Duration.ofMinutes(4));

5）检查点常用配置

// 1、启用检查点: 默认是barrier对齐的，周期为5s, 精准一次
env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        
// 2、指定检查点的存储位置
checkpointConfig.setCheckpointStorage("hdfs:///ip:port/dir");
// 3、checkpoint的超时时间: 默认10分钟
checkpointConfig.setCheckpointTimeout(60000);
// 4、同时运行中的checkpoint的最大数量
checkpointConfig.setMaxConcurrentCheckpoints(1);
// 5、最小等待间隔: 上一轮checkpoint结束 到 下一轮checkpoint开始 之间的间隔
checkpointConfig.setMinPauseBetweenCheckpoints(1000);
// 6、取消作业时，checkpoint的数据 是否保留在外部系统
// DELETE_ON_CANCELLATION:主动cancel时，删除存在外部系统的chk-xx目录 （如果是程序突然挂掉，不会删）
// RETAIN_ON_CANCELLATION:主动cancel时，外部系统的chk-xx目录会保存下来
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// 7、允许 checkpoint 连续失败的次数，默认 0 表示 checkpoint 一失败，job 就挂掉
checkpointConfig.setTolerableCheckpointFailureNumber(10);