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

为什么我要把这段话放在最前面呢？因为博主有了大发现，博主在总结学习的过程中，总结了除了 Flink CP、Chandy-Lamport 全局一致性快照算法之外的一种 通用全局一致性快照算法！！！。

这套 通用算法 包含 Chandy-Lamport 算法 ≈ Flink 非对齐 CP 算法 包含 Flink 对齐 CP 算法。

可能这一套 通用算法 之前已经有人提过了，但是博主是自己在总结 Flink CP、Chandy-Lamport 算法的过程中，逆推总结出来的，并没有借助外力！！！

1.前言

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

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

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

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

本文最主要的内容就是解释了：

一个分布式应用是怎么异步做一个全局一致性快照？

2.名词解释

⭐ Single-Token conservation：一个最常见的分布式应用，单 Token 流转分布式应用
⭐ Process：指分布式应用中的进程，举个 Flink 中的例子就是 TaskManager
⭐ Channel：指分布式应用中进程之间的传输通道，举个 Flink 中的例子就是 TaskManager 之间传输数据的网络传输通道

3.分布式应用全局一致性快照要记录的状态内容

首先在分析一个复杂的大数据应用的全局一致性快照之前，我们先以最简单的分布式应用为例。

Single-Token conservation：其有 p 和 q 两个进程，p 可以通过 Channel pq（记为 Cpq）向 q 发消息，q 可以通过 Channel qp（记为 Cqp）向 p 发消息，其中有一个叫 token 的消息，在这个系统中一直不停的传输流转，从 p 到 q，再从 q 到 p。

⭐ 首先我们来分析这个应用中，全局一致性快照应该包含哪些内容？

⭐ 结论：全局一致性快照 = Process 状态 Channel 状态。

⭐ 原因：以上面的四幅图为例，每一幅图代表一个时刻，如果我们以拍照这种方式做全局一致性快照来理解时，那么同一时刻，Process 和 Channel 同时都会存在数据，这些数据都是作为全局一致性快照的一部分内容。

使用上述的这个结论，我们可以得到上图 Single-Token conservation 示例中的全局一致性快照 S = S(p) S(Cpq) S(q) S(Cqp)

其中：

S：全局一致性快照
S(p)：p 进程的状态
S(Cpq)：p 进程到 q 进程的 Channel 状态
S(q)：q 进程的状态
S(Cqp)：q 进程到 p 进程的 Channel 状态

这里就碰到了我们要分析的关键问题：做全局一致性快照时，小伙伴萌都容易理解S(p)，S(q)这两个，因为这两份状态数据就真实的存在我们的分布式应用中，但是S(Cpq)，S(Cqp)这两个怎么理解呢？这些数据都是在网络中传输啊，我们做全局一致性快照时用啥方法才能把这些数据也记录下来呢？接下来详细讲讲博主的理解

4.Process 状态记录的内容

记录和实际业务相关的状态内容。举例：id 去重就存储历史所有的 id 就可以了。

5.Channel 状态记录的内容

还是以前文的 Single-Token conservation 为例：

token 在 p 时（对应第一张图），这时的全局一致性快照为：

S(token-in-p) = S(p-token-in-p) S(Cpq-token-in-p) S(q-token-in-p) S(Cqp-token-in-p)

其中：

S(token-in-p)：token 在 p 时，做的全局一致性快照
S(p-token-in-p)：token 在 p 时，p 进程的状态
S(Cpq-token-in-p)：token 在 p 时，p 进程到 q 进程的 Channel 状态
S(q-token-in-p)：token 在 p 时，q 进程的状态
S(Cqp-token-in-p)：token 在 p 时，q 进程到 p 进程的 Channel 状态

其中 S(p-token-in-p) 好理解，做快照时，token 还没有从 p 发出去，p 肯定知道 token 还在 p；但是站在 Cpq 做状态时来说：Cpq 做状态时，怎么保障 Cpq 知道 token in p？

在分析上面这个问题前，博主先使用 数学的方式 分析一下 S(Cpq) 到底记录了哪些内容。

⭐ 第一步：定义变量

n：在 p 的状态记录前，p 记录的 p 发往 Cpq 的 msg 数；
n′：在 Cpq 的状态记录前，Cpq 记录的 p 发往 Cpq 的 msg 数；
m：在 q 的状态记录前，q 记录的 q 从 Cpq 中接收到的 msg 数；
m′：在 Cpq 的状态记录前，Cpq 记录的 q 从 Cpq 中接收到的 msg 数；

