【Flink 精选】阐述 Flink 的容错机制，剖析 Checkpoint 实现流程

Posted 2023-03-08

参考技术A

Flink 容错机制主要是 状态的保存和恢复，涉及 state backends 状态后端、checkpoint 和 savepoint，还有 Job 和 Task 的错误恢复 。

Flink 状态后端是指 保存 Checkpoint 数据的容器 ，其分类有 MemoryStateBackend、FsStateBackend、RocksDBStateBackend ，状态的分类有 operator state 和 keyed state 。

Flink 状态保存和恢复主要依靠 Checkpoint 机制和 Savepoint 机制，两者的区别如下表所示。

快照的概念来源于相片，指照相馆的一种冲洗过程短的照片。在计算机领域， 快照是数据存储的某一时刻的状态记录 。 Flink Snapshot 快照是指作业状态的全局一致记录 。一个完整的快照是包括 source 算子的状态（例如，消费 kafka partition 的 offset）、状态算子的缓存数据和 sink 算子的状态（批量缓存数据、事务数据等）。

Checkpoint 检查点可以自动产生快照，用于Flink 故障恢复 。Checkpoint 具有分布式、异步、增量的特点。

Savepoint 保存点是用户手动触发的，保存全量的作业状态数据 。一般使用场景是作业的升级、作业的并发度缩放、迁移集群等。

Flink 是采用轻量级的分布式异步快照，其实现是采用栅栏 barrier 作为 checkpoint 的传递信号，与业务数据一样无差别地传递下去 ，目的是使得数据流被切分成微批，进行 checkpoint 保存为 snapshot。当 barrier 经过流图节点的时候，Flink 进行 checkpoint 保存状态数据。
如下图所示，checkpoint n 包含每个算子的状态，该状态是指checkpoint n 之前的全部事件，而不包含它之后的所有事件。

针对用户作业出现故障而导致结果丢失或者重复的问题，Flink 提供3种语义：
① At-Least-Once 最少一次 ：不会丢失数据，但可能会有重复结果。
② Exactly-Once 精确一次 ：checkpoint barrier 对齐机制可以保障精确一次。

① FailureRateRestartStrategy ：允许在指定时间间隔内的最大失败次数，同时可以设置重启延时时间。
② FixedDelayRestartStrategy ：允许指定的失败次数，同时可以设置重启延时时间。
③ NoRestartStrategy ：不需要重启，即 Job 直接失败。
④ ThrowingRestartStrategy ：不需要重启，直接抛异常。
Job Restart 策略可以通过 env 设置。

上述策略的父类接口是RestartStrategy，其关键是restart（重启操作）。

① RestartAllStrategy ：重启全部 task，默认策略。
② RestartIndividualStrategy ：恢复单个 task。如果该 task 没有source，可能导致数据丢失。
③ NoOpFailoverStrategy ：不恢复 task。
上述策略的父类接口是FailoverStrategy，其关键是Factory的create（创建 strategy）、onTaskFailure（处理错误）。

如何产生可靠的全局一致性快照是分布式系统的难点，其传统方案是使用的全局时钟，但存在单点故障、数据不一致等可靠性问题 。为了解决该问题， Chandy-Lamport 算法采用 marker 的传播来代替全局时钟 。

① 进程 Pi 记录自己的进程状态，同时生产一个标识信息 marker（与正常 message 不同），通过 ouput channel 发送给系统里面的其他进程。
② 进程 Pi 开始记录所有 input channel 接收到的 message

进程 Pj 从 input channel Ckj 接收到 marker。如果 Pj 还没有记录自己的进程状态，则 Pj 记录自己的进程状态，向 output channel 发送 marker；否则 Pj 正在记录自己的进程状态（该 marker 之前的 message）。

所有的进程都收到 marker 信息并且记录下自己的状态和 channel 的状态（包含的 message）。

Flink 的分布式异步快照实现了Chandy Lamport 算法，其核心思想是 在 source 插入 barrier 代替 Chandy-Lamport 算法中的 marker，通过控制 barrier 的同步来实现 snapshot 的备份和 Exactly-Once 语义 。

Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint。

source task向下游广播barrier。

当source task备份完自己的状态后，会将备份数据的地址（state handle）通知 Checkpoint Coordinator。

map和sink task收集齐上游source的barrier n，执行本地快照。下面例子是RocksDB增量Checkpoint 的流程：首先RocksDB会全量保存到磁盘上（红色大三角表示），然后Flink会从中选择没有上传的文件进行持久化备份（紫色小三角）。

map和sink task在完成Checkpoint 之后，将状态地址state handle返回通知 Coordinator。

当Checkpoint Coordinator收到全部task的state handle，就确定该Checkpoint已完成，并向持久化存储中备份一个Checkpoint Meta（元数据，包括该checkpoint状态数据的备份地址）。