flink1.12.x Checkpoint 清除策略

作者：马育民 • • 阅读：10229

说明

当 Flink 任务 取消 时，默认删除 外部保存的 CheckPoint 信息，再次运行 Flink 任务时，无法恢复状态

注意： 这里说的是通过命令取消任务，不是运行报错导致的崩溃退出

设置

通过下面代码设置清除策略

env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

参数解释：

  • DELETE_ON_CANCELLATION：值为 true，取消任务时会删除外部checkpoint，因此取消任务后，不能从 Checkpoint 位置进行恢复任务。默认值。
    应用场景：失败时，Checkpoint 才会被保留，并用于任务恢复

  • RETAIN_ON_CANCELLATION：值为 false，取消任务时，保留外部checkpoint
    注意：状态信息将永远保留在外部的持久化存储中，需要手动清除 Checkpoint 状态信息

如何选择

如果 清除策略配置为 DELETE_ON_CANCELLATION，那么在 取消任务 时 Checkpoint 的状态信息会 被清理掉，我们就 无法通过 Checkpoint 来恢复 任务了。

为了从 Checkpoint 中轻量级地进行任务恢复，一般将该参数配置为 RETAIN_ON_CANCELLATION

配置项说明

RETAIN_ON_CANCELLATION

如果配置 RETAIN_ON_CANCELLATION ，一直不清理就会占用 HDFS 存储空间。

DELETE_ON_CANCELLATION

当人为取消任务时，会自动删除检查点，不会占用 HDFS 存储空间

注意：如果 Flink 任务失败了，Checkpoint 的状态信息将被保留，即：需要定期清理

代码

关键代码

env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

完整代码

Configuration conf = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

// 设置检查点和状态后端
env.enableCheckpointing(10*1000L);
// 设置相邻的两个检查点最小的时间间隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10*1000L);
// 清楚策略 env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

必须定期清理 Checkpoint

生产环境中， 状态比较大 的任务，状态信息在 HDFS 三副本 总共占用 几百G - 1 T

如果配置 RETAIN_ON_CANCELLATION，则需要定期清理

参考：https://blog.csdn.net/weixin_34460380/article/details/112103164

原文出处：/show_1IX49aMKB2qQ.html

吉ICP备15003043号