⭐ 第二步：提出假设

假设 Channel 和 Process 一样，也可以自主的去将做快照时 Channel 中进行网络传输的数据作为状态保存下来；对应到上述案例中就是 Cpq 可以主动的将做好的 S(Cpq) 状态保存下来；

⭐ 第三步：先说结论

Cpq 记录S(Cpq)时，必然会有 n = n' ≥ m = m'；
一个 Channel 要记录的状态是，它 sender 记录自己状态之前 channel 所接收到 sender 发的的 msg 列表，再减去 receiver 记录自己状态之前 channel 已经发给 receiver 的 msg 列表，减去的之后的 msg 就是还在 Channel 中的数据，这些数据是需要 Channel 作为状态记录下来的。
而如果 n′ = m′，那么 Channel c 中要记录的 msg 列表就是 empty 列表。如果 n′ > m′，那么要记录的列表是 (m′ 1),…n′ 号消息对应的 msg 列表。

⭐ 第四步：给出证明

首先：n = n'，利用反证法：如果 n != n'，则会有：

n > n' 时，假设：
- n = 10（p 记录状态前，p 记录 p 发往 Cpq msg 数为 10（msg 编号 1 - 10））；
- n' = 7（Cpq 记录状态前，Cpq 记录 p 发往 Cpq 的 msg 数为 7（msg 编号 1 - 7））；
- 那么假设 token 这条 msg 的编号为 9，就会出现 p 记录的状态为S(p-token-in-Cpq)，Cpq 记录的状态为S(p-token-in-p)，实际这是不符合全局一致性快照的要求的；
n < n' 时，假设：
- n = 7（p 记录状态前，p 记录 p 发往 Cpq msg 数为 7（编号 1 - 7））；
- n' = 10（Cpq 记录状态前，Cpq 记录 p 发往 Cpq 的 msg 数为 10（编号 1 - 10））；
- 那么假设 token 这条 msg 的编号为 9，就会出现 p 记录的状态为S(p-token-in-p)，Cpq 记录的状态为S(p-token-in-Cpq)，实际这是不符合全局一致性快照的要求的；
n = n' 时，假设：
- p 做出S(p-token-in-p)的状态时，因为 n = n'，这就代表 p 没有把 token 发出去，Cpq 也没有接受到 token，Cpq 就知道 token 没有发过来，则只有这种情况可以满足S(Cpq-token-in-p)；

其次：m = m'，同样利用反证法可以得到，下文只举 m > m' 的案例：

m > m' 时：
- n = n' = m = 10（q 记录状态前，Cpq 记录 q 从 Cpq 接收到的 msg 数为 10（编号 1 - 10））；
- m' = 7（Cpq记录状态前，Cpq 记录的 q 从 Cpq 接收到的 msg 数为 7（编号 1 - 7））；
- 那么假设 token 这条 msg 的编号为 9，就会出现 Cpq 记录的状态为S(Cpq-token-in-Cpq)，q 记录的状态为S(q-token-in-p)，实际这是不符合全局一致性快照的要求的；

最后：n' ≥ m' and n ≥ m：同样可以利用反证法得到，此处不再举例。

⭐ 第五步：解答 2.4 节提出的 Cpq 怎么知道 token in p 的问题

通过 n = n' ≥ m = m' 其实就可以推论出 Cpq 一定会知道 token in p。

为了帮大家更容易的理解一个分布式应用包含的全局一致性快照包含的数据内容，接下来我用伪代码描述一下，会比文字更好理解~

6.伪代码描述一个分布式应用全局一致快照包含的数据内容

伪代码如下：

代码语言：javascript复制

// S_all 即一个分布式应用的全局一致性快照
S_all = null;

// 假设总共有 x 个 Process，S_all 先把所有 Process 的状态记录下来
for (int i = 1; i <= x; i  ) {
  // 第 i 个 Process 的状态为 S_P_i，直接按照  = 写，勿喷
  S_all  = S_P_i;
}

