前言

Flink 是通过状态快照实现容错处理

一、State Backends

由 Flink 管理的 keyed state 是一种分片的键/值存储，每个 keyed state 的工作副本都保存在负责该键的 taskmanager 本地中。

• 一种基于 RocksDB 内嵌 key/value 存储将其工作状态保存在磁盘上；
• 一种基于堆的 state backend，将其工作状态保存在 Java 的堆内存。

针对第二种，又细化出如下两种类型

• FsStateBackend，将其状态快照持久化到分布式文件系统；
• MemoryStateBackend，它使用 JobManager 的堆保存状态快照。

具体参考如下表格

名称	Working State	状态备份	快照
RocksDBStateBackend	本地磁盘（tmp dir）	分布式文件系统	全量 / 增量
支持大于内存大小的状态 经验法则：比基于堆的后端慢10倍
FsStateBackend	JVM Heap	分布式文件系统	全量
快速，需要大的堆内存 受限制于 GC
MemoryStateBackend	JVM Heap	JobManager JVM Heap	全量
适用于小状态（本地）的测试和实验

二、Checkpoint Storage

作用

Flink 用来定期对每个算子的所有状态进行持久化快照，并将快照复制到更持久的地方。

分类

• 一种持久保存其状态快照 到一个分布式文件系统；
• 另一种是使用 JobManager 的堆。

名称	状态备份
FileSystemCheckpointStorage	分布式文件系统
支持非常大的状态大小 高度可靠 推荐用于生产部署
JobManagerCheckpointStorage	JobManager JVM Heap
适合小状态（本地）的测试和实验

 三、状态快照

定义

• 快照 – 是 Flink 作业状态全局一致镜像的通用术语。快照包括指向每个数据源的指针（例如，到文件或 Kafka 分区的偏移量）以及每个作业的有状态运算符的状态副本，该状态副本是处理了 sources 偏移位置之前所有的事件后而生成的状态。

• Checkpoint – 一种由 Flink 自动执行的快照，其目的是能够从故障中恢复。Checkpoints 可以是增量的，并为快速恢复进行了优化。

• 外部化的 Checkpoint – 通常 checkpoints 不会被用户操纵。Flink 只保留作业运行时的最近的 n 个 checkpoints（n 可配置），并在作业取消时删除它们。但你可以将它们配置为保留，在这种情况下，你可以手动从中恢复。

• Savepoint – 用户出于某种操作目的（例如有状态的重新部署/升级/缩放操作）手动（或 API 调用）触发的快照。Savepoints 始终是完整的，并且已针对操作灵活性进行了优化。

 状态快照的工作原理

当 checkpoint coordinator（job manager 的一部分）指示 task manager 开始 checkpoint 时，它会让所有 sources 记录它们的偏移量，并将编号的 checkpoint barriers 插入到它们的流中。这些 barriers 流经 job graph，标注每个 checkpoint 前后的流部分。

Checkpoint n 将包含每个 operator 的 state，这些 state 是对应的 operator 消费了严格在 checkpoint barrier n 之前的所有事件，并且不包含在此（checkpoint barrier n）后的任何事件后而生成的状态。

当 job graph 中的每个 operator 接收到 barriers 时，它就会记录下其状态。拥有两个输入流的 Operators（例如 CoProcessFunction）会执行 barrier 对齐（barrier alignment） 以便当前快照能够包含消费两个输入流 barrier 之前（但不超过）的所有 events 而产生的状态。

Flink 的 state backends 利用写时复制（copy-on-write）机制允许当异步生成旧版本的状态快照时，能够不受影响地继续流处理。只有当快照被持久保存后，这些旧版本的状态才会被当做垃圾回收。