FLINK基础149:DS状态机制(14) CheckPoint机制二Checkpoint 执行机制详解

在介绍 Checkpoint 的执行机制前，我们需要了解一下 state 的存储，因为state 是 Checkpoint 进行持久化备份的主要角色。

1 Statebackend 的分类

下图阐释了目前 Flink 内置的三类 state backend，其中 MemoryStateBackend和 FsstateBackend 在运行时都是存储在 java heap 中的，只有在执行 Checkpoint 时，FsstateBackend 才 会 将 数 据 以 文 件 格 式 持 久 化 到 远 程 存 储 上。 而RocksDBStateBackend 则借用了 RocksDB（内存磁盘混合的 LSM DB）对 state进行存储。

对于 HeapKeyedStateBackend，有两种实现： ●支持异步 Checkpoint（默认）：存储格式 copyOnWriteStateMap ●仅支持同步 Checkpoint：存储格式 nestedStateMap 特别在 MemoryStateBackend 内使用 HeapKeyedStateBackend 时，Checkpoint 序列化数据阶段默认有最大 5 MB 数据的限制 对于 RocksDBKeyedStateBackend，每个 state 都存储在一个单独的 column family 内，其中 keyGroup，Key 和 Namespace 进行序列化存储在 DB 作为 key。

2 Checkpoint 执行机制详解

　　本小节将对 Checkpoint 的执行流程逐步拆解进行讲解，下图左侧是 Checkpoint Coordinator，是整个 Checkpoint 的发起者，中间是由两个 source，一个sink 组成的 Flink 作业，最右侧的是持久化存储，在大部分用户场景中对应 HDFS。 1.第一步，Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint；。

2.第二步，source 节点向下游广播 barrier，这个 barrier 就是实现 Chandy-Lamport 分布式快照算法的核心，下游的 task 只有收到所有 input 的 barrier 才会执行相应的 Checkpoint。

3.第三步，当 task 完成 state 备份后，会将备份数据的地址（state handle）通知给 Checkpoint coordinator。

4.第四步，下游的 sink 节点收集齐上游两个 input 的 barrier 之后，会执行本地快照，这里特地展示了 RocksDB incremental Checkpoint 的流程，首先 RocksDB 会全量刷数据到磁盘上（红色大三角表示），然后 Flink 框架会从中选择没有上传的文件进行持久化备份（紫色小三角）。

5. 同样的，sink 节点在完成自己的 Checkpoint 之后，会将 state handle 返回通知 Coordinator。

 

6. 最后，当 Checkpoint coordinator 收集齐所有 task 的 state handle，就认为这一次的 Checkpoint 全局完成了，向持久化存储中再备份一个 Checkpoint Meta 文件。

 

3 Checkpoint 的 EXACTLY_ONCE 语义

　　为了实现 EXACTLY ONCE 语义，Flink 通过一个 input buffer 将在对齐阶段收到的数据缓存起来，等对齐完成之后再进行处理。而对于 AT LEAST ONCE 语义，无需缓存收集到的数据，会对后续直接处理，所以导致 restore 时，数据可能会被多次处理。下图是官网文档里面就 Checkpoint align 的示意图：

需要特别注意的是，Flink 的 Checkpoint 机制只能保证 Flink 的计算过程可以做到 EXACTLY ONCE，端到端的 EXACTLY ONCE 需要 source 和 sink 支持。

4 Savepoint 与 Checkpoint 的区别

作业恢复时，二者均可以使用，主要区别如下：