Question

本文档描述了 Flink 的流数据流容错机制。

介绍

Apache Flink 提供容错机制，以持续恢复数据流应用程序的状态。该机制确保即使存在故障，程序的状态最终也只能反映数据流中的每条记录 一次 。请注意，有一个开关 将 保证 降级__至少一次 （如下所述）。

容错机制连续地绘制分布式流数据流的 SNAPSHOT。对于状态较小的流应用程序，这些 SNAPSHOT 非常轻量级，可以经常绘制，而不会对性能产生太大影响。流应用程序的状态存储在可配置的位置（例如主节点或 HDFS）。

如果程序失败（由于机器，网络或软件故障），Flink 将停止分布式流数据流。然后，系统重新启动算子并将其重置为最新的成功检查点。输入流将重置为状态 SNAPSHOT 的点。作为重新启动的并行数据流的一部分处理的任何记录都保证不会成为先前检查点状态的一部分。

注意： 默认情况下，禁用检查点。有关如何启用和配置检查 点 的详细信息，请参阅 检查 点。

注意： 要使此机制实现其完全保证，数据流源（例如消息队列或代理）需要能够将流回滚到定义的最近点。 Apache Kafka 具有这种能力，Flink 与 Kafka 的连接器利用了这种能力。有关 Flink 连接器提供的保证的更多信息，请参阅 数据源和接收器的容错保证 。

注意： 由于 Flink 的检查点是通过分布式 SNAPSHOT 实现的，因此我们可以互换使用 SNAPSHOT 和 检查点 。

检查点

Flink 的容错机制的核心部分是绘制分布式数据流和算子状态的一致 SNAPSHOT。这些 SNAPSHOT 充当一致的检查点，系统可以在发生故障时退回。Flink 用于绘制这些 SNAPSHOT 的机制在“ 分布式数据流的轻量级异步 SNAPSHOT ”中进行了描述。它受到分布式 SNAPSHOT 的标准 Chandy-Lamport 算法的 启发，专门针对 Flink 的执行模型而定制。

障碍

Flink 分布式 SNAPSHOT 的核心数据元是 流障碍 。这些障碍被注入数据流并与记录一起作为数据流的一部分流动。障碍永远不会超过记录，流量严格符合要求。屏障将数据流中的记录分为进入当前 SNAPSHOT 的记录集和进入下一个 SNAPSHOT 的记录。每个屏障都带有 SNAPSHOT 的 ID，该 SNAPSHOT 的记录在其前面推送。障碍不会中断流的流动，因此非常轻。来自不同 SNAPSHOT 的多个障碍可以同时在流中，这意味着可以同时发生各种 SNAPSHOT。

流障碍被注入流源的并行数据流中。注入 SNAPSHOT n 的障碍（我们称之为 S _n ）的点是源流中 SNAPSHOT 覆盖数据的位置。例如，在 Apache Kafka 中，此位置将是分区中最后一条记录的偏移量。该位置 S _n 被报告给 检查点协调员 （Flink 的 JobManager）。

然后障碍物向下游流动。当中间算子从其所有输入流中收到 SNAPSHOT n 的屏障时，它会为其所有输出流中的 SNAPSHOT n 发出屏障。一旦接收器算子（流式 DAG 的末端）从其所有输入流接收到障碍 n ，它就向 SNAPSHOT n 确认检查点协调器。在所有接收器确认 SNAPSHOT 之后，它被认为已完成。

一旦完成 SNAPSHOT n ，作业将永远不再向源请求来自 S _n 之前的记录，因为此时这些记录（及其后代记录）将通过整个数据流拓扑。

接收多个输入流的 算子需要在 SNAPSHOT 屏障上 对齐 输入流。上图说明了这一点：

• 一旦算子从输入流接收到 SNAPSHOT 屏障 n ，它就不能处理来自该流的任何其他记录，直到它从其他输入接收到屏障 n 为止。否则，它会混合属于 SNAPSHOT n 的 记录和属于 SNAPSHOT n + 1 的记录 。
• 报告障碍 n 的 流暂时被搁置。从这些流接收的记录不会被处理，而是放入输入缓冲区。
• 一旦最后一个流接收到屏障 n ，算子就会发出所有挂起的传出记录，然后自己发出 SNAPSHOT n 个 屏障。
• 之后，它恢复处理来自所有输入流的记录，在处理来自流的记录之前处理来自输入缓冲区的记录。

状态

当 算子包含任何形式的 状态时 ，此状态也必须是 SNAPSHOT 的一部分。算子状态有不同的形式：

