简介

Apache Flink 提供了容错机制来恢复数据流应用的状态。这种机制保证即使在错误出现时，应用的状态会最终反应数据流中的每条记录恰好一次(exactly once)。注意，可以选择降级到至少一次的保证（at least once）

这种容错机制不断的为分布式数据流建立快照。对于拥有小状态（数据量较小）的流应用，这种快照特别的轻量，在不影响太多性能的情况下不断地建立快照。这个状态存放在配置好的地方（例如 主节点或HDFS）

当遇到应用失败（因为机器、网络或软件失败），Flink 就会停止分布式数据流。然后，系统会重启operator，并将它们设置为最新成功的检查点（checkpoint）。输入数据流会被设置到状态快照对应的点。保证重启后的并行数据流所处理的任何记录都在检查点状态之后。

注意：为了实现该机制的所有保证，数据流源（例如消息队列或broker）需要能够回滚到近段历史上的某个点。Kafka 拥有这个能力，而且Flink的Kafka 连接器实现了这个能力。

注意：因为Flink的检查点通过分布式快照实现，因此我们使用快照（snapshot）和 检查点(checkpoint) 表示。

个人理解： exactly once 仅对Flink内部状态而言，对外部系统的影响是at least once。主要作用为：

1. 对于流聚合的应用，保证聚合状态的正确性

2. 如果状态会影响流的处理，保证流处理结果的正确性

检查点

Flink 容错机制的关键部分是为分布式数据系统建立一致性快照和操作状态。这些快照充当一致性检查点，在出现失败时，就可以回滚。Flink这种建立快照的机制在“分布式数据流中的轻量级异步快照” （http://arxiv.org/abs/1506.08603）中进行详细描述。它受“Chandy-Lamport algorithm”启发，并为Flink的执行模型做了适配。

栅栏（Barriers）

Flink分布式快照中的一个关键元素是流栅栏（stream barriers）。这个栅栏被注入到数据流中，这些记录流作为数据流的一部分。Barriers 绝不超车其他的记录，会严格的保证顺序。Barrier将记录分割成记录集，并流入不同的快照中。每一个barrier都为快照携带一个ID。Barries不会打断数据流的流动，因此非常的轻量。不同快照的多个barrier可以在一个流中同时出现，这就意味着不同的快照会同时发生。

  屏幕快照 2018-09-18 下午9.33.57.png

流栅栏会在数据流源头被注入到并行数据流中。为快照n(Sn)产生的栅栏注入的点就是在源头数据流中包含这些快照数据的位置。例如，在kafka中，这个位置就是最后一个记录在分区内的位置。Sn的位置会被报告给检查点的协调者（Flink的JobManager）。

栅栏接下来就会向下游流动。当一个中游的operator从所有的输入流中收到了快照n的barrier，他就会相同的所有下游流发送快照n的barrier。一旦尾operator（sink operator，流DAG的终点）已经从所有的输入流中收到了barrier n，它就会将快照n想检查点协调者反馈。当所有的sink反馈了该快照，他就被认为已经完成了。

当快照n已经完成了，可以确定Sn之前的所有记录在source中都不再需要，因为这些记录（及其后代）已经通过了整个拓扑。

  屏幕快照 2018-11-05 上午11.00.44.png

接收多个输入流的Operator需要将多个输入的快照栅栏对齐。说明如下：

• 一旦operator从一个输入流中收到了快照barrier n，它将不能再处理这个流中的其他记录知道它从其他的流中也收到了该barrier n。否则，它会将来自于快照n中的记录和来自于快照n+1中的快照混淆。

• 报告barrier n的数据流会临时性的挑出来。从这些流中收到的记录不会处理，但是会放入到输入缓存中。

• 一旦从最后一个流收到了barrier n，这个operator会发送所有积压的记录（个人注：将barrier之前的数据都发送出去），然后发送快照n的barrier。

• 然后，它继续处理从所有输入流中的数据，先处理输入缓存中的数据，然后处理流中的数据。

