Flink 原理与实现：Checkpoint

约定不等于承诺〃 2023-06-16 14:58 3阅读 0赞

扫码关注公众号免费阅读全文：**冰山烈焰的黑板报**  
![在这里插入图片描述][20200906232656299.jpg_pic_center]

众所周知，Flink 采用 Asynchronous Barrier Snapshotting（简称 ABS）算法实现分布式快照的。但是，本文着重介绍 Flink Checkpoint 工作过程，并且用图形化方式描述 Checkpoint 在 Flink 中的实现，Failure Recovery Mechanism（失败恢复机制），以及 Performance of Checkpointing。

# 1. Consistent Checkpoint #

## 1.1 Naive Checkpoint ##

有状态的流式应用程序的 Consistent Checkpoint 是指，在某一时刻每个算子在所有的算子已经处理完相同的输入时的 State 副本（建议先阅读[Flink 原理与实现：State][Flink _State]）。Naive Consistent Checkpoint 可以描述成以下步骤：

1.  暂停所有输入流的接收；
2.  等待正在处理的数据完成计算，换言之，所有算子必须已经处理完它们所有的输入数据；
3.  拷贝每个算子的 State 到远程，并持久化存储之。当所有算子都已经完成了各自的拷贝，那么这次的 Checkpoint 也就完成了；
4.  继续接收输入流。

但是，Flink 并没有直接采用 Naive Consistent Checkpoint，我将在后文介绍 Flink 采用的更精致的 Checkpoint 算法。图1 展示了 Naive Consistent Checkpoint 的过程。  
![图1 流式应用程序的 Naive Consistent Checkpoint][1 _ Naive Consistent Checkpoint]  
上图中有一个输入流持续生产自增的正整数，即 1、2、3……Source 将这些数字按照奇偶分到两个分区中，并将输入流的当前 Offset 作为 State 存储起来。然后，由 Sum 算子对接收到的数字进行求和运算。在图1 中，做 Checkpoint 时，Source 中的 Offset 是 5，奇偶 Sum 算子中的和分别是 9、6。

## 1.2 Recovery from Consistent Checkpoint ##

在流式应用程序执行的过程中，Flink 会周期性地 Checkpoint 应用程序的 State。为了从失败中恢复，Flink 会从最近 Checkpoint 的 State 中重新启动处理数据。这种 Recovery Mechanism 可以分为 3 个步骤：

1.  重启整个应用程序；
2.  把所有状态算子的 State 重置为最近一次的 Checkpoint State；
3.  继续所有算子的处理。

这种 Failure Recovery Mechanism 可以用图2 简单的表示。  
![图2 Recovering from a checkpoint][2 Recovering from a checkpoint]  
通过对所有算子做 Checkpoint，存储所有算子的 State，以及所有输入流可以把被消费的位置重置到 Checkpoint 的 State 的地方，这种 Checkpoint 和 Failure Recovery Mechanism 方法可以保证 ***Extractly-Once*** 语义。由此观之，一个流式应用程序能否支持 ***Extractly-Once*** ，取决于输入流是否满足以下特性：

1.  输入流提供 Offset；
2.  输入流可以重放，以便重置到之前的 Offset。

比如，订阅 Apache Kafka 消息的流式应用程序可以支持 ***Extractly-Once***，相反，消费 Socket 的流式应用程序却做不到。**需要注意的一点是**，当流式应用程序从 Checkpoint 的 State 重启之后，它也会处理该 Checkpoint 之后和本次失败之前的数据。换句话说，系统会对一些数据处理两次，即便如此，这种机制仍然可以保证 State 一致性，或者说结果的正确性，因为所有算子的 State 会被重置至它们还没处理到这些数据的位置（关于 ***Extractly-Once*** 语义的真正含义，建议阅读[流计算中的 Exactly Once 语义][Exactly Once]）。

另一个值得一提的事情是，Flink 的 Checkpoint 和 Failure Recovery Mechanism 仅仅只能保证内部 State 的 ***Extractly-Once*** 。对于有 Sink 算子的应用程序，在 Recovery 的过程中，有些数据会被多次 Sink 到外部存储系统，比如文件系统，数据库系统等。对于 Flink 的 End-to-End Extractly-Once（端到端的 Extractly-Once），我后续会有文章专门讨论（请关注笔者 (♥◠‿◠♥)）。

# 2. Flink Checkpoint Algorithm #

