Flink Checkpoint 中的Aligned Checkpoint 和 Unaligned Checkpoint

知识点

反压

反压是流式系统中关于处理能力的动态反馈机制，并且是从下游到上游的反馈，一般是在实时数据处理的过程中，上游节点的生产速度大于下游节点的消费速度。在Flink中，反压主要有两个部分：跨TaskManager的反压过程和TaskManager内的反压过程。

https://blog.csdn.net/Stray_Lambs/article/details/120578294

Checkpoint

Flink 借助Checkpoint机制来保证有状态的分布式计算

所谓Checkpoint，就是周期性的进行Snapshot的过程

当算子Failover的时候，通过快照恢复算子的状态，

Flink的Checkpoint是基于Chandy-Lamport（CL）算法改进的

ps：
Spark中Shuffle是划分Stage的边界，每个Shuffle阶段数据会进行分区、排序、聚合、写入磁盘等操作，自然的进行了状态的保存。所以Spark中无复杂的Checkpoint机制。

Barrier

先说明一下barrier这个概念
以一定的间隔被插入到Source节点，经过若干个算子，不断的向下游传递。

当barrier到达算子的时候，触发Snapshot，当前算子的状态保存完毕后，传递到下一个算子。

当一个算子对应多个输入时，也就意味着这个算子会接收到多个barrier，此时为了保证全局一致性，那么就需要对齐多个输入的barrier，这个就是Aligned Checkpoint

Aligned Checkpoint

存在的问题：

1. 对齐时间长，反压时被完全阻塞。
   原因：每条链路的处理速度是不一样的，barrier对齐需要时间，如果某一条链路有反压，会因为需要对齐这个过程，使得整个链路被阻塞，因为barrier没有对齐，为了保障数据一致性，checkpoint也无法进行，接下来Checkpoint超时失败，重新进行Checkpoint，但是由于反压仍然存在，最终陷入失败–重启–失败的循环

Unaligned Checkpoint

为了解决Aligned Checkpoint中存在的反压严重时Checkpoint失败的问题

提出了Unaligned Checkpoint

核心思想

允许Barrier超越ongoing data（正在进行的数据，在buffer中，还没有进入task的数据，比如下图中蓝色的2、3、4、5），barrier超越了这些数据，那如果算子Failover的话，这些数据不就丢失了，解决的方法是，将这些数据也进行快照，在Failover时重放这部分数据。

barrier1顺利超越ongoing data，进入到task中，算子收到全部task后开始进行snapshot

实现原理

假设当前task的上游并行度为3，下游并行度为2，如上图所示，task有3个输入和两个输出，矩形表示buffer中的一条一条数据

Unaligned Checkpoint 这里简称UC

整个UC分为UC同步阶段、UC异步阶段两部分

UC同步阶段

UC开始后，task的3个input-buffer会陆续收到上游发送的barrier，如图所示input-buffer1收到了barrier1，其它的input-buffer还没有收到barrier，当某一个input-buffer接收到barrier时，task会直接开始UC的第一阶段，即UC同步阶段。

这个阶段只需要三个input-buffer中任意一个buffer的barrier进入到task的网络缓冲内存中，task就会直接开始UC，不用等其它的Input-buffer接收到barrier，也不需要处理完input-buffer1接收到barrier1之前的数据。

Flink 网络缓冲区
https://www.jianshu.com/p/cfbb0cf69ae3

UC同步阶段：barrier超越ongoing data
如下图，可以看到barrier超越了input-buffer和output-buffer中的所有数据，到达下游output-buffer的头部，被快速的传递给下游的task，这也解释了为什么在反压情况下UC可以成功

从task层面来看，barrier可以在task内部实现快速超车
从Job层面来看，如果每一个task内部，barrier都可以快速超车，那么barrier就可以从source task快速的超车到Sink task

为了保证数据一致性，UC同步阶段，task不能处理数据

UC同步阶段的四个主要流程

1. barrier超车，当算子的某个input-buffer接收到barrier时，超越ongoing data，快速的将其传递到output-buffer的头部，保证其可以快速到达下游算子
2. buffer引用，对buffer进行引用，这里不进行snapshot，真正的快照在UC异步阶段进行
3. 调用task的SnapshotState方法
4. StateBackend同步快照

UC异步阶段

UC同步阶段完成后，task继续处理数据，同时进行UC的第二个阶段，barrier对齐和UC异步阶段。

首先异步阶段要快照同步阶段所有引用的input-buffer和output-buffer以及同步阶段算子内部引用的State

UC异步阶段其实也有barrier对齐，当task开始UC的时候，很多input-buffer没有接收到barrier，这些input-buffer之前可能还有一些buffer需要快照，例如上图的绿色数据块，所以UC异步阶段要等到所有的input-buffer barrier都到达，且barrier之前的所有buffer都需要快照，这就是UC异步阶段的barrier对齐

这个对齐过程理论上会很快，因为链路中的每一个task，barrier都可以快速的超越所有input-buffer、output-buffer，优先传递barrier到下游task。

异步阶段需要写三部分数据到DFS，分别是UC同步阶段引用的算子内部的State、同步阶段引用的所有input-buffer和output-buffer、以及其它input-buffer barrier之前的buffer

这三部分数据写完之后，task会将结果汇报到TaskManager。

资料：
Flink Unaligned Checkpoint 在 Shopee 的优化和实践