State Snapshot

相关概念

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

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

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

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

状态快照如何工作

Flink 使用 Chandy-Lamport algorithm 算法的一种变体，称为异步 barrier snapshotting。

当 checkpoint coordinator（JobManager 的一部分）指示 TaskManager 开始 checkpoint 时，会让所有 Source 记录自身的偏移量，并将编号的 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 之前（但不超过）的所有事件而产生的状态。

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

确保精确一次（Exactly once）

当流处理应用程序发生错误的时候，可能会产生丢失或者重复的结果。Flink 根据应用程序和集群的配置，可以产生以下结果：

• 不会从快照中进行恢复（At most once）

• 没有任何丢失，但是可能会得到重复的结果（At least once）

• 没有丢失或重复的结果（Exactly once）

Flink 通过回退和重新发送 Source 数据流从故障中恢复，当理想情况被描述为 Exactly Once 时，这并不意味着每个事件都将被精确一次处理。相反，这意味着每一个事件都会影响 Flink 管理的状态精确一次。

Barrier 只有在需要提供精确一次的语义保证时需要进行对齐（Barrier alignment）。如果不需要这种语义，可以通过配置 CheckpointingMode.AT_LEAST_ONCE 关闭 Barrier 对齐来提高性能。

端到端精确一次

为了实现端到端的精确一次，使 Source 中的每个事件都仅精确一次对 Sink 生效，必须满足以下条件：

1. Source 必须是可重放的

2. Sink 必须是事务性的（或幂等的）

Source 和 Sink 的容错保证

当程序出现错误的时候，Flink 的容错机制能恢复并继续运行程序。

只有当 Source 参与了快照机制的时候，Flink 才能保证对自定义状态的精确一次更新。下表列举了 Flink 与其自带 Source 连接器的状态更新的保证。

为了保证端到端精确一次的数据交付（在精确一次的状态语义上更进一步），Sink 需要参与 checkpointing 机制。下表列举了 Flink 与其自带 Sink 的更新保证（假设精确一次状态更新）。

[1]  https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/learn-flink/fault_tolerance/