Flink状态一致性保证

前言

一个Flink作业由一系列算子构成，每个算子可以有多个并行实例，这些实例被称为 subTask，每个subTask运行在不同的进程或物理机上，以实现作业的并行处理。在这个复杂的分布式场景中，任何一个节点故障都有可能导致 Flink 作业宕机，Flink 状态本地化虽然可以实现极致的访问速度，但是节点故障后的状态恢复问题也是Flink必须要解决的。

状态持久化

恢复状态最简单粗暴的方式，就是回溯全量数据，重新计算一遍。不过缺点也很明显，首先，有的数据源压根就不支持保存全量数据，例如Kafka可能就只保存近几天甚至几小时的数据；其次，回溯全量数据必然会消耗大量时间，导致作业产出结果出现较大延时，这本身就和Flink高吞吐低延时的目标相悖。

于是，Flink 推出了状态持久化方案，Flink 作业运行时会自动、定时地将状态数据持久化到远程分布式文件系统中，一旦 Flink 作业异常重启，就会从远程分布式文件系统中读取最新的快照恢复状态，避免了状态数据丢失的问题。

如下示例，数据源会不断产生一些数字，Flink 作业会对这些数字求和，并输出到目标数据库。第一步，数字1输入，subTask更新本地状态sum=1，然后将其持久化到远程文件系统，此时作业异常宕机，本地状态丢失；第二步，Flink 作业重启，从远程文件系统恢复状态sum=1；第三步，subTask继续处理数据，整个过程就像没发生故障一样。

状态一致性

状态持久化只实现了基本的异常容错，用户往往还有“状态一致性”的诉求。发生故障时，Flink不仅要能从远程文件系统中恢复状态数据，还要能协调所有subTask节点在故障恢复后实现数据的精准一次处理，也就是数据即不会多算，也不会少算，以保证作业的计算结果如同没有发生过故障一样。

流计算的状态一致性有三个等级：

• at-most-once 最多计算一次，允许数据丢失，最弱的一致性保证
• at-least-once 至少计算一次，允许数据重复计算，对于自身具备幂等性写入的业务指标可以保证一致性
• exactly-once 精准计算一次，最强的一致性保证，数据不会多算也不会少算

仅仅通过状态持久化，只能保证 at-most-once 一致性，本地状态更新后还没来得及保存到远程文件系统时发生故障，数据就会丢失，导致漏算。

如下图所示，数据2处理完，本地状态更新sum=3，状态还没来得及持久化就发生故障，重启后恢复状态sum=1，数据2的计算丢失了，数据漏算。

要想避免数据漏算，可以通过故障恢复时向前回溯一部分数据来解决，例如回溯前一小时的数据甚至全部数据，这样可以保证数据至少被计算一次，也就是满足 at-least-once 一致性，但是会有数据被重复计算，对于本身具备幂等性的业务指标这没什么问题，非幂等性的业务指标计算结果仍不准确。

最理想的一致性场景就是 exactly-once，数据精准计算一次，既不多算也不少算。在咱们这个例子中，要想实现 exactly-once 一致性，除了同步sum状态，还要同步作业处理数据的偏移量offset，故障恢复时，根据恢复的offset从指定的位置重新读取数据进行处理。

如图所示，第一步处理数字1求和，更新本地状态sum=1、offset=1并持久化到远程文件系统；第二步处理数字2求和，更新本地状态sum=3、offset=2，状态还没持久化时发生故障，本地状态丢失；第三步从远程文件系统恢复状态；第四步从offset=1处开始继续处理数据2，更新本地状态并输出结果，整个流程就像没发生过故障一样。

由此可见，要满足 exactly-once 一致性，有以下几个条件：

• 数据源支持根据偏移量回溯
• subTask持久化状态的同时，也要持久化偏移量offset
• subTask持久化状态和处理数据要互斥，不能持久化状态的同时还处理数据