前面讨论过 Naive Consistent Checkpoint 需要流式应用程序暂停、Checkpoint、重启，这就引入了 “Stop-the-World（STW）”。由于 STW，这对于中等延迟要求的应用程序而言都不实际。虽然 Flink 的 Checkpoint 也是基于 Consistent Checkpoint，不过幸运的是，Flink 实现 Checkpoint 是采用了 Chandy-Lamport 算法的改进版 Asynchronous Barrier Snapshotting 算法。该算法把 Checkpoint 从算子处理中解耦，在不停止整个应用程序的情况下，某些算子可以继续处理数据，而其他算子则持久化它们的 State。下文就开始介绍这个算法的工作过程了。

Flink Checkpoint 算法使用了一个特殊的数据结构——Checkpoint Barrier。Checkpoint Barrier 被 Source 算子注入到数据流中，但是不会影响数据的正常处理。Checkpoint Barrier 用 Checkpoint ID 标志它所属的 Checkpoint，并在逻辑上将流分为两部分。**Checkpoint Barrier 之前的数据的所有 State 修改，包含在这个 Barrier 对应的 Checkpoint 中；这个 Checkpoint Barrier 之后的数据的所有 State 修改，包含在后续的 Checkpoint 中**。例如，在图3 中 Checkpoint Barrier N 前面红色的数据和 Checkpoint Barrier N 的颜色相同，表示这些数据都是由 Checkpoint Barrier N 应该负责，而 Checkpoint Barrier N 后面黄色的数据就不属于 Checkpoint Barrier N。  
![图3 Checkpoint Barrier][3 Checkpoint Barrier]  
明白 Checkpoint Barrier 之后，现在我用下面这个例子一步一步解释这个过程：假设有两个 Source 算子分别消费两个生产自增正整数的输入流；然后，Source 的结果分区到奇偶两个 Sum 算子，Sum 算子会对接收到的正整数进行求和运算；最后，算子的结果会下发到 Sink 算子。这个过程可以简单的表示成图4。  
![图4 2 个有状态的 Source，2 个有状态的算子，2 个无状态的 Sink][4 2 _ Source_2 _2 _ Sink]  
JobManager 发送一条带有 Checkpoint ID 的 Barrier 到每个 Source 算子，进行 Checkpoint 的初始化，如图5 所示。  
![图5 JobManager 初始化 Checkpoint，并发往所有的 Source][5 JobManager _ Checkpoint_ Source]  
当 Source 算子接收到 Barrier 时，它就暂停提交数据，同时 Checkpoint 本地 State 到 State Backend，并且广播携带 Checkpoint ID 的这个 Checkpoint Barrier 到所有下游分区。一旦算子 的 State Checkpoint 完成，并且算子确认在 JobManager 中有这个 Checkpoint 的信息，State Backend 就会通知该算子。当所有的 Barrier 发出后，Source 算子继续照常执行。通过注入 Barrier，Source 算子知道 Checkpoint 发生在流中的哪个位置。图6 展示这一过程。  
![图6 Source checkpoint 各自的 state，并且提交 Checkpoint Barrier][6 Source checkpoint _ state_ Checkpoint Barrier]  
Checkpoint Barrier 广播到所有连接的并行算子，以确保每个算子可以从输入分区中接收 Barrier。当一个算子接收到一次新的 Checkpoint 对应的 Barrier 时，它会等待所有输入分区中属于这次 Checkpoint 的 Barrier 到达。然而在等待的过程中，它会继续处理来自尚未提供 Barrier 的流中的数据。在算子的其中一个输入分区的 Barrier 之后到达，且属于该 Barrier 的已到达数据不能被处理并被缓存，以等待其他输入分区的 Barrier 到达。等待所有 Barrier 的过程叫作 ***Barrier Alignment***，它可以用图7 简单的表示出来。  
![图7 Task 等待接收每个输入分区的 Barrier；缓存输入流中已到达的 Barrier 对应的数据；正常处理其他所有数据][7 Task _ Barrier_ Barrier]  
如果算子接收到它的输入分区中的 Barrier，它将在 State Backend 中初始化一个 Checkpoint，并且广播该 Checkpoint 的 Barrier 到它的所有下游，如图7 所示。  
![图8 一旦所有的 Barrier 都到达时，Task checkpoint 它们各自的状态，并且将 Checkpoint Barrier 继续向下游发送][8 _ Barrier _Task checkpoint _ Checkpoint Barrier]  
一旦所有的 Barrier 都已提交，算子就开始处理缓存的数据。当缓存中的所有数据也提交之后，算子继续处理输入流。图9 表示了这个过程。  
![图9 在 Checkpoint Barrier 下发之后，算子继续照常处理数据][9 _ Checkpoint Barrier]  
当 Sink 算子接收到一个 Barrier 的时候，它也会像其他算子那样——进行 Barrier Alignment，Checkpoint 它自己的 State，确认 JobManager 已收到该 Barrier。当 JobManager 收到应用程序的所有算子的 Checkpoint 的确认，就标志着这个应用程序的 Checkpoint 过程已完成。图10 描述了这最后一步。完成的 Checkpoint 也可像前文描述的那样用于失败恢复。  
![图10 Sink 收到 Checkpoint Barrier 复信告知 JobManager，并且当所有的 Task 都已确认各自的 State checkpoint 成功，则该 Checkpoint 完成][10 Sink _ Checkpoint Barrier _ JobManager_ Task _ State checkpoint _ Checkpoint]  
为了简单起见，还可以采用填表的方法描述 Flink Checkpoint 过程。还用上面的例子，当黄蓝色这个 Job 的 所有算子在表中填满之后，表示本次 Checkpoint 完成，如图11 所示。  
![图11 Checkpoint 填表法][11 Checkpoint]

