一致性检查点（Checkpoints）

Flink 故障恢复机制的核心，就是应用状态的一致性检查点

有状态流应用的一致检查点，其实就是所有任务的状态，在某个时间点的一份拷贝（一份快照）；这个时间点，应该是所有任务都恰好处理完一个相同的输入数据的时候

某一时刻,Flink中所有的Operator的当前State的全局快照,一般存在磁盘上。

表示了一个Flink Job在一个特定时刻的一份全局状态快照，即包含了所有Operator的状态。

可以理解为Checkpoint是把State数据定时持久化存储了。

比如KafkaConsumer算子中维护的Offset状态,当任务重新恢复的时候可以从Checkpoint中获取。

检查点实现算法

Flink中的Checkpoint底层使用了Chandy-Lamport algorithm分布式快照算法可以保证数据的在分布式环境下的一致性!

可参考：https://zhuanlan.zhihu.com/p/53482103

检查点分界线（Checkpoint Barrier）

• Flink 的检查点算法用到了一种称为分界线（barrier）的特殊数据形式，用来把一条流上数据按照不同的检查点分开
• 分界线之前到来的数据导致的状态更改，都会被包含在当前分界线所属的检查点中；而基于分界线之后的数据导致的所有更改，就会被包含在 之后的检查点中

Checkpoint执行流程

执行流程如上图：

• 1.Flink的JobManager创建CheckpointCoordinator
• 2.Coordinator向所有的SourceOperator发送Barrier栅栏(执行Checkpoint的信号)
• 3.SourceOperator接收到Barrier之后,暂停当前的操作(暂停的时间很短,因为后续的写快照是异步的),并制作State快照, 然后将自己的快照保存到指定的介质中(如HDFS), 一切 ok之后向Coordinator汇报并将Barrier发送给下游的其他Operator
• 4.其他的如TransformationOperator接收到Barrier,重复第3步,最后将Barrier发送给Sink
• 5.sink接收到Barrier之后重复第3步
• 6.Coordinator接收到所有的Operator的执行ok的汇报结果,认为本次快照执行成功

注意:

1.在往介质(如HDFS)中写入快照数据的时候是异步的(为了提高效率)

2.分布式快照执行时的数据一致性由Chandy-Lamport algorithm分布式快照算法保证!

复杂流程

下图左侧是 Checkpoint Coordinator，是整个 Checkpoint 的发起者，中间是由两个 source，一个 sink 组成的 Flink 作业，最右侧的是持久化存储，在大部分用户场景中对应 HDFS。

1.Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint。

2.source 节点向下游广播 barrier，这个 barrier 就是实现 Chandy-Lamport 分布式快照算法的核心，下游的 task 只有收到所有 input 的 barrier 才会执行相应的 Checkpoint。

3.当 task 完成 state 备份后，会将备份数据的地址（state handle）通知给 Checkpoint coordinator。

4.下游的 sink 节点收集齐上游两个 input 的 barrier 之后，会执行本地快照，(栅栏对齐)

这里还展示了 RocksDB incremental Checkpoint (增量Checkpoint)的流程，首先 RocksDB 会全量刷数据到磁盘上（红色大三角表示），然后 Flink 框架会从中选择没有上传的文件进行持久化备份（紫色小三角）。

5.同样的，sink 节点在完成自己的 Checkpoint 之后，会将 state handle 返回通知 Coordinator。

6.最后，当 Checkpoint coordinator 收集齐所有 task 的 state handle，就认为这一次的 Checkpoint 全局完成了，向持久化存储中再备份一个 Checkpoint meta 文件。

栅栏对齐

作业图中的每个算子在接收到检查点栅栏时会记录其状态。拥有两个输入流的算子(如CoProcessFunction)会执行栅栏对齐，以便当前快照能够包含消费在两个输入流检查点栅栏之前(但不超过)的所有事件而产生的状态。

栅栏对齐的工作流程图如下：

栅栏只有在需要提供“精确一次”语义保存时，才需要“栅栏对齐”。如果不需要这种语义，则可以通过配置CheckpointingMode.AT_LEAST_ONCE关闭“栅栏对齐”来提高性能。

保存点（Savepoint）

Savepoint:保存点,类似于以前玩游戏的时候,遇到难关了/遇到boss了,赶紧手动存个档,然后接着玩,如果失败了,赶紧从上次的存档中恢复,然后接着玩。

在实际开发中,可能会遇到这样的情况:如要对集群进行停机维护/扩容...

那么这时候需要执行一次Savepoint也就是执行一次手动的Checkpoint/也就是手动的发一个barrier栅栏,那么这样的话,程序的所有状态都会被执行快照并保存,

当维护/扩容完毕之后,可以从上一次Savepoint的目录中进行恢复!

Checkpoint和Savepoint区别：

	Checkpoint	Savepoint
触发管理方式	由flink自动触发管理	由用户手动触发管理
主要用途	在task发生异常时快速恢复	有计划地进行备份，使作业能停止后再恢复
特点	轻量 自动从故障中恢复 在作业停止后默认清除	持久 以标准格式存储，允许代码或配置发生改变 手动触发从savepoint的恢复

检查点和重启策略配置

//检查点配置
//参数一：Checkpoint时间间隔
//参数二：检查点策略
env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
checkpointConfig.setCheckpointTimeout(60000);//超时时间
checkpointConfig.setMaxConcurrentCheckpoints(4);//最大同时进行Checkpoints
checkpointConfig.setMinPauseBetweenCheckpoints(100);//两次Checkpoints之间最小间隔
checkpointConfig.setPreferCheckpointForRecovery(true);//更倾向于使用Checkpoint做恢复
checkpointConfig.setTolerableCheckpointFailureNumber(3);//容忍Checkpoint失败的次数

//重启策略配置
//1.固定延迟重启
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 60000));
//2.在两分钟内失败三次，每次尝试重启间隔时间是1分钟
env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(2), Time.minutes(1)));