Flink核心概念之有状态的流式处理

什么是状态

虽然数据流中的许多操作一次只查看一个单独的事件（例如事件解析器），但有些操作会记住跨多个事件的信息（例如窗口操作符）。这些操作称为有状态的。

有状态操作的一些示例：

当应用程序搜索某些事件模式时，状态将存储迄今为止遇到的事件序列。
在每分钟/小时/天聚合事件时，状态会保存待处理的聚合。
在数据点流上训练机器学习模型时，状态会保存模型参数的当前版本。
当需要管理历史数据时，状态允许有效访问过去发生的事件。

Flink 需要了解状态，以便使用检查点和保存点使其容错。

有关状态的知识还允许重新缩放 Flink 应用程序，这意味着 Flink 负责在并行实例之间重新分配状态。

可查询状态允许您在运行时从 Flink 外部访问状态。

在使用状态时，阅读 Flink 的状态后端可能也很有用。 Flink 提供了不同的状态后端来指定状态的存储方式和位置。

Keyed State

Keyed State存储在键值存储后端的。状态与有状态操作符读取的流一起严格分区和分布。因此，只能在keyed state上访问键/值状态，即在keyed/分区数据交换之后，并且仅限于与当前事件键关联的值。对齐流和状态的键确保所有状态更新都是本地操作，保证一致性而没有事务开销。这种对齐还允许 Flink 重新分配状态并透明地调整流分区。

Keyed State 被进一步组织成所谓的 Key Groups。 Key Groups 是 Flink 可以重新分配 Keyed State 的原子单元； Key Groups 的数量与定义的最大并行度完全相同。

在执行期间，keyed operator运算符的每个并行实例都与一个或多个key groups的键一起工作。

状态持久化

Flink 使用流重放和检查点的组合来实现容错。检查点标记每个输入流中的特定点以及每个运算符的相应状态。流式数据流可以从检查点恢复，同时通过恢复操作符的状态并从检查点重放记录来保持一致性（恰好一次处理语义）。

检查点间隔是一种权衡执行期间容错开销与恢复时间（需要重放的记录数）的方法。

容错机制不断绘制分布式流数据流的快照。对于状态较小的流式应用程序，这些快照非常轻量级，可以频繁绘制，而不会对性能产生太大影响。流应用程序的状态存储在可配置的位置，通常在分布式文件系统中。

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

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

因为 Flink 的检查点是通过分布式快照实现的，所以我们可以互换使用快照和检查点这两个词。通常我们也使用术语快照来表示检查点或保存点。

检查点

Flink 容错机制的核心部分是绘制分布式数据流和算子状态的一致快照。这些快照充当一致的检查点，系统可以在发生故障时回退到这些检查点。 Flink 绘制这些快照的机制在“Lightweight Asynchronous Snapshots for Distributed Dataflows”中有描述。它受到分布式快照的标准 Chandy-Lamport 算法的启发，专门针对 Flink 的执行模型量身定制。

请记住，与检查点有关的所有事情都可以异步完成。检查点屏障不会以锁定步骤移动，操作可以异步快照它们的状态。

从 Flink 1.11 开始，检查点可以在有或没有对齐的情况下进行。在本节中，我们首先描述对齐的检查点。

屏障

Flink 分布式快照的一个核心元素是流屏障。这些屏障被注入到数据流中，并作为数据流的一部分与记录一起流动。屏障永远不会超过记录，它们严格按照规定流动。屏障将数据流中的记录分成进入当前快照的记录集和进入下一个快照的记录。每个屏障都带有将其记录推送到其前面的快照的 ID。屏障不会中断流的流动，因此非常轻巧。来自不同快照的多个屏障可以同时在流中，这意味着各种快照可能同时发生。

流屏障被注入到流源的并行数据流中。快照 n 的屏障注入点（我们称之为 Sn）是源流中快照覆盖数据的位置。例如，在 Apache Kafka 中，此位置将是分区中最后一条记录的偏移量。这个位置 Sn 被报告给检查点协调器（Flink 的 JobManager）。

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

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

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

一旦算子从传入流中接收到快照屏障 n，它就无法处理来自该流的任何进一步的记录，直到它也从其他输入接收到屏障 n。否则，它将混合属于快照 n 的记录和属于快照 n+1 的记录。
一旦最后一个流接收到屏障 n，算子就会发出所有待处理的传出记录，然后自己发出快照 n 屏障。
它对状态进行快照并继续处理来自所有输入流的记录，在处理来自流的记录之前处理来自输入缓冲区的记录。
最后，算子将状态异步写入状态后端。

请注意，所有具有多个输入的算子和经过shuffle的算子在消耗多个上游子任务的输出流时都需要对齐。

对状态算子做快照

当算子包含任何形式的状态时，该状态也必须是快照的一部分。

算子在从其输入流接收到所有快照屏障的时间点以及在将屏障发送到其输出流之前的时间点快照其状态。那时，在屏障之前的记录中对状态的所有更新都已经完成，并且在屏障应用之后没有依赖于记录的更新。因为快照的状态可能很大，所以它存储在可配置的状态后端中。默认情况下，这是 JobManager 的内存，但对于生产用途，应配置分布式可靠存储（例如 HDFS）。存储状态后，算子确认检查点，将快照屏障发送到输出流中，然后继续。

生成的快照现在包含：

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

从检查点恢复

这种机制下的恢复很简单：一旦发生故障，Flink 会选择最近完成的检查点 k。然后系统重新部署整个分布式数据流，并为每个算子提供作为检查点 k 的一部分快照的状态。源设置为从位置 Sk 开始读取流。例如在 Apache Kafka 中，这意味着告诉消费者从偏移量 Sk 开始获取。

如果状态是增量快照，则算子从最新的完整快照的状态开始，然后对该状态应用一系列增量快照更新。

有关详细信息，请参阅重新启动策略。

未对齐的检查点

检查点也可以在未对齐的情况下执行。基本思想是，只要流动中的数据成为算子状态的一部分，检查点就可以超越所有流动中的数据。

请注意，这种方法实际上更接近 Chandy-Lamport 算法，但 Flink 仍然在源中插入屏障以避免检查点协调器过载。

该图描述了算子如何处理未对齐的检查点障碍：

算子对存储在其输入缓冲区中的第一个屏障做出反应。
它通过将屏障添加到输出缓冲区的末尾，立即将屏障转发给下游算子。
算子将所有被超越的记录标记为异步存储，并创建自己状态的快照。

因此，算子只需短暂停止输入处理以标记缓冲区、转发屏障并创建其他状态的快照。

未对齐的检查点确保障碍物尽快到达接收器。它特别适用于具有至少一个缓慢移动数据路径的应用程序，其中对齐时间可能达到数小时。但是，由于它增加了额外的 I/O 压力，因此当状态后端的 I/O 成为瓶颈时，它也无济于事。有关其他限制，请参阅 ops 中更深入的讨论。

请注意，保存点将始终对齐。

从未对齐的检查点恢复

算子首先在未对齐检查点中开始处理来自上游算子的任何数据之前恢复运行中的数据。除此之外，它执行与对齐检查点恢复期间相同的步骤。

状态后端

存储键/值索引的确切数据结构取决于所选的状态后端。一个状态后端将数据存储在内存中的哈希映射中，另一个状态后端使用 RocksDB 作为键/值存储。除了定义保存状态的数据结构外，状态后端还实现了获取键/值状态的时间点快照并将该快照存储为检查点的一部分的逻辑。可以在不更改应用程序逻辑的情况下配置状态后端。