flink的Snapshot

Posted 2023-05-10

参考技术A     flink的snapshot的算法在 这篇 论文，本文讲述也是基于这部分内容，如果有兴趣可以自行参考原文。

  目前已知的能够保证只执行一次（exactly once）语义，依赖于全局、一致性运行状态的快照。但是这么做有两个明显的缺点：

1. 同步的快照需要停止当前所有正在计算的任务

2. 需要保存很多计算无关的状态和记录，需要占用很多额外的空间。

一种简单的实现方案可以分三步来进行：

1. 挂起所有正在执行的计算

2. 快照

3. 快照结束后，继续执行任务

这种实现方案最大的问题是对吞吐量和存储空间都有很大的影响，而且会依赖于上游发送方的数据备份。

Flink采用的是ABS（Asynchronous Barrier Snapshotting）算法。ABS只保存节点的状态，而不保存channel的状态。

1. 由中心调度节点（JobManager）向所有的source注入barrier。当source收到barrier，会对其当前状态做快照，并且广播到所有的output（如图a）

2. 当一个非source节点，收到barrier，那么其会阻塞当前的input，直到收到所有input的barrier（如图b）

3. 收到所有barrier之后，会保存当前的快照，并向所有的output广播barrier（如图c）

4. 快照结束后，解除input的阻塞，继续计算。全局的快照是所有的节点快照

环状图的快照和无环图不完全一致：

1. 首先环状可能会无限的收到某个input的barrier

2. 数据在环状随意流动，可能会丢失数据。

为了解决以上问题，Flink对之前的算法做了一些扩展

1. 通过深度优先查找，首先得到所有的back-edge（就是回路）

2. 运算过程中，当收到所有的barrier时候（不包括回路的barrier），开始备份所有从back-edge发过来的记录，直到收到back-edge的barrier（如图b）

3. 这样在循环里的记录会被记录到本次的快照，如图c

异步屏障快照ABS

Checkpoint & Snapshot

检查点是Flink为流计算过程提供的容错和故障恢复机制。当程序出错时，Flink会重启受到影响的那部分算子及计算逻辑，并将它们重置到最后一次成功checkpoint时的状态。每次成功的checkpoint产生的“状态数据”其实就是这个流式计算任务在那一时刻的快照。

Flink作业可以抽象成有向图表示，图的顶点是算子（operator），边是数据流（data stream），与Chandy-Lamport算法提出的“进程-链路”图模型恰好对应。直接套用C-L算法的思路，我们可以得出如下推论：

• Flink作业的快照要包含两部分，即算子所处的状态以及数据流承载的数据。算子每收到/发出一条数据，以及数据流每流入/流出一条数据，都会造成全局状态的改变。
• 算子可以感知到自己的状态，但数据流的状态不容易记录，主要是因为承载的数据量太大，并且总是在变化。
• 时间是无法静止的（即数据总是在流动的），并且快照不能stop-the-world，否则会造成延迟和数据堆积，降低吞吐量。

所以解决方案的要点有二：一是通过每个算子自己记录的状态合并出全局快照，二是引入一个标记把数据流从时域上切分成段。下面就可以了解ABS算法的基础——屏障。

Barrier

之前已经讲过，C-L算法引入了marker消息来作为快照的边界，即区分“当前快照的数据”和“下一个快照的数据”。ABS算法也有自己的marker消息，不过称为检查点屏障（checkpoint barrier），简称屏障。

屏障由Flink的JobManager周期性产生（周期长度由StreamExecutionEnvironment.enableCheckpointing()方法来指定），并广播给所有Source算子，沿着数据流流动下去。下图示出一条带有屏障的数据流。

可见，第n - 1个屏障之后、第n个屏障之前的所有数据都属于第n个检查点。下游算子如果检测到屏障的存在，就会触发快照动作，不必再关心时间无法静止的问题。下面继续了解快照阶段是如何执行的。

Snapshotting & Barrier Alignment

举例说明检查点流程。下图是论文中给出的并行度为2的Word Count示例，注意该作业的执行计划为有向无环图（DAG）。

快照算法的步骤如下：

a) Source算子接收到JobManager产生的屏障，生成自己状态的快照（其中包含数据源对应的offset/position信息），并将屏障广播给下游所有数据流；

b)、c) 下游非Source的算子从它的某个输入数据流接收到屏障后，会阻塞这个输入流，继续接收其他输入流，直到所有输入流的屏障都到达（图中的count-2算子接收的两个屏障就不是同时到达的）。一旦算子收齐了所有屏障，它就会生成自己状态的快照，并继续将屏障广播给下游所有数据流；

d) 快照生成后，算子解除对输入流的阻塞，继续进行计算。Sink算子接收到屏障之后会向JobManager确认，所有Sink都确认收到屏障标记着这一周期checkpoint过程结束，快照成功。

可见，如果算子只有一个输入流的话，问题就比较简单，只需要在收到屏障之后立即做快照。但是如果有多个输入流，就必须要等待收到所有屏障才能做快照，以避免将检查点n与检查点n + 1的数据混淆。这个等待的过程就叫做对齐（alignment），图来自官方文档。注意算子内部有个输入缓冲区，用来在对齐期间缓存数据。

下图是从Flink系统的角度示出整个checkpoint流程里屏障的流动，以及快照数据向状态后端的写入。注意Source记录的offset值以及Sink收到所有屏障后的ack信号。

Exactly-Once vs At-Least-Once

上面讲到的屏障对齐过程是Flink exactly-once语义的基础，因为屏障对齐能够保证多输入流的算子正常处理不同checkpoint区间的数据，避免它们发生交叉，即不会有数据被处理两次。

但是对齐过程需要时间，有一些对延迟特别敏感的应用可能对准确性的要求没有那么高。所以Flink也允许在StreamExecutionEnvironment.enableCheckpointing()方法里指定At-Least-Once语义，会取消屏障对齐，即算子收到第一个输入的屏障之后不会阻塞，而是触发快照。这样一来，部分属于检查点n + 1的数据也会包括进检查点n的数据里， 当恢复时，这部分交叉的数据就会被重复处理。

Asynchronous

“屏障”和“快照”都讲过了，“异步”呢？这个词实际上指的是快照数据写入的异步性：算子收齐屏障并触发快照之后，不会等待快照数据全部写入状态后端，而是一边后台写入，一边立刻继续处理数据流，并将屏障发送到下游，实现了最小化延迟。

当然，引入异步性之后，checkpoint成功的条件除了所有Sink都报告ack之外，还得加上一条：所有有状态的算子都报告ack，否则JobManager就无法确认异步写入到底完成没有。

DCG？

ABS的精华讲完了。最后看论文中提到的特殊情况，即作业的执行计划是个有向有环图（DCG）。很显然这种情况会造成死锁，环内的算子就会无限等待收齐屏障。面对该问题，ABS算法会单独处理回边（back edge）——即从下游流回上游的数据流，因为回边的存在会导致我们无法单纯地通过每个算子的状态合并出全局快照。

思路如下图所示，重点在于回边终点的那个算子。当该算子的非回边输入流的屏障都到达之后，它会生成一个本地的快照备份，并于此同时开始记录回边流入的数据，直到再次从回边收到相同的屏障。这样就靠算子的状态记录了回边的状态，当从快照恢复时，能够将回边的数据重新放回数据流传输。

The End

原文链接：https://blog.csdn.net/nazeniwaresakini/article/details/106379153