为了使Flink的状态具有良好的容错性，Flink提供了检查点机制（Checkpoints）。通过检查点机制，Flink定期在数据流上生成checkpoint barrier ，当某个算子收到barrier时，即会基于当前状态生成一份快照，然后再将该barrier传递到下游算子，下游算子接收到该barrier后，也基于当前状态生成一份快照，依次传递直至到最后的Sink算子上。当出现异常后，Flink 就可以根据最近的一次的快照数据将所有算子恢复到先前的状态。
开启检查点
默认情况下，检查点机制是关闭的，需要在程序中进行开启
// 开启检查点机制，并指定状态检查点之间的时间间隔 env.enableCheckpointing(1000); // 其他可选配置如下： // 设置语义 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 设置两个检查点之间的最小时间间隔 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); // 设置执行Checkpoint操作时的超时时间 env.getCheckpointConfig().setCheckpointTimeout(60000); // 设置最大并发执行的检查点的数量 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 将检查点持久化到外部存储 env.getCheckpointConfig().enableExternalizedCheckpoints( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 如果有更近的保存点时，是否将作业回退到该检查点 env.getCheckpointConfig().setPreferCheckpointForRecovery(true);保存点机制（Savepoints）是检查点机制一种特殊的实现，它允许你通过手工方式来触发Checkpoint，并将结果持久化存储到指定路径中，主要用于避免Flink集群在重启或升级时导致状态丢失。
# 触发指定id的作业的Savepoint，并将结果存储到指定目录下 bin/flink savepoint :jobId [:targetDirectory]首先最容易注意到的是 Savepoint 是一种特殊的 Checkpoint，实际上它们的存储格式也是一致的，它们主要的不同在于定位。Checkpoint机制的目标在于保证Flink作业意外崩溃重启不影响exactly once准确性，通常是配合作业重启策略使用的。而 Savepoint 的目的在于在 Flink 作业维护（比如更新作业代码）时将作业状态写到外部系统，以便维护结束后重新提交作业可以到恢复原本的状态。 换句话讲，Checkpoint 是为 Flink runtime 准备的，Savepoint 是为 Flink 用户准备的。因此 Checkpoint 是由 Flink runtime 定时触发并根据运行配置自动清理的，一般不需要用户介入，而 Savepoint 的触发和清理都由用户掌控。
其次，由于Checkpoint 频率远远大于Savepoint，Flink对Checkpoint格式进行针对不同StateBackend 的优化，因此它在底层存储效率更高，而代价是耦合性更强，比如不保证rescaling （即改变作业并行度）的特性和跨版本兼容。这里说”不保证”而不是”不支持”，原因是实际上 RocksDB 的 Checkpoint 是支持 rescaling 的，”不保证”更多是从系统设计出发而言。跨版本兼容性也同理。
最后，Savepoint 的定义有提及它是 non-incremental 的，这是相对于 incremental Checkpoint 来说。因为 Checkpoint 是秒级频繁触发的，两个连续 Checkpoint 通常高度相似，因此对于 State 特别大的作业来说，每次 Checkpoint 只增量地补充 diff 可以大大地节约成本，这就是 incremental Checkpoint 的由来。而 Savepoint 并不会连续地触发，而且比起效率，Savepoint 更关注的是可移植性和版本兼容性，所以不支持 incremental 也是理所当然。
参考
https://www.whitewood.me/2018/09/06/Flink-Checkpoint-Savepoint-%E5%B7%AE%E5%BC%82/
公众号
名称：大数据计算 微信号：bigdata_limeng
