（3）重放数据
从检查点恢复状态后还有一个问题：如果直接继续处理数据，那么保存检查点之后、到发生故障这段时间内的数据，也就是第4、5个数据（“flink”“hello”）就相当于丢掉了；这会造成计算结果的错误。为了不丢数据，我们应该从保存检查点后开始重新读取数据，这可以通过Source任务向外部数据源重新提交偏移量（offset）来实现，如图所示。
这样，整个系统的状态已经完全回退到了检查点保存完成的那一时刻。
（4）继续处理数据
接下来，我们就可以正常处理数据了。首先是重放第4、5个数据，然后继续读取后面的数据，如图所示。
当处理到第5个数据时，就已经追上了发生故障时的系统状态。之后继续处理，就好像没有发生过故障一样；我们既没有丢掉数据、也没有重复计算数据，这就保证了计算结果的正确性。在分布式系统中，这叫做实现了“精确一次”（exactly-once）的状态一致性保证。
14.1.3 检查点算法
在Flink中，采用了基于Chandy-Lamport算法的分布式快照，可以在不暂停整体流处理的前提下，将状态备份保存到检查点。
1. 检查点分界线（Barrier）
我们可以借鉴水位线（watermark）的设计，在数据流中插入一个特殊的数据结构，专门用来表示触发检查点保存的时间点。收到保存检查点的指令后，Source任务可以在当前数据流中插入这个结构；之后的所有任务只要遇到它就开始对状态做持久化快照保存。由于数据流是保持顺序依次处理的，因此遇到这个标识就代表之前的数据都处理完了，可以保存一个检查点；而在它之后的数据，引起的状态改变就不会体现在这个检查点中，而需要保存到下一个检查点。
这种特殊的数据形式，把一条流上的数据按照不同的检查点分隔开，所以就叫做检查点的“分界线”（Checkpoint Barrier）。与水位线很类似，检查点分界线也是一条特殊的数据，由Source算子注入到常规的数据流中，它的位置是限定好的，不能超过其他数据，也不能被后面的数据超过。检查点分界线中带有一个检查点ID，这是当前要保存的检查点的唯一标识，如图所示。
这样，分界线就将一条流逻辑上分成了两部分：分界线之前到来的数据导致的状态更改，都会被包含在当前分界线所表示的检查点中；而基于分界线之后的数据导致的状态更改，则会被包含在之后的检查点中。
在JobManager中有一个“检查点协调器”（checkpoint coordinator），专门用来协调处理检查点的相关工作。检查点协调器会定期向TaskManager发出指令，要求保存检查点（带着检查点ID）；TaskManager会让所有的Source任务把自己的偏移量（算子状态）保存起来，并将带有检查点ID的分界线（barrier）插入到当前的数据流中，然后像正常的数据一样像下游传递；之后Source任务就可以继续读入新的数据了。
每个算子任务只要处理到这个barrier，就把当前的状态进行快照；在收到barrier之前，还是正常地处理之前的数据，完全不受影响。比如上图中，Source任务收到1号检查点保存指令时，读取完了三个数据，所以将偏移量3保存到外部存储中；而后将ID为1的barrier注入数据流；与此同时，Map任务刚刚收到上一条数据“hello”，而Sum任务则还在处理之前的第二条数据(world, 1)。下游任务不会在这时就立刻保存状态，而是等收到barrier时才去做快照，这时可以保证前三个数据都已经处理完了。同样地，下游任务做状态快照时，也不会影响上游任务的处理，每个任务的快照保存并行不悖，不会有暂停等待的时间。
2. 分布式快照算法
watermark指示的是“之前的数据全部到齐了”，而barrier指示的是“之前所有数据的状态更改保存入当前检查点”：它们都是一个“截止时间”的标志。所以在处理多个分区的传递时，也要以是否还会有数据到来作为一个判断标准。
具体实现上，Flink使用了Chandy-Lamport算法的一种变体，被称为“异步分界线快照”（asynchronous barrier snapshotting）算法。算法的核心就是两个原则：当上游任务向多个并行下游任务发送barrier时，需要广播出去；而当多个上游任务向同一个下游任务传递barrier时，需要在下游任务执行“分界线对齐”（barrier alignment）操作，也就是需要等到所有并行分区的barrier都到齐，才可以开始状态的保存。
为了详细解释检查点算法的原理，我们对之前的word count程序进行扩展，考虑所有算子并行度为2的场景，如图所示。
我们有两个并行的Source任务，会分别读取两个数据流（或者是一个源的不同分区）。这里每条流中的数据都是一个个的单词：“hello”“world”“hello”“flink”交替出现。此时第一条流的Source任务（为了方便，下文中我们直接叫它“Source 1”，其它任务类似）读取了3个数据，偏移量为3；而第二条流的Source任务（Source 2）只读取了一个“hello”数据，偏移量为1。
接下来就是检查点保存的算法。具体过程如下：
（1）JobManager发送指令，触发检查点的保存；Source任务保存状态，插入分界线
JobManager 会周期性地向每个 TaskManager发送一条带有新检查点 ID 的消息，通过这种方式来启动检查点。收到指令后，TaskManger会在所有Source 任务中插入一个分界线（barrier），并将偏移量保存到远程的持久化存储中，如图所示。
并行的Source任务保存的状态为3和1，表示当前的1号检查点应该包含：第一条流中截至第三个数据、第二条流中截至第一个数据的所有状态更改。可以发现Source任务做这些的时候并不影响后面任务的处理，Sum任务已经处理完了第一条流中传来的(world, 1)，对应的状态也有了更改。
（2）状态快照保存完成，分界线向下游传递
状态存入持久化存储之后，会返回通知给 Source 任务；Source 任务就会向 JobManager 确认检查点完成，然后跟数据一样把barrier向下游任务传递，如图所示。
由于Source和Map之间是一对一（forward）的传输关系（这里没有考虑算子链operator chain），所以barrier可以直接传递给对应的Map任务。之后Source任务就可以继续读取新的数据了。与此同时，Sum 1已经将第二条流传来的(hello，1)处理完毕，更新了状态。
（3）向下游多个并行子任务广播分界线，执行分界线对齐
Map任务没有状态，所以直接将barrier继续向下游传递。这时由于进行了keyBy分区，所以需要将barrier广播到下游并行的两个Sum任务，如图所示。同时，Sum任务可能收到来自上游两个并行Map任务的barrier，所以需要执行“分界线对齐”操作。
此时的Sum 2收到了来自上游两个Map任务的barrier，说明第一条流第三个数据、第二条流第一个数据都已经处理完，可以进行状态的保存了；而Sum 1只收到了来自Map 2的barrier，所以这时需要等待分界线对齐。在等待的过程中，如果分界线尚未到达的分区任务Map 1又传来了数据(hello, 1)，说明这是需要保存到检查点的，Sum任务应该正常继续处理数据，状态更新为3；而如果分界线已经到达的分区任务Map 2又传来数据，这已经是下一个检查点要保存的内容了，就不应立即处理，而是要缓存起来、等到状态保存之后再做处理。
（4）分界线对齐后，保存状态到持久化存储
各个分区的分界线都对齐后，就可以对当前状态做快照，保存到持久化存储了。存储完成之后，同样将barrier向下游继续传递，并通知JobManager保存完毕，如图所示。
这个过程中，每个任务保存自己的状态都是相对独立的，互不影响。我们可以看到，当Sum将当前状态保存完毕时，Source 1任务已经读取到第一条流的第五个数据了。
（5）先处理缓存数据，然后正常继续处理
完成检查点保存之后，任务就可以继续正常处理数据了。这时如果有等待分界线对齐时缓存的数据，需要先做处理；然后再按照顺序依次处理新到的数据。当JobManager收到所有任务成功保存状态的信息，就可以确认当前检查点成功保存。之后遇到故障就可以从这里恢复了。由于分界线对齐要求先到达的分区做缓存等待，一定程度上会影响处理的速度；当出现背压（backpressure）时，下游任务会堆积大量的缓冲数据，检查点可能需要很久才可以保存完毕。为了应对这种场景，Flink 1.11之后提供了不对齐的检查点保存方式，可以将未处理的缓冲数据（in-flight data）也保存进检查点。这样，当我们遇到一个分区barrier时就不需等待对齐，而是可以直接启动状态的保存了。
14.1.4 检查点配置
检查点的作用是为了故障恢复，我们不能因为保存检查点占据了大量时间、导致数据处理性能明显降低。为了兼顾容错性和处理性能，我们可以在代码中对检查点进行各种配置。
1. 启用检查点
默认情况下，Flink程序是禁用检查点的。如果想要为Flink应用开启自动保存快照的功能，需要在代码中显式地调用执行环境的.enableCheckpointing()方法：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 每隔1秒启动一次检查点保存 env.enableCheckpointing(1000);
这里需要传入一个长整型的毫秒数，表示周期性保存检查点的间隔时间。如果不传参数直接启用检查点，默认的间隔周期为500毫秒，这种方式已经被弃用。检查点的间隔时间是对处理性能和故障恢复速度的一个权衡。如果我们希望对性能的影响更小，可以调大间隔时间；而如果希望故障重启后迅速赶上实时的数据处理，就需要将间隔时间设小一些。
2. 检查点存储（Checkpoint Storage）
检查点具体的持久化存储位置，取决于“检查点存储”（CheckpointStorage）的设置。默认情况下，检查点存储在JobManager的堆（heap）内存中。而对于大状态的持久化保存，Flink也提供了在其他存储位置进行保存的接口，这就是CheckpointStorage。具体可以通过调用检查点配置的.setCheckpointStorage()来配置，需要传入一个CheckpointStorage的实现类。Flink主要提供了两种CheckpointStorage：作业管理器的堆内存（JobManagerCheckpointStorage）和文件系统（FileSystemCheckpointStorage）。
// 配置存储检查点到JobManager堆内存 env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage()); // 配置存储检查点到文件系统 env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));
对于实际生产应用，我们一般会将CheckpointStorage配置为高可用的分布式文件系统（HDFS，S3等）。
3. 其它高级配置
检查点还有很多可以配置的选项，可以通过获取检查点配置（CheckpointConfig）来进行设置。CheckpointConfig checkpointConfig = env.getCheckpointConfig(); 我们这里做一个简单的列举说明：
1.检查点模式（CheckpointingMode） 设置检查点一致性的保证级别，有“精确一次”（exactly-once）和“至少一次”（at-least-once）两个选项。默认级别为exactly-once，而对于大多数低延迟的流处理程序，at-least-once就够用了，而且处理效率会更高。关于一致性级别，我们会在10.2节继续展开。 2.超时时间（checkpointTimeout） 用于指定检查点保存的超时时间，超时没完成就会被丢弃掉。传入一个长整型毫秒数作为参数，表示超时时间。 3.最小间隔时间（minPauseBetweenCheckpoints） 用于指定在上一个检查点完成之后，检查点协调器（checkpoint coordinator）最快等多久可以出发保存下一个检查点的指令。这就意味着即使已经达到了周期触发的时间点，只要距离上一个检查点完成的间隔不够，就依然不能开启下一次检查点的保存。这就为正常处理数据留下了充足的间隙。当指定这个参数时，maxConcurrentCheckpoints的值强制为1。 4.最大并发检查点数量（maxConcurrentCheckpoints） 用于指定运行中的检查点最多可以有多少个。由于每个任务的处理进度不同，完全可能出现后面的任务还没完成前一个检查点的保存、前面任务已经开始保存下一个检查点了。这个参数就是限制同时进行的最大数量。 如果前面设置了minPauseBetweenCheckpoints，则maxConcurrentCheckpoints这个参数就不起作用了。 5.开启外部持久化存储（enableExternalizedCheckpoints） 用于开启检查点的外部持久化，而且默认在作业失败的时候不会自动清理，如果想释放空间需要自己手工清理。里面传入的参数ExternalizedCheckpointCleanup指定了当作业取消的时候外部的检查点该如何清理。 DELETE_ON_CANCELLATION：在作业取消的时候会自动删除外部检查点，但是如果是作业失败退出，则会保留检查点。 RETAIN_ON_CANCELLATION：作业取消的时候也会保留外部检查点。 6.检查点异常时是否让整个任务失败（failOnCheckpointingErrors） 用于指定在检查点发生异常的时候，是否应该让任务直接失败退出。默认为true，如果设置为false，则任务会丢弃掉检查点然后继续运行。 7..不对齐检查点（enableUnalignedCheckpoints） 不再执行检查点的分界线对齐操作，启用之后可以大大减少产生背压时的检查点保存时间。这个设置要求检查点模式（CheckpointingMode）必须为exctly-once，并且并发的检查点个数为1。
代码中具体设置如下：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 启用检查点，间隔时间1秒 env.enableCheckpointing(1000); CheckpointConfig checkpointConfig = env.getCheckpointConfig(); // 设置精确一次模式 checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 最小间隔时间500毫秒 checkpointConfig.setMinPauseBetweenCheckpoints(500); // 超时时间1分钟 checkpointConfig.setCheckpointTimeout(60000); // 同时只能有一个检查点 checkpointConfig.setMaxConcurrentCheckpoints(1); // 开启检查点的外部持久化保存，作业取消后依然保留 checkpointConfig.enableExternalizedCheckpoints( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 启用不对齐的检查点保存方式 checkpointConfig.enableUnalignedCheckpoints(); // 设置检查点存储，可以直接传入一个String，指定文件系统的路径 checkpointConfig.setCheckpointStorage("hdfs://my/checkpoint/dir")
14.1.5 保存点（Savepoint）
除了检查点（checkpoint）外，Flink还提供了另一个非常独特的镜像保存功能——保存点（savepoint）。从名称就可以看出，这也是一个存盘的备份，它的原理和算法与检查点完全相同，只是多了一些额外的元数据。事实上，保存点就是通过检查点的机制来创建流式作业状态的一致性镜像（consistent image）的。保存点中的状态快照，是以算子ID和状态名称组织起来的，相当于一个键值对。从保存点启动应用程序时，Flink会将保存点的状态数据重新分配给相应的算子任务。
1. 保存点的用途
保存点与检查点最大的区别，就是触发的时机。检查点是由Flink自动管理的，定期创建，发生故障之后自动读取进行恢复，这是一个“自动存盘”的功能；而保存点不会自动创建，必须由用户明确地手动触发保存操作，所以就是“手动存盘”。因此两者尽管原理一致，但用途就有所差别了：检查点主要用来做故障恢复，是容错机制的核心；保存点则更加灵活，可以用来做有计划的手动备份和恢复。保存点可以当作一个强大的运维工具来使用。我们可以在需要的时候创建一个保存点，然后停止应用，做一些处理调整之后再从保存点重启。它适用的具体场景有：
1.版本管理和归档存储 对重要的节点进行手动备份，设置为某一版本，归档（archive）存储应用程序的状态。 2.更新Flink版本 目前Flink的底层架构已经非常稳定，所以当Flink版本升级时，程序本身一般是兼容的。这时不需要重新执行所有的计算，只要创建一个保存点，停掉应用、升级Flink后，从保存点重启就可以继续处理了。 3.更新应用程序 我们不仅可以在应用程序不变的时候，更新Flink版本；还可以直接更新应用程序。前提是程序必须是兼容的，也就是说更改之后的程序，状态的拓扑结构和数据类型都是不变的，这样才能正常从之前的保存点去加载。 这个功能非常有用。我们可以及时修复应用程序中的逻辑bug，更新之后接着继续处理；也可以用于有不同业务逻辑的场景，比如A/B测试等等。 4.调整并行度 如果应用运行的过程中，发现需要的资源不足、或者已经有了大量剩余，也可以通过从保存点重启的方式，将应用程序的并行度增大或减小。 5.暂停应用程序 有时候我们不需要调整集群或者更新程序，只是单纯地希望把应用暂停、释放一些资源来处理更重要的应用程序。使用保存点就可以灵活实现应用的暂停和重启，可以对有限的集群资源做最好的优化配置。 需要注意的是，保存点能够在程序更改的时候依然兼容，前提是状态的拓扑结构和数据类型不变。我们知道保存点中状态都是以算子ID-状态名称这样的key-value组织起来的，算子ID可以在代码中直接调用SingleOutputStreamOperator的.uid()方法来进行指定：
DataStream<String> stream = env .addSource(new StatefulSource()) .uid("source-id") .map(new StatefulMapper()) .uid("mapper-id") .print();
对于没有设置ID的算子，Flink默认会自动进行设置，所以在重新启动应用后可能会导致ID不同而无法兼容以前的状态。所以为了方便后续的维护，强烈建议在程序中为每一个算子手动指定ID。
2. 使用保存点
保存点的使用非常简单，我们可以使用命令行工具来创建保存点，也可以从保存点恢复作业。
（1）创建保存点
要在命令行中为运行的作业创建一个保存点镜像，只需要执行：
bin/flink savepoint :jobId [:targetDirectory]
这里jobId需要填充要做镜像保存的作业ID，目标路径targetDirectory可选，表示保存点存储的路径。对于保存点的默认路径，可以通过配置文件flink-conf.yaml中的state.savepoints.dir项来设定：
state.savepoints.dir: hdfs:///flink/savepoints
当然对于单独的作业，我们也可以在程序代码中通过执行环境来设置：
env.setDefaultSavepointDir("hdfs:///flink/savepoints");
由于创建保存点一般都是希望更改环境之后重启，所以创建之后往往紧接着就是停掉作业的操作。除了对运行的作业创建保存点，我们也可以在停掉一个作业时直接创建保存点：
bin/flink stop --savepointPath [:targetDirectory] :jobId
（2）从保存点重启应用
我们已经知道，提交启动一个Flink作业，使用的命令是flink run；现在要从保存点重启一个应用，其实本质是一样的：
bin/flink run -s :savepointPath [:runArgs]
这里只要增加一个-s参数，指定保存点的路径就可以了，其它启动时的参数还是完全一样的。细心的读者可能还记得我们在第三章使用web UI进行作业提交时，可以填入的参数除了入口类、并行度和运行参数，还有一个“Savepoint Path”，这就是从保存点启动应用的配置。
14.2 状态一致性
之前讲到检查点又叫做“一致性检查点”，是Flink容错机制的核心。接下来我们就对状态一致性的概念进行展开，结合理论和实际应用场景，讨论一下Flink流式处理架构中的应对机制。
14.2.1 一致性的概念和级别
简单来讲，一致性其实就是结果的正确性。对于分布式系统而言，强调的是不同节点中相同数据的副本应该总是“一致的”，也就是从不同节点读取时总能得到相同的值；而对于事务而言，是要求提交更新操作后，能够读取到新的数据。对于Flink来说，多个节点并行处理不同的任务，我们要保证计算结果是正确的，就必须不漏掉任何一个数据，而且也不会重复处理同一个数据。流式计算本身就是一个一个来的，所以正常处理的过程中结果肯定是正确的；但在发生故障、需要恢复状态进行回滚时就需要更多的保障机制了。我们通过检查点的保存来保证状态恢复后结果的正确，所以主要讨论的就是“状态的一致性”。一般说来，状态一致性有三种级别：
1.最多一次（AT-MOST-ONCE） 当任务发生故障时，最简单的做法就是直接重启，别的什么都不干；既不恢复丢失的状态，也不重放丢失的数据。每个数据在正常情况下会被处理一次，遇到故障时就会丢掉，所以就是“最多处理一次”。 我们发现，如果数据可以直接被丢掉，那其实就是没有任何操作来保证结果的准确性；所以这种类型的保证也叫“没有保证”。尽管看起来比较糟糕，不过如果我们的主要诉求是“快”，而对近似正确的结果也能接受，那这也不失为一种很好的解决方案。 2.至少一次（AT-LEAST-ONCE） 在实际应用中，我们一般会希望至少不要丢掉数据。这种一致性级别就叫做“至少一次”（at-least-once），就是说是所有数据都不会丢，肯定被处理了；不过不能保证只处理一次，有些数据会被重复处理。 在有些场景下，重复处理数据是不影响结果的正确性的，这种操作具有“幂等性”。比如，如果我们统计电商网站的UV，需要对每个用户的访问数据进行去重处理，所以即使同一个数据被处理多次，也不会影响最终的结果，这时使用at-least-once语义是完全没问题的。当然，如果重复数据对结果有影响，比如统计的是PV，或者之前的统计词频word count，使用at-least-once语义就可能会导致结果的不一致了。 为了保证达到at-least-once 的状态一致性，我们需要在发生故障时能够重放数据。最常见的做法是，可以用持久化的事件日志系统，把所有的事件写入到持久化存储中。这时只要记录一个偏移量，当任务发生故障重启后，重置偏移量就可以重放检查点之后的数据了。Kafka就是这种架构的一个典型实现。 3.精确一次（EXACTLY-ONCE） 最严格的一致性保证，就是所谓的“精确一次”（exactly-once，有时也译作“恰好一次”）。这也是最难实现的状态一致性语义。exactly-once意味着所有数据不仅不会丢失，而且只被处理一次，不会重复处理。也就是说对于每一个数据，最终体现在状态和输出结果上，只能有一次统计。 exactly-once可以真正意义上保证结果的绝对正确，在发生故障恢复后，就好像从未发生过故障一样。
很明显，要做的exactly-once，首先必须能达到at-least-once的要求，就是数据不丢。所以同样需要有数据重放机制来保证这一点。另外还需要有专门的设计保证每个数据只被处理一次。Flink中使用的是一种轻量级快照机制——检查点（checkpoint）来保证exactly-once语义。
14.2.2 端到端的状态一致性
我们已经知道检查点可以保证Flink内部状态的一致性，而且可以做到精确一次（exactly-once）。那是不是说，只要开启了检查点，发生故障进行恢复，结果就不会有任何问题呢？
没那么简单。在实际应用中，一般要保证从用户的角度看来，最终消费的数据是正确的。而用户或者外部应用不会直接从Flink内部的状态读取数据，往往需要我们将处理结果写入外部存储中。这就要求我们不仅要考虑Flink内部数据的处理转换，还涉及到从外部数据源读取，以及写入外部持久化系统，整个应用处理流程从头到尾都应该是正确的。
所以完整的流处理应用，应该包括了数据源、流处理器和外部存储系统三个部分。这个完整应用的一致性，就叫做“端到端（end-to-end）的状态一致性”，它取决于三个组件中最弱的那一环。一般来说，能否达到at-least-once一致性级别，主要看数据源能够重放数据；而能否达到exactly-once级别，流处理器内部、数据源、外部存储都要有相应的保证机制。在下一节，我们就将详细讨论端到端的exactly-once一致性语义如何保证。
14.3 端到端精确一次（end-to-end exactly-once）
实际应用中，最难做到、也最希望做到的一致性语义，无疑就是端到端（end-to-end）的“精确一次”（exactly-once）。我们知道，对于Flink内部来说，检查点机制可以保证故障恢复后数据不丢（在能够重放的前提下），并且只处理一次，所以已经可以做到exactly-once的一致性语义了。所以端到端一致性的关键点，就在于输入的数据源端和输出的外部存储端。
14.3.1 输入端保证
输入端主要指的就是Flink读取的外部数据源。对于一些数据源来说，并不提供数据的缓冲或是持久化保存，数据被消费之后就彻底不存在了，例如socket文本流。对于这样的数据源，故障后我们即使通过检查点恢复之前的状态，可保存检查点之后到发生故障期间的数据已经不能重发了，这就会导致数据丢失。所以就只能保证at-most-once的一致性语义，相当于没有保证。
想要在故障恢复后不丢数据，外部数据源就必须拥有重放数据的能力。常见的做法就是对数据进行持久化保存，并且可以重设数据的读取位置。一个最经典的应用就是Kafka。在Flink的Source任务中将数据读取的偏移量保存为状态，这样就可以在故障恢复时从检查点中读取出来，对数据源重置偏移量，重新获取数据。
数据源可重放数据，或者说可重置读取数据偏移量，加上Flink的Source算子将偏移量作为状态保存进检查点，就可以保证数据不丢。这是达到at-least-once一致性语义的基本要求，当然也是实现端到端exactly-once的基本要求。
14.3.2 输出端保证
有了Flink的检查点机制，以及可重放数据的外部数据源，我们已经能做到at-least-once了。但是想要实现exactly-once却有更大的困难：数据有可能重复写入外部系统。
因为检查点保存之后，继续到来的数据也会一一处理，任务的状态也会更新，最终通过Sink任务将计算结果输出到外部系统；只是状态改变还没有存到下一个检查点中。这时如果出现故障，这些数据都会重新来一遍，就计算了两次。我们知道对Flink内部状态来说，重复计算的动作是没有影响的，因为状态已经回滚，最终改变只会发生一次；但对于外部系统来说，已经写入的结果就是泼出去的水，已经无法收回了，再次执行写入就会把同一个数据写入两次。
所以这时，我们只保证了端到端的at-least-once语义。为了实现端到端exactly-once，我们还需要对外部存储系统、以及Sink连接器有额外的要求。能够保证exactly-once一致性的写入方式有两种：
幂等写入 事务写入
我们需要外部存储系统对这两种写入方式的支持，而Flink也为提供了一些Sink连接器接口。接下来我们进行展开讲解。
1. 幂等（idempotent）写入
所谓“幂等”操作，就是说一个操作可以重复执行很多次，但只导致一次结果更改。也就是说，后面再重复执行就不会对结果起作用了。这相当于说，我们并没有真正解决数据重复计算、写入的问题；而是说，重复写入也没关系，结果不会改变。所以这种方式主要的限制在于外部存储系统必须支持这样的幂等写入：比如Redis中键值存储，或者关系型数据库（如MySQL）中满足查询条件的更新操作。需要注意，对于幂等写入，遇到故障进行恢复时，有可能会出现短暂的不一致。因为保存点完成之后到发生故障之间的数据，其实已经写入了一遍，回滚的时候并不能消除它们。如果有一个外部应用读取写入的数据，可能会看到奇怪的现象：短时间内，结果会突然“跳回”到之前的某个值，然后“重播”一段之前的数据。不过当数据的重放逐渐超过发生故障的点的时候，最终的结果还是一致的。
2. 事务（transactional）写入
如果说幂等写入对应用场景限制太多，那么事务写入可以说是更一般化的保证一致性的方式。输出端最大的问题，就是写入到外部系统的数据难以撤回。而利用事务就可以实现对已写入数据的撤回。事务（transaction）是应用程序中一系列严密的操作，所有操作必须成功完成，否则在每个操作中所作的所有更改都会被撤消。事务有四个基本特性：原子性(Atomicity)、一致性(Correspondence)、隔离性(Isolation)和持久性(Durability)，这就是著名的ACID。在Flink流处理的结果写入外部系统时，如果能够构建一个事务，让写入操作可以随着检查点来提交和回滚，那么自然就可以解决重复写入的问题了。所以事务写入的基本思想就是：用一个事务来进行数据向外部系统的写入，这个事务是与检查点绑定在一起的。当Sink任务遇到barrier时，开始保存状态的同时就开启一个事务，接下来所有数据的写入都在这个事务中；待到当前检查点保存完毕时，将事务提交，所有写入的数据就真正可用了。如果中间过程出现故障，状态会回退到上一个检查点，而当前事务没有正常关闭（因为当前检查点没有保存完），所以也会回滚，写入到外部的数据就被撤销了。具体来说，又有两种实现方式：预写日志（WAL）和两阶段提交（2PC）
（1）预写日志（write-ahead-log，WAL） 我们发现，事务提交是需要外部存储系统支持事务的，否则没有办法真正实现写入的回撤。那对于一般不支持事务的存储系统，能够实现事务写入呢？ 预写日志（WAL）就是一种非常简单的方式。具体步骤是： ①先把结果数据作为日志（log）状态保存起来 ②进行检查点保存时，也会将这些结果数据一并做持久化存储 ③在收到检查点完成的通知时，将所有结果一次性写入外部系统。 我们会发现，这种方式类似于检查点完成时做一个批处理，一次性的写入会带来一些性能上的问题；而优点就是比较简单，由于数据提前在状态后端中做了缓存，所以无论什么外部存储系统，理论上都能用这种方式一批搞定。在Flink中DataStream API提供了一个模板类GenericWriteAheadSink，用来实现这种事务型的写入方式。 需要注意的是，预写日志这种一批写入的方式，有可能会写入失败；所以在执行写入动作之后，必须等待发送成功的返回确认消息。在成功写入所有数据后，在内部再次确认相应的检查点，这才代表着检查点的真正完成。这里需要将确认信息也进行持久化保存，在故障恢复时，只有存在对应的确认信息，才能保证这批数据已经写入，可以恢复到对应的检查点位置。 但这种“再次确认”的方式，也会有一些缺陷。如果我们的检查点已经成功保存、数据也成功地一批写入到了外部系统，但是最终保存确认信息时出现了故障，Flink最终还是会认为没有成功写入。于是发生故障时，不会使用这个检查点，而是需要回退到上一个；这样就会导致这批数据的重复写入。
（2）两阶段提交（two-phase-commit，2PC） 前面提到的各种实现exactly-once的方式，多少都有点缺陷；而更好的方法就是传说中的两阶段提交（2PC）。 顾名思义，它的想法是分成两个阶段：先做“预提交”，等检查点完成之后再正式提交。这种提交方式是真正基于事务的，它需要外部系统提供事务支持。 具体的实现步骤为： ①当第一条数据到来时，或者收到检查点的分界线时，Sink任务都会启动一个事务。 ②接下来接收到的所有数据，都通过这个事务写入外部系统；这时由于事务没有提交，所以数据尽管写入了外部系统，但是不可用，是“预提交”的状态。 ③当Sink任务收到JobManager发来检查点完成的通知时，正式提交事务，写入的结果就真正可用了。 当中间发生故障时，当前未提交的事务就会回滚，于是所有写入外部系统的数据也就实现了撤回。这种两阶段提交（2PC）的方式充分利用了Flink现有的检查点机制：分界线的到来，就标志着开始一个新事务；而收到来自JobManager的checkpoint成功的消息，就是提交事务的指令。每个结果数据的写入，依然是流式的，不再有预写日志时批处理的性能问题；最终提交时，也只需要额外发送一个确认信息。所以2PC协议不仅真正意义上实现了exactly-once，而且通过搭载Flink的检查点机制来实现事务，只给系统增加了很少的开销。
Flink提供了TwoPhaseCommitSinkFunction接口，方便我们自定义实现两阶段提交的SinkFunction的实现，提供了真正端到端的exactly-once保证。不过两阶段提交虽然精巧，却对外部系统有很高的要求。这里将2PC对外部系统的要求列举如下：
1.外部系统必须提供事务支持，或者Sink任务必须能够模拟外部系统上的事务。 2.在检查点的间隔期间里，必须能够开启一个事务并接受数据写入。 3.在收到检查点完成的通知之前，事务必须是“等待提交”的状态。在故障恢复的情况下，这可能需要一些时间。如果这个时候外部系统关闭事务（例如超时了），那么未提交的数据就会丢失。 4.Sink任务必须能够在进程失败后恢复事务。 5.提交事务必须是幂等操作。也就是说，事务的重复提交应该是无效的。
可见，2PC在实际应用同样会受到比较大的限制。具体在项目中的选型，最终还应该是一致性级别和处理性能的权衡考量。
14.3.3 Flink和Kafka连接时的精确一次保证
在流处理的应用中，最佳的数据源当然就是可重置偏移量的消息队列了；它不仅可以提供数据重放的功能，而且天生就是以流的方式存储和处理数据的。所以作为大数据工具中消息队列的代表，Kafka可以说与Flink是天作之合，实际项目中也经常会看到以Kafka作为数据源和写入的外部系统的应用。在本小节中，我们就来具体讨论一下Flink和Kafka连接时，怎样保证端到端的exactly-once状态一致性。
1. 整体介绍
既然是端到端的exactly-once，我们依然可以从三个组件的角度来进行分析：
（1）Flink内部 Flink内部可以通过检查点机制保证状态和处理结果的exactly-once语义。 （2）输入端 输入数据源端的Kafka可以对数据进行持久化保存，并可以重置偏移量（offset）。所以我们可以在Source任务（FlinkKafkaConsumer）中将当前读取的偏移量保存为算子状态，写入到检查点中；当发生故障时，从检查点中读取恢复状态，并由连接器FlinkKafkaConsumer向Kafka重新提交偏移量，就可以重新消费数据、保证结果的一致性了。 （3）输出端 输出端保证exactly-once的最佳实现，当然就是两阶段提交（2PC）。作为与Flink天生一对的Kafka，自然需要用最强有力的一致性保证来证明自己。
Flink官方实现的Kafka连接器中，提供了写入到Kafka的FlinkKafkaProducer，它就实现了TwoPhaseCommitSinkFunction接口：
public class FlinkKafkaProducer<IN> extends TwoPhaseCommitSinkFunction<IN, FlinkKafkaProducer.KafkaTransactionState, FlinkKafkaProducer.KafkaTransactionContext> { ... }
也就是说，我们写入Kafka的过程实际上是一个两段式的提交：处理完毕得到结果，写入Kafka时是基于事务的“预提交”；等到检查点保存完毕，才会提交事务进行“正式提交”。如果中间出现故障，事务进行回滚，预提交就会被放弃；恢复状态之后，也只能恢复所有已经确认提交的操作。
2. 具体步骤
为了方便说明，我们来考虑一个具体的流处理系统，由Flink从Kafka读取数据、并将处理结果写入Kafka，如图所示。
这是一个Flink与Kafka构建的完整数据管道，Source任务从Kafka读取数据，经过一系列处理（比如窗口计算），然后由Sink任务将结果再写入Kafka。Flink与Kafka连接的两阶段提交，离不开检查点的配合，这个过程需要JobManager协调各个TaskManager进行状态快照，而检查点具体存储位置则是由状态后端（State Backend）来配置管理的。一般情况，我们会将检查点存储到分布式文件系统上。实现端到端exactly-once的具体过程可以分解如下：
（1）启动检查点保存
检查点保存的启动，标志着我们进入了两阶段提交协议的“预提交”阶段。当然，现在还没有具体提交的数据。
如图所示，JobManager通知各个TaskManager启动检查点保存，Source任务会将检查点分界线（barrier）注入数据流。这个barrier可以将数据流中的数据，分为进入当前检查点的集合和进入下一个检查点的集合。
（2）算子任务对状态做快照
分界线（barrier）会在算子间传递下去。每个算子收到barrier时，会将当前的状态做个快照，保存到状态后端。
如图所示，Source任务将barrier插入数据流后，也会将当前读取数据的偏移量作为状态写入检查点，存入状态后端；然后把barrier向下游传递，自己就可以继续读取数据了。接下来barrier传递到了内部的Window算子，它同样会对自己的状态进行快照保存，写入远程的持久化存储。
（3）Sink任务开启事务，进行预提交
如图所示，分界线（barrier）终于传到了Sink任务，这时Sink任务会开启一个事务。接下来到来的所有数据，Sink任务都会通过这个事务来写入Kafka。这里barrier是检查点的分界线，也是事务的分界线。由于之前的检查点可能尚未完成，因此上一个事务也可能尚未提交；此时barrier的到来开启了新的事务，上一个事务尽管可能没有被提交，但也不再接收新的数据了。对于Kafka而言，提交的数据会被标记为“未确认”（uncommitted）。这个过程就是所谓的“预提交”（pre-commit）。
（4）检查点保存完成，提交事务
当所有算子的快照都完成，也就是这次的检查点保存最终完成时，JobManager会向所有任务发确认通知，告诉大家当前检查点已成功保存，如图所示。
当Sink任务收到确认通知后，就会正式提交之前的事务，把之前“未确认”的数据标为“已确认”，接下来就可以正常消费了。在任务运行中的任何阶段失败，都会从上一次的状态恢复，所有没有正式提交的数据也会回滚。这样，Flink和Kafka连接构成的流处理系统，就实现了端到端的exactly-once状态一致性。
3. 需要的配置
在具体应用中，实现真正的端到端exactly-once，还需要有一些额外的配置：
（1）必须启用检查点（2）在FlinkKafkaProducer的构造函数中传入参数Semantic.EXACTLY_ONCE（3）配置Kafka读取数据的消费者的隔离级别这里所说的Kafka，是写入的外部系统。预提交阶段数据已经写入，只是被标记为“未提交”（uncommitted），而Kafka中默认的隔离级别isolation.level是read_uncommitted，也就是可以读取未提交的数据。这样一来，外部应用就可以直接消费未提交的数据，对于事务性的保证就失效了。所以应该将隔离级别配置为read_committed，表示消费者遇到未提交的消息时，会停止从分区中消费数据，直到消息被标记为已提交才会再次恢复消费。当然，这样做的话，外部应用消费数据就会有显著的延迟。（4）事务超时配置Flink的Kafka连接器中配置的事务超时时间transaction.timeout.ms默认是1小时，而Kafka集群配置的事务最大超时时间transaction.max.timeout.ms默认是15分钟。所以在检查点保存时间很长时，有可能出现Kafka已经认为事务超时了，丢弃了预提交的数据；而Sink任务认为还可以继续等待。如果接下来检查点保存成功，发生故障后回滚到这个检查点的状态，这部分数据就被真正丢掉了。所以这两个超时时间，前者应该小于等于后者。