状态

当operator包含了任意类型的状态，这些状态必须加入到快照中。Operator中的状态包含几个类型：

• 用户定义的状态：这种状态由transformation 的函数（如map() 或filter()）直接创建和修改。用户定义的状态可以是在函数java对象中的一个简单变量，或者函数绑定的key/value状态。

• 系统状态：这些状态是operator计算过程中的数据缓存。这种状态的典型例子是窗口缓存，在其中，系统为窗口收集（或聚合）记录知道窗口被触发。

Operator在收到所有输入流的快照栅栏时，且发送barrier到输出流前，将状态存为快照。这时，barrier之前的记录会更新状态，这些更新不会依赖与barrier之后的的记录。因为快照的状态可能会比较大，它被存储在配置好的状态后端。默认，会存放在JobManager的内存中，但是为了正式环境设置（serious setups），应该配置一个分布式的存储（例如HDFS）.在保存完成状态后，operator会反馈检查点，发送快照栅栏到输出流，然后继续处理。

现在快照包含：

• 对于任意分布式流数据源，快照开始时的位置

• 对于operator, 状态也会存储在快照中。

  
    屏幕快照 2018-11-05 上午11.02.12.png

准确一次 vs 至少一次

对齐步骤（barrier对对齐）可能会给流应用带来延迟。正常情况下，额外的延迟大约在几毫秒，但是我们见过一写延迟增长很多的情况。对于所有记录需要持续超低延迟（几毫秒）的应用，Flink有在检查点中跳过流对齐的选项。一旦从每一个输入流中收到检查点barrier，就会建立检查点快照。

当跳过对齐步骤时，即使在n检查点的检查点栅栏之后，operator持续处理所有的输入。这样，在检查点n建立快照之前，operator也会处理属于检查点n+1的元素。在重启时（on a restore）,这些记录会出现重复，因为他们都会包含在检查点n内，也会作为检查点n后的数据被重放。

注意： 对齐之后发生在operator处理多个输入时（join）或者有多个发送者时（stream 重排或重分区）。因此，对于只有一个并发度的operator(map()、flatMap()、filter()……)及时是至少一次模式下也会保证准确一次。

异步状态快照

注意上面描述的机制意味着当operator在状态服务中保存状态的快照时停止处理输入的记录。同步状态快照会在每次建立快照时引入延迟。

如果Operator使得状态存储在后台异步完成，那么在存储状态快照时继续处理数据是可能的。为此，Operator必须能够生产一种对后期对operator状态的修改不会影响到当前状态对象的状态对象。在RocksDB中使用的copy-on-write式的数据结构就是一个例子。

在输入流中收到检查点栅栏后，operator开始异步复制快照状态。它立马就会向输出流发送barrier并继续常规数据流处理。一旦后台复制流程完成，它就会将检查点协调者（JobManager）汇报检查点。检查点只有在所有的sink收到barrier并且有状态的operator后台汇报处理完成时才会结束（有时会比barrier到达sink的时间要晚）。

恢复

在这种机制下恢复就简单了。一旦失败，Flink选择最近完成的检查点k。然后，系统会重新部署整个分布式数据流，给与每一个operator检查点k对应的状态。数据源会从Sk对应的位置读取数据流。以Apache Kafka为例，这意味着告诉消费者从offset Sk 处拉取数据。

如果状态是增量快照的，那么operator启动时一最新的全量快照状态，然后将增量快照应用于该状态。

Operator 快照实现

当operator采用快照时，存在两部分，同步和异步。

Operator 和状态服务使用Java FutureTask提供快照。这个任务包含同步部分完成时的状态，异步部分在阻塞中。然后，该检查点的一个后台线程会执行异步部分。

纯异步检查点的Operator会返回一个已经完成的FutureTask。如果一个异步操作需要执行，他会在FutureTask的run()方法内执行

这个任务是可以取消的，以便于释放数据流及其他处理操作所需要的资源。

 

转载自链接：https://www.jianshu.com/p/84c89078c60d