从盘古开天辟地说起为什么 Flink CP 能实现精确一次？（下）

1.前言

对于很多做离线或者实时数仓的小伙伴来说，我先问几个问题，看看小伙伴萌能回答上来吗？

1. ⭐ 你知道状态是什么吗？在离线数据开发的经历中，你碰到过状态的概念吗？
2. ⭐ 为什么离线数仓不需要状态，实时数据开发中老是提到状态的概念？
3. ⭐ Flink 中的状态、状态后端、全局一致性快照（CheckpointSavepoint） 的作用都是什么，这三个概念的关联又是什么？
4. ⭐ Flink 是通过什么机制来做 Checkpoint 的？为什么这套机制能够做到故障恢复呢？
5. ⭐ Flink Checkpoint 是基于 Chandy-Lamport 算法的，但是 Flink 的实现相比 Chandy-Lamport 算法之间又有哪些优点、缺点？
6. ⭐ Flink Checkpoint 用到了 barrier，为什么用了 barrier 做的快照就能保证全局一致性快照的正确性？barrier 到底起到了什么作用？
7. ⭐ Flink 对齐 Checkpoint 和非对齐 Checkpoint 的区别是什么？非对齐 Checkpoint 也能保障精确一次吗？

小伙伴们思考一下，都能回答上来么，如果对于某些问题你还有疑问，楼主会通过本篇文章帮你解答这些问题，理清这些概念！

由于本文内容较多，所以博主将本文分为上，中，下三集，本文是下，三集内容是有连接关系的，如果小伙伴在看本文的过程中对有些概念不清楚，可以跳转到上文进行查看：

其中上集介绍了关于状态以及做状态管理能为我们的生产提供什么样的帮助。

从盘古开天辟地说起为什么 Flink CP 能实现精确一次？（上）

中集介绍了一个分布式应用的全局一致性快照包含的内容以及Flink Checkpoint 的 barrier 是干啥用的，为啥使用 barrier 就能做出来一个全局一致性的快照。

从盘古开天辟地说起为什么 Flink CP 能实现精确一次？（中）

本文是下集，主要就是基于上、中两集之后介绍博主开头介绍到的分布式应用通用全局一致性快照的算法流程。主要包括：

1. ⭐ 分布式应用通用异步全局一致性快照算法流程总结

这就是博主所说的：分布式应用通用异步全局一致性快照算法流程总结

1. ⭐ Chandy-Lamport 算法流程、例子

在介绍完了博主所说的通用算法之后，介绍一下 Chandy-Lamport 算法流程并以一个例子介绍其执行过程，并且其和博主的 通用异步全局一致性快照算法 之间的关系，你可以发现两者通用异步全局一致性快照算法是包含了 Chandy-Lamport 算法的

1. ⭐ Flink 实现的全局一致性快照介绍

介绍完上述两种算法之后，介绍 Flink 的异步全局一致性快照，并且也说明了其和 通用异步全局一致性快照算法、Chandy-Lamport 算法 之间的关系

1. ⭐ 分布式应用异步全局一致性快照方法、Chandy-Lamport 算法、Flink 全局一致性快照之间的关系

主要解释了 Chandy-Lamport 算法、Flink 全局一致性快照是分布式应用异步全局一致性快照算法一种特殊情况

1. ⭐ 参考文献

2.分布式应用通用异步全局一致性快照算法流程总结

2.1.名词解释

1. ⭐ process：指分布式应用中的进程，举个 Flink 中的例子就是 TaskManager
2. ⭐ channel：指分布式应用中进程之间的传输通道，举个 Flink 中的例子就是 TaskManager 之间传输数据的网络传输通道；input channel 指一个 process 的输入 channel

2.2.算法流程总结

1. ⭐ 发起快照：有一个 manager process（这个 manager 可以是所有 process 中的任意一个 process，也可以是一个单独的中央管理者）告诉所有的 process 说可以开始做快照了；
2. ⭐ process 记录 process 快照：所有 process 就开始记录自己本地的状态（不包括 input channel 的状态）了，记录完本地状态，然后发 marker 给下游所有的 channel；
3. ⭐ process 记录 channel 快照：每个 process 做完自己的快照之后，就对每一个 input channel 记录所有的输入消息，直到 marker 消息到达之后，将这期间的所有消息作为这个 input channel 的快照；
4. ⭐ process 完成快照：当收到上游所有 marker 之后，这个 process 要记录的状态就全部得到了，然后告诉 manager process 说做完快照了；
5. ⭐ 终止快照：manager process 接收到所有 process 做完快照的消息之后，整个全局一致性快照就做完了。

2.3.算法流程示例

2.3.1.发起快照

1. ⭐ 其中 p 和 q 是两个进程，p 和 q 的消息可以自己处理，也可以互相发送
2. ⭐ 有一个 manager process（如下图，这个 manager process 就是 p 进程）告诉所有的 process （在图中告诉了 p 进程和 q 进程）可以开始做状态了；下图中蓝色圆圈为进程的消息，其中的编号是每个进程的消息的编号

41

2.3.2.process 执行快照

1. ⭐ 所有 process 就开始记录自己本地的状态（不包括 input channel 的状态），记录完本地状态，然后发 marker 给下游所有的 channel；以下图案例来说，p 进程记录 S(p)，q 进程记录 S(q)

42

1. ⭐ 每个 process 做完自己的快照之后，就对每一个 input channel 记录所有的输入消息，直到 marker 消息到达之后，将这期间的所有消息作为这个 input channel 的快照；以下图案例来说，p 进程记录 S(Cqp)，q 进程记录 S(Cpq)，将做完状态后及 marker 之间的消息作为 channel 的状态

