标签：Checkpoint 栅栏 快照 Barrier Flink 容错 检查点 机制
一致性检查点（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)));
标签：Checkpoint,栅栏,快照,Barrier,Flink,容错,检查点,机制 来源： https://www.cnblogs.com/wwjj4811/p/15251146.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。