# 3. Performance of Checkpointing #

Flink Checkpoint 算法可以避免 STW，但是，为了进一步提高流式应用程序的处理延迟，Flink 做了以下这些调整，可以在某些情况下提升性能：

1.  Extractly-Once with Incremental Checkpointing。
2.  At-Least-Once

在 Extractly-Once 的场景中，任务 Checkpoint 它的 State 时，需要缓存数据以便 Barrier Alignment。由于 State 会变得比较大，Checkpoint 通过网络写数据到远程存储时，可能会花费数秒钟到数分钟不等。在 Flink 的设计中，State Backend 负责存储 Checkpoint。所以，具体怎样拷贝任务的 State 取决于 State Backend 的实现。比如，文件系统和 RocksDB 作为 Stat Backend 时，都支持异步 Checkpoint。当 Checkpoint 触发时，State Backend 会创建本地的 State 拷贝。本地拷贝创建完之后，任务继续照常运行。后台线程会异步将本地拷贝快照到远程存储，并且当 Checkpoint 完成时，通知任务。异步 Checkpoint 大大减少了任务继续处理数据前的时间。另外，RocksDB State Backend 的 Incremental Checkpointing 特性，可以降低数据传输时间。

另一个降低 Checkpoint 延迟的方法通过对 **Barrier Alignment** 这个环节的调整。对于某些实时性要求高的流式应用程序，Flink 可以配置成在 **Barrier Alignment** 期间处理所有到达的数据，而不是缓存它们。一旦一个 Checkpoint 的所有 Barrier 都到达，该算子就 Checkpoint 它的 State，这也包括属于下一个 Checkpoint 的数据。这种情况下，当失败恢复时，这些数据会被再次处理，此时 Flink 提供的就是 At-Least-Once 保证。

# 4. 总结 #

本文介绍了 Naive Consistent Checkpoint、Flink Checkpoint Algorithm、Failure Recovery Mechanism，和如何调整 Flink Checkpoint 的性能。

[20200906232656299.jpg_pic_center]: https://img-blog.csdnimg.cn/20200906232656299.jpg#pic_center
[Flink _State]: https://gatsbynewton.blog.csdn.net/article/details/103266395
[1 _ Naive Consistent Checkpoint]: https://img-blog.csdnimg.cn/20191206231325497.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[2 Recovering from a checkpoint]: https://img-blog.csdnimg.cn/20191206232912308.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[Exactly Once]: https://gatsbynewton.blog.csdn.net/article/details/95724945
[3 Checkpoint Barrier]: https://img-blog.csdnimg.cn/20191207163734678.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[4 2 _ Source_2 _2 _ Sink]: https://img-blog.csdnimg.cn/201912071437372.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[5 JobManager _ Checkpoint_ Source]: https://img-blog.csdnimg.cn/20191208085140890.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[6 Source checkpoint _ state_ Checkpoint Barrier]: https://img-blog.csdnimg.cn/20191208085211321.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[7 Task _ Barrier_ Barrier]: https://img-blog.csdnimg.cn/20191208085242751.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[8 _ Barrier _Task checkpoint _ Checkpoint Barrier]: https://img-blog.csdnimg.cn/20191208085754496.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[9 _ Checkpoint Barrier]: https://img-blog.csdnimg.cn/20191208085826922.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[10 Sink _ Checkpoint Barrier _ JobManager_ Task _ State checkpoint _ Checkpoint]: https://img-blog.csdnimg.cn/20191208085851947.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[11 Checkpoint]: https://img-blog.csdnimg.cn/20191207212652234.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70