// 假设总共有 y 个 Channel，S_all 把所有 Channel 的状态记录下来
for (int i = 1; i <= y; i  ) {
  // 1. S_C_i：第 i 个 Channel 的状态
  // 2. m_i：第 i 个 Channel 做快照前，发往下游 Process 的消息（数据）编号，m_i 其实就是上文变量中的 m
  // 3. n_i：第 i 个 Channel 做快照前，接受上游 Process 的消息（数据）编号，n_i 其实就是上文变量中的 n
  // 4. 需要注意，每一个 Channel 的 m_i 和 n_i 的数值都可能是不一样的
  // 5. Message[m_i   1] ：代表编号为 m_i   1 的那条消息（数据）。举例 Message[0] 代表编号为 0 的那条消息（数据）
  S_C_i = Message[m_i   1]   ...   Message[n_i];
  
  S_all  = S_C_i;
}

// 状态做完了

7.怎样去记录 Channel 的状态？

通过上面的分析，我们已经讨论得到了S(Cpq)都包含了什么内容，并且其之间要满足什么样的数学关系。但是在现实实际生活中，消息在 Channel 上传输（光纤上传输）时，我们是无法记录这些消息作为 Channel 的状态的。

那么有没有什么思路可以让我们也能够去记录 Channel 的消息呢？

当然有。

因为只要我们分布式应用的传输这些消息的光纤没有被挖断，消息终究会通过 Channel 到达 Process 的，因此我们就可以自然的想到其实可以在消息传输 终点的 Process 去记录这些消息作为 Channel 的状态。对应到上述的 Single-Token conservation 案例来说，我们可以在 q 中记录 Channel pq 的S(Cpq)，在 p 中记录 Channel pq 的S(Cqp)。

如果是按照这个思路去分析的话，上面那段伪代码就可以简化为下面这样：

代码语言：javascript复制

// S_all 即全局一致性快照
S_all = null;

// 假设总共有 x 个 Process
for (int i = 1; i <= x; i  ) {
  // S_i_all：第 i 个 Process 要记录的所有状态
  S_i_all = null;
  
  // S_P_i：第 i 个 process 的状态
  S_i_all  = S_P_i; // 【直接按照  = 写，勿喷】
  
  // 第 i 个 Process 总共有 y 个输入 channel，下文中 j 即指代第 i 个 Process 的第 j 个上游 Channel
  for (int j = 1; j <= y; j  ) {
    // 1.S_C_j：第 j 个 Channel 的状态
    // 2.m_j：第 j 个 Channel 做快照前，发往下游 Process 的消息（数据）编号
    // 3.n_j：第 j 个 Channel 做快照前，接受上游 Process 的消息（数据）编号
    S_C_j = Message[m_j   1]   ...   Message[n_j];

    S_i_all  = S_C_j;
  }
  
  S_all  = S_i_all;
}

// 状态做完了

解释一下上面的伪代码：

S_all：所有的进程记录的状态之和，即所有S_i_all之和
S_i_all：每一个进程要记录的所有状态之和
Process i 在做S_i_all其实只有一个变量在做快照时是不知道到的，那就是 n_j（即第 i 个 channel 做快照前，接受到 j（上游）的消息个数）
- S_P_i是 Process 自己的状态，所以是明确已知的
- S_C_j是 Channel 状态，由Message[m_j 1] ... Message[n_j]组成，其中 m_j 的含义是这个 Channel j 在做快照时发给当前 Process i 的消息编号，前文已经介绍到 m = m'，即 m_j 值就等于是是当前 Process i 在做S_P_i时，接收到上游 Channel 的消息编号，所以 m_j 是明确已知的。但是 Process i 在做上游 Channel j 快照时 n_j 是无法获取到的。

那么 Process i 在做S_C_j如何获取到 n_j 呢？

这里我们就可以想到 n = n'：

因为 n = n'，所以 n_j 的值就等于 Channel j 的上游 Process 在做快照时，Process 发往 Channel j 的消息个数；
所以我们可以认为 n_j 是 Channel j 的输入 Process 在做快照时已知的一个值，则只需要这个 Process 把这个值发给 Process i 就行；
Channel j 的输入 Process 告知 Process i 的方式其做完快照之后，直接发一个 marker 下去，这个 marker 不会对计算有任何影响（即不会对状态产生任何影响），marker 只是一个标识，当 Process i 接受到这个 marker 时，就知道 n_j 的具体值了。

大家注意到了没，这个 marker 其实就对应到了 Flink 中 Checkpoint 的 barrier。

8.总结

本文主要讲了一个分布式应用异步做一个全局一致性快照的机制。

好，今天主要就讲这么多，下集我们再说说 Chandy-Lambort，Flink CP 和今天介绍的全局一致性快照原理的关系。

flink 大数据编程算法

0 人点赞