从0到1Flink的成长之路(二十)-Flink 高级特性之 Flink 容错机制

Posted 2021-07-25 熊老二-

Flink 容错机制

checkpoints
Checkpoint

State Vs Checkpoint
State：
维护/存储的是某一个Operator的运行的状态/历史值，是维护在内存中。
一般指一个具体的Operator的状态（operator的状态表示一些算子在运行的过程中会产生的一些历史结果，如前面的maxBy底层会维护当前的最大值，也就是会维护一个keyedOperator，这个State里面存放就是maxBy这个Operator中的最大值）；State数据默认保存在Java的堆内存中/TaskManage节点的内存中，State可以被记录，在失败的情况下数据还可以恢复。
Checkpoint:
某一时刻，Flink中所有的Operator的当前State的全局快照，一般存在磁盘上。
表示了一个Flink Job在一个特定时刻的一份全局状态快照，即包含了所有Operator的状态。可以理解为Checkpoint是把State数据定时持久化存储，比如FlinkKafkaConsumer算子中维护的Offset状态，当任务重新恢复的时候可以从Checkpoint中获取。  注意:
北京市昌平区建材城西路金燕龙办公楼一层 电话：400-618-9090
Flink中的Checkpoint底层使用了Chandy-Lamport algorithm分布式快照算法，可以保证数据的在分布式环境下的一致性，https://zhuanlan.zhihu.com/p/53482103。
Chandy-Lamport algorithm算法的作者也是Zookeeper中Paxos一致性算法的作者
https://www.cnblogs.com/shenguanpu/p/4048660.html。
Flink中使用Chandy-Lamport algorithm分布式快照算法取得了成功，后续StructuredStreaming也借鉴了该算法。
Checkpoint 执行流程

1）、简单流程

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

注意：
1.在往介质(如HDFS)中写入快照数据的时候是异步的(为了提供效率)
2.分布式快照执行时的数据一致性由Chandy-Lamport algorithm分布式快照算法保证
2）、复杂流程
下图左侧是 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 文件。