Flink的检查点（CheckPoint）算法_flink检查点算法

Flink检查点的核心作用是确保状态正确，即使遇到程序中断，也要正确。

flink的检查点算法用到了一种特殊的数据类型  checkpoint barrier （检查点分割线），在checkpoint barrier到来之前的所有状态更改都会存到该分割线所属检查点中，而在checkpoint barrier到来后的数据会存到下一个检查点中

以下举例解析

val stream: DataStream[(String, Int)] = ... 
val counts: DataStream[(String, Int)] = stream
.keyBy(record => record._1)
.mapWithState(  (in: (String, Int), state: Option[Int])  => 
state match { 
case Some(c) => ( (in._1, c + in._2), Some(c + in._2) ) 
case None => ( (in._1, in._2), Some(in._2) )
})

该程序有两个算子：keyBy按照数据的第一个元素分组发往不同的分区，再将数据发送给下一个算子mapWithState（有状态的map），该算子会将收到的数据第二个元素加到现有的结果中，再发送出去。

1.下图显示了该程序的初始状态，有六条数据分三个 source 传入，每两条数据中有一个 CheckPoint Barrier （ckpt），根据数据的第一个元素，会经过keyBy算子发往不同的map算子执行。 注意 初始状态下 计数状态都是0（圆柱体中的数值），每条数据的处理顺序 将会严格按照顺序，比如 “b”,2 会在ckpt之前，“a”,2 则会在ckpt之后。

2、ckpt是一种特殊的数据类型，他并不会参与map算子的计算，但是会触发与检查点相关的操作。当ckpt到达数据源的时候（该图例与keyBy画在一起），他将输入流中的位置存到检查点中（最下边的圆柱体）如果输入流是kafka，那这个位置就是偏移量。 检查点存放的位置可以是HDFS

 

3、ckpt随着数据一起流动，当所有的ckpt流动到map算子时，会将 map算子的状态信息异步存储（不会影响程序的正常进行）到持久化存储中

说明：假如有某个 map 的 ckpt 先到，不会立即触发 存储动作，而是所有的 ckpt 全部到达以后才会触发（即ckpt前边的数据全部被处理完）

 4、当map算子的状态和检查点的位置备份完成以后，将会 存储到 持久化设备，会被jobManager标记为检查点操作完成。

 5.如果一个检查点的操作失败了，flink 会直接忽略，因为很有可能在下一个检查点存储成功，尽管这会让错误恢复的时间更长。只有当连续的多个检查点失败后，flink才会报错。

6.假如一个检查点已经完成，发生了错误，flink会如何恢复：

 7.在这种情况下，flink会重新拓扑，并且会将检查点后的数据重播，在我们这个例子中“a”,2、"b",2、"c",2 会被重播，下图展示了这一过程

8.该操作可以保证flink 重新计算以后，在检查点的map的状态不会发生变化，得到恢复

Flink检查点算法的正式名称是异步分界线快照(asynchronous barrier snapshotting)。该算法大致基于Chandy-Lamport分布式快照算法。

 Flink的检查点算法，可以让其内部保证 exactly-once，且不会牺牲性能