flink-checkpoint实战使用

依赖在我上几篇文章有

// 每 1000ms 开始一次 checkpoint
env.enableCheckpointing(1000);

// 高级选项：

// 设置模式为精确一次 (这是默认值)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 确认 checkpoints 之间的时间会进行 500 ms
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// Checkpoint 必须在一分钟内完成，否则就会被抛弃
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 同一时间只允许一个 checkpoint 进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 开启在 job 中止后仍然保留的 externalized checkpoints
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

以上是代码配置在实际项目中的配置情况如下

连接kafka消费 输出


测试一下 flink run  -c hbase.kafkaconn  kafka_test.jar

在kafka输入数据

能正常消费

数据分布在不同节点 所以我只截取了一个节点的数据
此时stop作业后 继续在kafka生产数据

这是程序运行后的 flink checkpoint 安装作业ID 生成的备份 (最后面是作业ID)
开始 savepoint

此时已经把这个作业的 此刻最新的状态save到这个目录

然后启动flink 作业的时候从这个保存的状态去运行恢复

然后flink会继续消费之前没有消费到的数据