其他分享
首页 > 其他分享> > Flink:容错机制

Flink:容错机制

作者:互联网

一致性检查点(Checkpoints)

image-20210910090421765

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)

Checkpoint执行流程

image-20210910094218032

执行流程如上图:

注意:

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

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

复杂流程

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

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

image-20210910095303765

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

image-20210910095342317

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

image-20210910095431360

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

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

image-20210910095506313

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

image-20210910095540656

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

image-20210910095606673

栅栏对齐

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

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

image-20210910101656420

image-20210910101821696

栅栏只有在需要提供“精确一次”语义保存时,才需要“栅栏对齐”。如果不需要这种语义,则可以通过配置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