一个完整的Flink作业由若干个subTask构成，运行在一个复杂的分布式环境中，Flink作业状态一致性的前提是每个subTask先保证自身状态一致性。对于Source算子subTask来说，如果数据源支持根据offset回溯数据，那么执行上述流程不会有问题。但是对于下游非Source算子subTask来说，情况会显得更加复杂。

Source算子subTask读取到数据后，是通过Socket传输给下游subTask的，Socket通道的数据首先不支持回溯，其次数据压根就没有offset，这就意味着下游subTask可能会漏算数据，又回到 at-most-once 一致性了。丢失的这些数据不能让上游subTask重发，因为上游subTask根本就不知道下游subTask的处理结果是成功还是失败，如果再额外引入一套ACK机制，增加复杂度不说，额外的性能消耗也是Flink无法承受的。

既然上游不支持重发，就只能下游subTask自己解决了。下游subTask在收到上游传过来的数据时，除了计算并更新本地状态外，还要将收到的这部分数据也写进状态里面，打快照时和状态一同持久化。故障恢复时，除了恢复状态外，再把这部分数据拿出来重新计算一下，最终的状态结果就是准确的了。下游subTask也无需保存接收到的所有数据，只要数据被计算过且打过快照，这部分数据就没用了，所以下游subTask要保存的数据，只有上游subTask开始执行快照到下游subTask开始执行快照时的这部分数据，怎么让下游subTask知道上游subTask在执行快照呢？很简单，上游subTask执行快照时给下游subTask广播一条特殊的消息即可，这个消息被称为“barrier”（屏障）。

再次总结一下，要满足 exactly-once 一致性，满足以下条件：

• 数据源支持根据offset回溯
• Source算子持久化offset，并向下游算子广播barrier
• subTask持久化状态和处理数据要互斥，不能持久化状态的同时还处理数据
• 非Source算子subTask要持久化两部分数据：本地状态数据、上游subTask执行快照到自己执行快照这段时间接收到的数据
• 所有subTask快照执行成功，才算一次完整的快照

故障恢复时，Source算子从远程文件系统恢复offset，根据offset回溯数据源，并发送给下游subTask；下游subTask先从远程文件系统恢复状态，再读取之前上游发送给自己的数据，重新计算一遍这部分数据恢复自身状态，再继续处理上游发给自己的数据。

Checkpoint机制

有了上述理论，再看Flink Checkpoint机制就很容易理解了。

Flink 以 Chandy-Lamport 算法理论为基础，实现了一套分布式轻量级异步快照算法，即 Flink Checkpoint。

每个需要Checkpoint的Flink应用启动时，JobManager都会为其创建一个 CheckpointCoordinator（检查点协调器）的组件，由它来负责生成全局快照，流程如下：

• CheckpointCoordinator周期性的向所有Source算子的subTask发送barrier，开始执行快照
• Source算子收到barrier，暂停处理数据，将本地状态持久化到远程文件系统，并向CheckpointCoordinator报告自己的快照结果，同时向下游subTask广播barrier
• 下游subTask收到barrier，同样暂停数据处理。对于有多个输入的subTask来说，需要收到所有上游发来的subTask才会开始执行快照，这里就存在barrier对齐的问题。subTask同样地将本地状态持久化到远程文件系统，并向CheckpointCoordinator报告自己的快照结果，同时将barrier转发给下游subTask，直到Sink算子
• 当CheckpointCoordinator收到所有算子的快照成功报告之后，认为该周期的快照制作成功。如果没有在指定时间内收到所有算子的报告，则认定为快照制作失败。

Checkpoint 优化了subTask执行快照的时机，避免了整个快照期间，所有subTask都要暂停处理数据的问题。CheckpointCoordinator负责通知Source算子执行快照，而下游算子执行快照的时机，依赖于上游算子发送过来的barrier，这套机制执行快照无需暂停整个作业的数据处理，有效降低了流处理作业的延时问题。