• 用户定义的状态 ：这是由转换函数（如 map() 或 filter() ）直接创建和修改的状态。有关详细信息，请参阅 Streaming Applications 中的 State 。
• 系统状态 ：此状态是指作为 算子计算一部分的数据缓冲区。此状态的典型示例是 窗口缓冲区 ，系统在其中收集（和聚合）窗口记录，直到窗口被评估和逐出。

算子在他们从输入流接收到所有 SNAPSHOT 障碍时，以及在向其输出流发出障碍之前对其状态进行 SNAPSHOT。此时，将根据障碍之前的记录对状态进行所有更新，并且在应用障碍之后不依赖于记录的更新。由于 SNAPSHOT 的状态可能很大，因此它存储在可配置的 状态后台中 。默认情况下，这是 JobManager 的内存，但对于生产使用，应配置分布式可靠存储（例如 HDFS）。在存储状态之后，算子确认检查点，将 SNAPSHOT 屏障发送到输出流中，然后继续。

生成的 SNAPSHOT 现在包含：

• 对于每个并行流数据源，启动 SNAPSHOT 时流中的偏移/位置
• 对于每个 算子，指向作为 SNAPSHOT 的一部分存储的状态的指针

完全一次与至少一次

对齐步骤可以增加流式传输程序的等待时间。通常，这种额外的延迟大约为几毫秒，但我们已经看到一些异常值的延迟显着增加的情况。对于要求所有记录始终具有超低延迟（几毫秒）的应用程序，Flink 可以在检查点期间跳过流对齐。一旦算子看到每个输入的检查点障碍，仍然会绘制检查点 SNAPSHOT。

当跳过对齐时，即使在检查点 n 的 某些检查点障碍到达之后，算子仍继续处理所有输入。这样，算子还可以在获取检查点 n 的状态 SNAPSHOT 之前处理属于检查点 n + 1 的 数据元。在还原时，这些记录将作为重复记录出现，因为它们都包含在检查点 n 的状态 SNAPSHOT 中，并将在检查点 n 之后作为数据的一部分进行重放。

注意 ：对齐仅适用于具有多个前驱（连接）的 算子以及具有多个发送方的 算子（在流重新分区/随机播放之后）。正因为如此，数据流只有尴尬的并行流 算子操作（ map() ， flatMap() ， filter() ，...）实际上给 正好一次 保证了即使在 至少一次 模式。

异步状态 SNAPSHOT

注意，上述机制意味着算子在将 状态 的 SNAPSHOT 存储在 状态后台时 停止处理输入记录。每次拍摄 SNAPSHOT 时，此 同步 状态 SNAPSHOT 都会引入延迟。

可以让算子在存储状态 SNAPSHOT 时继续处理，有效地让状态 SNAPSHOT 在后台 异步 发生。为此，算子必须能够生成一个状态对象，该状态对象应以某种方式存储，以便对算子状态的进一步修改不会影响该状态对象。例如，诸如 RocksDB 中使用 的写时复制 数据结构具有这种行为。

在接收到输入的检查点障碍后，算子启动其状态的异步 SNAPSHOT 复制。它立即释放其输出的障碍，并继续进行常规流处理。后台复制过程完成后，它会向检查点协调员（JobManager）确认检查点。检查点现在仅在所有接收器都已收到障碍并且所有有状态算子已确认其完成备份（可能在障碍物到达接收器之后）之后才完成。

有关状态 SNAPSHOT 的详细信息，请参阅 状态后台 。

复苏

在这种机制下的恢复是直截了当的：当失败时，Flink 选择最新完成的检查点 k 。然后，系统重新部署整个分布式数据流，并为每个算子提供作为检查点 k 的 一部分进行 SNAPSHOT 的状态。设置源以开始从位置 S _k 读取流。例如，在 Apache Kafka 中，这意味着告诉消费者从偏移量 S _k 开始提取。

如果状态以递增方式 SNAPSHOT，则算子从最新完整 SNAPSHOT 的状态开始，然后对该状态应用一系列增量 SNAPSHOT 更新。

有关更多信息，请参阅 重启策略

算子 SNAPSHOT 实施

在执行算子 SNAPSHOT 时，有两部分： 同步 和 异步 部分。

算子和状态后台将其 SNAPSHOT 作为 Java 提供 FutureTask 。该任务包含完成 同步 部分且 异步 部分处于挂起状态的状态。然后，异步部分由该检查点的后台线程执行。

检查点纯粹同步返回已经完成的算子 FutureTask 。如果需要执行异步 算子操作，则以该 run() 方法执行 FutureTask 。

任务是可取消的，因此可以释放流和其他资源消耗句柄。