Garden | Flink Fault Tolerance

Flink 基于 Checkpoint 机制实现容错，它的原理是不断地生成分布式 Streaming 数据流 Snapshot。

在流处理失败时，通过这些 Snapshot 可以恢复数据流处理。

Barrier

理解 Flink 的容错机制，首先需要了解一下 Barrier 这个概念：

Stream Barrier 是 Flink 分布式 Snapshotting 中的核心元素，它会作为数据流的记录被同等看待，被插入到数据流中，将数据流中记录的进行分组，并沿着数据流的方向向前推进。
每个 Barrier 会携带一个 Snapshot ID，属于该 Snapshot 的记录会被推向该 Barrier 的前方。因为 Barrier 非常轻量，所以并不会中断数据流。带有 Barrier 的数据流。

如下图所示：

基于上图，我们通过如下要点来说明：

出现一个 Barrier，在该 Barrier 之前出现的记录都属于该 Barrier 对应的 Snapshot，在该 Barrier 之后出现的记录属于下一个 Snapshot。
来自不同 Snapshot 多个 Barrier 可能同时出现在数据流中，也就是说同一个时刻可能并发生成多个 Snapshot。
当一个中间（Intermediate）Operator 接收到一个 Barrier 后，它会发送 Barrier 到属于该 Barrier 的 Snapshot 的数据流中，等到 Sink Operator 接收到该 Barrier 后会向 Checkpoint Coordinator 确认该 Snapshot。
直到所有的 Sink Operator 都确认了该 Snapshot，才被认为完成了该 Snapshot。

这里还需要强调的是，Snapshot 并不仅仅是对数据流做了一个状态的 Checkpoint，它也包含了一个 Operator 内部所持有的状态，这样才能够在保证在流处理系统失败时能够正确地恢复数据流处理。

也就是说，如果一个 Operator 包含任何形式的状态，这种状态必须是 Snapshot 的一部分。

Operator 的状态包含两种：

一种是系统状态，一个 Operator 进行计算处理的时候需要对数据进行缓冲，所以数据缓冲区的状态是与 Operator 相关联的，以窗口操作的缓冲区为例，Flink 系统会收集或聚合记录数据并放到缓冲区中，直到该缓冲区中的数据被处理完成；
另一种是用户自定义状态（状态可以通过转换函数进行创建和修改），它可以是函数中的 Java 对象这样的简单变量，也可以是与函数相关的 Key/Value 状态。

对于具有轻微状态的 Streaming 应用，会生成非常轻量的 Snapshot 而且非常频繁，但并不会影响数据流处理性能。

Streaming 应用的状态会被存储到一个可配置的存储系统中，例如 HDFS。

在一个 Checkpoint 执行过程中，存储的状态信息及其交互过程，如下图所示：

在 Checkpoint 过程中，还有一个比较重要的操作——Stream Aligning。

当 Operator 接收到多个输入的数据流时，需要在 Snapshot Barrier 中对数据流进行排列对齐，如下图所示：

具体排列过程如下：

Operator 从一个 incoming Stream 接收到 Snapshot Barrier n，然后暂停处理，直到其它的 incoming Stream 的 Barrier n（否则属于 2 个 Snapshot 的记录就混在一起了）到达该 Operator。
接收到 Barrier n 的 Stream 被临时搁置，来自这些 Stream 的记录不会被处理，而是被放在一个 Buffer 中
一旦最后一个 Stream 接收到 Barrier n，Operator 会 emit 所有暂存在 Buffer 中的记录，然后向 Checkpoint Coordinator 发送 Snapshot n
继续处理来自多个 Stream 的记录

基于 Stream Aligning 操作能够实现 Exactly Once 语义，但是也会给流处理应用带来延迟，因为为了排列对齐 Barrier，会暂时缓存一部分 Stream 的记录到 Buffer 中。

尤其是在数据流并行度很高的场景下可能更加明显，通常以最迟对齐 Barrier 的一个 Stream 为处理 Buffer 中缓存记录的时刻点。

在 Flink 中，提供了一个开关，选择是否使用 Stream Aligning，如果关掉则 Exactly Once 会变成 At least once。