43

2.3.3.终止快照

1. ⭐ 当收到上游所有 marker 之后，这个 process 要记录的状态就全部得到了，然后告诉 manager process 说做完状态了；以下图为例，p 进程记录完成 S(p)，S(Cqp)，q 进程记录完成 S(q)，S(pq)，然后 p 进程就告诉 p 进程做完快照了，q 进程告诉 p 进程做完快照了
2. ⭐ manager process 接收到所有 process 做完快照的消息之后，整个全局一致性快照就做完了；注意其中 marker 不参与状态计算

44

3.Chandy-Lamport 算法流程、示例

3.1.算法流程总结

1. ⭐ 发起快照：有一个 manager process（这个 manager 是所有 process 中的任意一个 process）发起快照；
2. ⭐ manager process 记录 process 快照：manager process 就开始记录自己本地的状态（不包括 input channel 的状态），记录完本地状态，然后发 marker 给下游所有的 channel；
3. ⭐ manager process 记录 channel 快照：manager process 做完自己的快照之后，就对每一个 input channel 记录所有的输入消息，直到 marker 消息到达之后，将这期间的所有消息作为这个 input channel 的快照；
4. ⭐ 其他 process 记录 process 快照：process 接收到上游所有 input channel 的第一个 marker 之后开始记录 process 快照
5. ⭐ 其他 process 记录 channel 快照：process 做完自己的快照之后，就对每一个 input channel 记录所有的输入消息，直到 marker 消息到达之后，将这期间的所有消息作为这个 input channel 的快照（注意由于这个 process 有一个 input channel 的 marker 已经到了，所以其记录的所有 input channel 状态中，有一个 input channel 的状态会为空）；
6. ⭐ 终止快照：所有的 process 都把本地快照和 channel 快照做完之后，就向 manager process 发送做完的消息，manager process 接收到所有 process 做完快照的消息之后，整个全局一致性快照就做完了。

3.1.1.发起快照

17

3.1.2.process 执行快照

3.1.3.终止快照

20

3.2.算法流程示例

3.3.Chandy-Lamport 与上节分布式应用全局一致性快照算法的异同

47

Chandy-Lamport 就是上节分布式应用通用全局一致性快照算法的其中一种特殊形式；分布式应用通用全局一致性快照算法中说的是每个 process 在接收到 manager 做快照的消息之后就直接可以开始记录状态了，而 Chandy-Lamport 其实就是把这个 manager 的消息用从 input channel 接收到的第一个 marker 消息给代替了，用数学表达式表示就是接收到第一个 marker 的 channel 的 n = n' = m = m'，剩余的 channel 满足 n = n' ≥ m = m'，可以看到 Chandy-Lamport 就是通用全局一致性快照算法的一种特殊形式。

4.Flink 实现的全局一致性快照介绍

4.1.Flink 的全局一致性快照算法流程总结

Flink 全局一致性快照论文[1]

1. ⭐ 发起快照：有一个 manager process（这个 manager 就是 JobManager）发起快照，发消息给数据源 TM 告知可以开始做快照了；
2. ⭐ 数据源 TM 记录快照：数据源 TM 就开始记录自己本地的状态（数据源没有 input channel 的状态）了，记录完本地状态，然后发 barrier（其实就是 marker） 给下游所有的 channel；
3. ⭐ 下游 TM 记录快照：TM 是要接收到上游所有 input channel 的barrier（marker） 之后开始记录 TM 快照。注意：如果有的 channel barrier 相比其他 channel 的 barrier 先到了，那么不会接着处理这个 channel 的输入消息。并且由于 Flink 的 TM 要等到所有的 input channel 的 barrier 之后才开始做快照，所以 TM 就不用记录 input channel 的状态了；
4. ⭐ 终止快照：所有的 TM 都把本地快照做完之后，就向 JobManager 发送做完的消息，JobManager 接收到所有 TM 做完快照的消息之后，整个全局一致性快照就做完了。

关于 Exactly-Once 和 At-Least-Once 的区别如下：

4.2.Flink 的全局一致性快照与上节分布式应用全局一致性快照算法的异同

Flink 的全局一致性快照就是上节分布式应用通用全局一致性快照算法的其中一种特殊形式；分布式应用通用全局一致性快照算法中说的是每个 process 在接收到 manager 做快照的消息之后就直接可以开始记录状态了，而 Flink 其实就是将各个 process 开始做状态的时间点设为了接收到上游 input channel 所有的 barrier，这样一个好处就是由于各个 process 是接收到了上游所有 barrier 之后开始的，用数学表达式表示其实就满足了 n = n' = m = m'，就没有必要存储 channel 中的状态了。

上述介绍到的 Flink 的全局一致性快照是对齐 Checkpoint 的场景，其实非对齐 Checkpoint 的场景就是 Chandy-Lamport 算法，需要记录 Channel 中的状态。有的场景下会由于 barrier 对齐导致产出延时较大，而非对齐 Checkpoint 可以缓解这种问题，但是缺点在于需要记录 channel 中的状态，状态大小可能会变大。

5.分布式应用全局一致性快照算法、Chandy-Lamport 算法、Flink 全局一致性快照之间的关系

6.参考文献

• https://developer.aliyun.com/article/667562
• https://matt33.com/2019/10/27/paper-chandy-lamport/
• https://developer.aliyun.com/article/448900