一、检查点

Checkpoint

1. 介绍

• 有状态流应用中的检查点（checkpoint），其实就是所有任务的状态在某个时间点的一个快照（一份拷贝），这个时间点应该是所有任务都恰好处理完一个相同的输入数据的时刻。
• 在一个流应用程序运行时， Flink 会定期保存检查点，在检查点中会记录每个算子的 id 和状态；如果发生故障，Flink 就会用最近一次成功保存的检查点来恢复应用的状态，重新启动处理流程，就如同“读档”一样。
• 检查点是 Flink 容错机制的核心。这里的“检查”是指故障恢复之后继续处理的结果，应该与发生故障前完全一致，所以需要“检查”结果的正确性。因此又会把 checkpoint 叫作“一致性检查点”。

2. 检查点保存

• 在 Flink 中，检查点的保存是周期性触发的，间隔时间可以进行设置，当每隔一段时间检查点保存操作被触发时，就把每个任务当前的状态复制一份，按照一定的逻辑结构放在一起持久化保存起来，就构成了检查点。

• 当检查点保存操作被触发时，保存状态的时机是当所有任务都恰好处理完一个相同的输入数据的时候。首先，这样避免了除状态之外其他额外信息的存储，提高了检查点保存的效率。其次，一个数据要么就是被所有任务完整地处理完，状态得到了保存；要么就是没处理完，状态全部没保存，相当于构建了一个“事务”(transaction。

• 以 WordCount 为例，具体的检查点保存流程为：源（Source）任务从外部数据源读取数据，并记录当前的偏移量，作为算子状态（Operator State）保存下来；Sum 算子会把当前求和的结果作为按键分区状态（Keyed State）保存下来；当触发了检查点保存时，所有任务都已经处理了 3 条数据 (“hello”,“world”,“hello”)，此时就会将 Source 算子的偏移量 3 和 sum 算子的状态 (“hello” -> 2，“world” -> 1) 保存成一个检查点，写入外部存储中。具体的存储系统由状态后端的配置项 “检查点存储” (CheckpointStorage) 来决定的，有作业管理器的堆内存（JobManagerCheckpointStorage）和文件系统（FileSystemCheckpointStorage）两种选择。一般情况下，会将检查点写入持久化的分布式文件系统。

  

3. 检查点故障恢复

• 当 Flink 任务发生故障时，会使用最近的检查点来一致恢复应用程序的状态，并重新启动处理流程，发生故障时正在处理的所有数据都需要重新处理，所以需要让源（source）任务向数据源重新提交偏移量、请求重放数据，因此需要外部数据源能够重置偏移量，如 Kafka。

• 以 WordCount 为例，具体的检查点故障恢复流程为：

  • 程序之前在处理完 (“hello”,“world”,“hello”) 三个数据后保存了一个检查点，之后又正常处理了一个数据 (“flink”)，但在处理第五个数据 (“hello”) 时发生了故障，此时 Source 任务已经处理完毕且偏移量为 5，Map 任务也处理完成，而 sum 任务在处理中发生了故障，此时状态并未保存

    

  • 第一步是重启发生故障的应用，此时所有任务的状态会清空

    

  • 第二步是找到最近一次保存的检查点，从中读出每个算子任务状态的快照，分别填充到对应的状态中。此时，Flink 内部所有任务的状态，都恢复到了保存检查点的那一时刻，即刚好处理完 (“hello”,“world”,“hello”) 三个数据的时候；注意，虽然故障发生前 (“flink”) 数据已经处理完，但由于没保存在检查点中，所以恢复后的任务中并没有 (“flink”) 的处理结果

    

  • 第三步是通过 Source 任务向外部数据源重新提交偏移量（offset）来实现从保存检查点后开始重新读取数据，由于从检查点恢复状态时之前所有在检查点保存后到发生故障前的这段时间内的数据 (“flink”,“hello”) 都丢失了，为了重新处理这些丢失的数据，必须从数据源重放这些数据。

    

  • 第四步是重放第 4、5 个数据后，然后继续读取后面的数据正常处理，此时，既没有丢掉数据也没有重复计算数据，这就保证了计算结果的正确性。在分布式系统中，这叫作实现了“精确一次”（exactly-once）的状态一致性保证。

    

4. 检查点算法

Flink 采用了基于 Chandy-Lamport 算法的分布式快照保存检查点

4.1 Checkpoint Barrier

检查点分界线或检查点屏障

• 类似于 watermark，Checkpoint Barrier 是一种在数据流中插入的特殊数据结构，用来表示触发检查点保存的时间点，会把一条流上的数据按照不同的检查点分隔开。
• Checkpoint Barrier 是由 JobManager 生成的在 Source 算子中注入到常规数据流中，它的位置是限定好的， 不能超过其他数据，也不能被后面的数据超过。检查点分界线中带有一个检查点 ID，这是当前要保存的检查点的唯一标识。
• Checkpoint Barrier 能够保证在不暂停流处理的前提下，让每个任务“认出”触发检查点保存的那个数据，当一个任务遇到一个 Checkpoint Barrier 时就开始对状态做持久化快照保存。由于数据流是保持顺序依次处理的，因此遇到这个标识就代表之前的数据都处理完了，可以保存一个检查点；而在这个标识之后的数据引起的状态改变就不会体现在这个检查点中，而需要保存到下一个检查点中

4.2 具体算法过程

• 案例：一个有两个输入流的应用程序，用并行的两个 Source 任务来读取自然数，然后按照奇偶数进行分组后进行 sum 操作，最终输出结果

  

• 在 JobManager 中有一个“检查点协调器”（checkpoint coordinator），专门用来协调处理检查点的相关工作。检查点协调器会周期性地向每个 TaskManager 发送一条带有新检查点 ID 的消息，通过这种方式来启动检查点

  

• 收到指令后，TaskManager 会将带有检查点 ID 的分界线（barrier）插入到当前的数据流中并向下游传递，同时让所有的 Source 任务把自己的偏移量（算子状态）保存起来，状态后端在状态存入检查点之后，会返回通知给 source 任务，source 任务就会向 JobManager 确认检查点完成

  

• 带有检查点 ID 的分界线（barrier）会被上游的 Source 任务广播给下游所有的 Sum 分区任务，sum 任务会等待所有上游输入分区的 barrier 到达，即分界线对齐；对于 barrier 已经到达的分区，在 barrier 之后继续到达的数据会被缓存；对于 barrier 尚未到达的分区，其所有到达数据会被继续正常 sum 处理，直到该分区的 barrier 到达

  

• 当一个 sum 任务收到所有上游输入分区的 barrier 时，该任务就将其结果状态保存到状态后端的检查点中，然后将 barrier 继续向下游转发

  

• 该任务向下游转发检查点 barrier 后，会继续处理之前缓存或到达的数据

  

• Sink 任务向 JobManager 确认状态保存到 checkpoint 完毕，当所有任务都确认已成功将状态保存到检查点时，检查点就真正完成了

  

5. 保存点

• 除了检查点（checkpoint）外，Flink 提供了另一个非常独特的镜像保存功能，即保存点（Savepoint），原则上，创建保存点使用的算法与检查点完全相同，因此保存点可以认为就是具有一些额外元数据的检查点
• 保存点与检查点最大的区别是触发的时机。检查点是由 Flink 自动管理的，定期创建，发生故障之后自动读取进行恢复，这是一个“自动存盘”的功能；而保存点不会自动创建，必须由用户明确地手动触发保存操作，所以是“手动存盘”
• 检查点主要用来做故障恢复，是容错机制的核心；而保存点功能很强大，除了故障恢复外，还可以用于有计划的手动备份，更新应用程序，版本迁移，暂停和重启应用等

6. 检查点和重启策略配置

6.1 检查点配置

为了防止 Flink 任务在保存检查点时占据大量时间导致数据处理性能降低，可以在代码中对检查点进行相关配置

public class TestFlinkCheckpointSet {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //1.开启checkpoint功能，checkpoint默认是关闭的
        //env.enableCheckpointing(); //已过时，默认500毫秒间隔保存一次检查点
        env.enableCheckpointing(1000L); //推荐方法，参数为Long类型的毫秒数，表示保存检查点间隔时间
        
        //2.配置检查点存储
        //需要传入一个CheckpointStorage的实现类
        //2.1 配置存储检查点到 JobManager 堆内存
        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        
        //2.2 配置存储检查点到文件系统
		env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));
        
        //3.其他高级配置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //3.1 设置检查点一致性的保证级别
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//精确一次模式
        //checkpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);//至少一次模式
        
        //3.2 配置检查点保存的超时时间，超时没完成就会被丢弃掉
        checkpointConfig.setCheckpointTimeout(60000L); //传入Long型毫秒数
        
        //3.3 配置最大并发检查点数量
        //指定运行中的检查点最多可以有多少个。由于每个任务的处理进度不同，完全可能出现后面的任务还没完成前一个检查点的保存、前面任务已经开始保存下一个检查点了。这个参数就是限制同时进行的最大数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        
        //3.4 配置两次检查点保存的最小间隔时间
        //指定在上一个检查点完成之后，检查点协调器最快等多久可以出发保存下一个检查点的指令。即使已经达到了触发检查点保存的周期间隔时间点，只要距离上一个检查点完成的时间小于最小间隔时间，就依然不能开启下一次检查点的保存。这为正常处理数据留下了充足的间隙。
        //指定了最小间隔时间，则 maxConcurrentCheckpoints 的值强制为 1
        checkpointConfig.setMinPauseBetweenCheckpoints(500L); //传入Long型毫秒数
        
        //3.5 配置是否只从检查点恢复
        //默认值是 false，即从最近的检查点或保存点恢复
        checkpointConfig.setPreferCheckpointForRecovery(true); //即使有最近的保存点，也只从检查点恢复
        
        //3.6 配置容许检查点保存的失败次数
        checkpointConfig.setTolerableCheckpointFailureNumber(0);
        
        
        env.execute();
    }
}

6.2 重启策略配置

public class TestRestartStrategy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //1.固定延迟重启
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
            3,  //尝试重启次数
            100000L //尝试重启的时间间隔
		));
        
        //2.失败率重启
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
        	3, //失败次数
            Time.minutes(10), //重启时间范围
            Time.minutes(1) //重启时间间隔
        ));
        
        env.execute();
    }
}

二、状态一致性

1. 一致性的概念

​ 对于 Flink 流处理器内部来说，所谓的状态一致性，其实就是计算结果要保证准确。在流式计算正常的处理过程中，由于数据是来一个计算处理一个，所以结果肯定是正确的；但当发生故障后需要恢复状态进行回滚时也要保证发生故障后的数据既不丢失也不被重复计算，恢复以后的重新计算，结果应该也是完全正确的。

2. 一致性的级别

• 最多一次（AT-MOST-ONCE）： 当任务发生故障时，直接重启，既不恢复丢失的状态，也不重放丢失的数据。每个数据在正常情况下会被处理一次，遇到故障时就会丢掉，数据“最多被处理一次”。这种级别下没法保证结果的准确性，所以这种类型的保证也叫“没有保证”。如果主要的诉求是“快”，且可以接受近似正确的结果，则可以选择这种级别。
• 至少一次（AT-LEAST-ONCE）：在这种级别下所有的数据都不会丢失，都会被处理，但不能保证只处理一次，有些数据会被重复处理，所以数据“至少被处理一次”。在有些场景下，重复处理数据是不影响结果的正确性的，这种操作具有“幂等性”。比如，统计电商网站的 UV，需要对每个用户的访问数据进行去重处理，所以即使同一个数据被处理多次，也不会影响最终的结果，这时可以使用 at-least-once 级别。
• 精确一次（EXACTLY-ONCE）：最严格的一致性保证，也是最难实现的状态一致性语义。exactly-once 意味着所有数据不仅不会丢失，而且只被处理一次，不会重复处理。exactly-once 可以真正意义上保证结果的绝对正确，在发生故障恢复后，就好像从未发生过故障一样。要做的 exactly-once，首先必须能达到 at-least-once 的要求，就是数据不丢。所以同样需要有数据重放机制来保证这一点。另外，还需要有专门的设计保证每个数据只被处理一次。在 Flink 中使用的是检查点（checkpoint）来保证 exactly-once 语义。

3. 端到端的状态一致性

• 一个完整的流处理应用包括了数据源、流处理器和外部存储系统三个部分。这个完整应用的一致性，就叫作“端到端（end-to-end）的状态一致性”。
• 检查点只可以保证 Flink 内部流处理器的状态一致性，而且可以做到精确一次（exactly-once）级别；但整个端到端的一致性级别取决于所有组件中一致性最弱的组件级别。
• 一般来说，能否达到 at-least-once 一致性级别，主要看数据源是否能够重放数据；而能否达到 exactly-once 级别，则流处理器内部、数据源和外部存储都要有相应的保证机制。

4. 端到端精确一次保证

4.1 输入端保证

• 输入端主要指的就是 Flink 读取的外部数据源。对于一些数据源来说，并不提供数据的缓冲或是持久化保存，数据被消费之后就彻底不存在了，由于数据不能重发了，这就会导致数据丢失。所以就只能保证 at-most-once 的一致性语义
• 想要在故障恢复后不丢数据，外部数据源就必须拥有重放数据的能力。常见的做法就是对数据进行持久化保存，并且可以重设数据的读取位置。一个最经典的应用就是 Kafka。在 Flink 的 Source 任务中将数据读取的偏移量保存为状态，这样就可以在故障恢复时从检查点中读取出来，对数据源重置偏移量，重新获取数据。
• 数据源可重放数据，或者说可重置读取数据偏移量，加上 Flink 的 Source 算子将偏移量作为状态保存进检查点，就可以保证数据不丢。这是达到 at-least-once 一致性语义的基本要求，当然也是实现端到端 exactly-once 的基本要求。

4.2 流处理器保证

​ 对于 Flink 内部流处理器来说，检查点机制可以保证故障恢复后数据不丢（在能够重放的前提下），并且只处理一次，所以已经可以做到 exactly-once 的一致性语义了。

4.3 输出端保证

Flink 的 checkpoint 加上可重放数据的输入数据源就可以保证 At_Least_Once 级别一致性，但最终要实现 Exactly_Once 一致性则还需要保证数据不能重复输出写入到外部系统

4.3.1 幂等写入

• 幂等操作是指一个操作可以重复执行很多次，但只导致一次结果更改
• 幂等写入并没有解决 Sink 重复写入的问题，但由于重复写入只会导致结果发生一次更改，所以最终结果的一致性得到了保证
• 幂等写入的主要限制在于外部存储系统必须支持这样的幂等写入，比如 Redis 中键值存储，或者关系型数据库（如 MySQL）中满足查询条件的更新操作
• 幂等写入在故障进行恢复时，因为保存点完成之后到发生故障之间的数据，其实已经写入了一遍，回滚的时候并不能消除它们。因此在输入数据源进行数据重放时在短时间内输出的外部系统结果会突然“跳回”到之前的某个值，然后“重播”一段之前的数据。但当数据的重放逐渐超过发生故障的点的时候，最终的结果还是一致的

4.3.2 事务写入

• 事务（transaction）是应用程序中一系列严密的操作，所有操作必须成功完成，否则在每个操作中所做的所有更改都会被撤消
• 事务写入的基本思想：构建一个事务来进行数据向外部系统的写入，这个事务是与检查点绑定在一起的。当 Sink 任务遇到 barrier 时，开始检查点保存的同时就开启一个事务，接下来所有数据向外部系统的写入都在这个事务中进行，等到当前检查点保存完毕时，将事务进行提交，所有写入的数据就真正可用了。如果中间过程出现故障，状态会回退到上一个检查点，而当前事务因为没有正常关闭（因为当前检查点没有保存完），所以也会回滚，写入到外部的数据就被撤销了
• 事务写入实现方式：
  • 预写日志（Write-Ahead-Log，WAL）：不需要外部系统支持事务；DataStream API 提供了一个模板类 GenericWriteAheadSink 用来实现这种事务型的写入方式；可能存在写入丢失的情况
    • 先把结果数据作为日志（log）状态保存起来
    • 进行检查点保存时，会将这些结果数据一并做持久化存储
    • 在收到检查点完成的通知时，将所有结果一次性写入外部系统
  • 两阶段提交（Two-Phase-Commit，2PC）：Flink 提供了 TwoPhaseCommitSinkFunction 接口来自定义实现两阶段提交的 SinkFunction
    • 当第一条数据到来时，或者收到检查点的分界线时，Sink 任务都会启动一个事务
    • 接下来接收到的所有数据，都通过这个事务写入外部系统；这时由于事务没有提交，所以数据尽管写入了外部系统，但是不可用，是“预提交”的状态
    • 当 Sink 任务收到 JobManager 发来检查点完成的通知时，正式提交事务，写入的结果就真正可用了
• 2PC 方式对 Sink 外部系统的要求：
  • 外部系统必须提供事务支持，或者 Sink 任务必须能够模拟外部系统上的事务
  • 在检查点的间隔期间里，必须能够开启一个事务并接受数据写入
  • 在收到检查点完成的通知之前，事务必须是“等待提交”的状态。在故障恢复的情况下，这可能需要一些时间。如果这个时候外部系统关闭事务（例如超时了），那么未提交的数据就会丢失
  • Sink 任务必须能够在进程失败后恢复事务
  • 提交事务必须是幂等操作，事务的重复提交是无效的

4.4 不同 Source 和 Sink 的一致性组合

	不可重放 Source	可重放 Source
任意 Sink	At-most-once	At-least-once
幂等写入 Sink	At-most-once	Exactly-once（故障恢复时会出现暂时不一致）
预写入日志 Sink	At-most-once	At-least-once
两阶段提交 Sink	At-most-once	Exactly-once

5. Flink + Kafka 实现端到端精确一次

5.1 整体架构

• Flink 内部：通过检查点机制保证状态和处理结果的 exactly-once 语义，检查点具体存储位置则是由状态后端（State Backend）来配置管理的
• 输入端：数据源 Kafka 可以对数据进行持久化保存，并可以重置偏移量（offset）。在 Source 任务中使用 FlinkKafkaConsumer 将当前读取的偏移量保存为算子状态，写入到检查点中；当发生故障时，从检查点中读取恢复状态，并由连接器 FlinkKafkaConsumer 向 Kafka 重新提交偏移量，就可以重新消费数据、保证结果的一致性
• 输出端：Flink 官方实现的 Kafka 连接器中提供了写入数据到 Kafka 的 FlinkKafkaProducer 类，这个类实现了 TwoPhaseCommitSinkFunction 接口，所以从 Flink 写入 Kafka 的过程实际上是一个两阶段提交：处理完毕得到结果，写入 Kafka 时是基于事务的“预提交”；等到检查点保存完毕，才会提交事务进行“正式提交”。如果中间出现故障，事务进行回滚，预提交就会被放弃；恢复状态之后，也只能恢复所有已经确认提交的操作

5.2 具体过程

• 在Flink 流式处理程序中，一条数据由 Source 从数据源读取，经过 transform 操作过程，最终由 Sink 写入外部存储系统，当第一条数据从 Sink 写入 Kafka 时就会开启一个 kafka 的事务（transaction1），正常写入 Kafka 分区日志但标记为“未确认”（uncommitted），这就是“预提交”

  

• jobmanager 触发 checkpoint 操作，会将检查点分界线（barrier1）注入数据流从 source 开始向下传递

  

• Source 任务会将当前读取数据的偏移量作为状态写入检查点，存入状态后端；每个内部的 transform 任务遇到 barrier 时，都会把状态存到 checkpoint 里并告知 JobManager

  

• Sink 任务收到 barrier1 后，首先会保存自己当前的状态，存入 checkpoint 并通知 jobmanager，同时开启下一阶段的 kafka 的事务（transaction2），用于提交下一个检查点的数据，此时由于 barrier1 的检查点保存还未完成，所以上一个 Kafka 事务（transaction1）还未提交关闭

  

• 当 JobManager 收到所有算子任务完成 barrier1 的 checkpoint 保存时，JobManager 会向所有任务发通知确认这次 checkpoint 的完成，Sink 任务收到确认通知后，正式提交 transaction1 事务中的数据，kafka 关闭 transaction1 事务并将分区中未确认数据改为“已确认“，数据可以被正常消费

5.3 需要的配置

• 必须启用检查点
• 在 FlinkKafkaProducer 的构造函数中传入参数 Semantic.EXACTLY_ONCE
• 配置 Kafka 读取数据的消费者的隔离级别，这里的 Kafka，是写入的外部系统。预提交阶段数据已经写入，只是被标记为“未提交”（uncommitted），而 Kafka 中默认的隔离级别 isolation.level 是 read_uncommitted，所以消费者可以读取未提交的数据。这样对于事务性的保证就失效了。所以应该将隔离级别配置为 read_committed，表示消费者遇到未提交的消息时，会停止从分区中消费数据，直到消息被标记为已提交才会再次恢复消费。当然，这样做的话，外部应用消费数据就会有显著的延迟
• 事务超时配置：Flink 的 Kafka 连接器中配置的事务超时时间 transaction.timeout.ms 默认是 1 小时，而 Kafka 集群配置的事务最大超时时间 transaction.max.timeout.ms 默认是 15 分钟。所以在检查点保存时间很长时，有可能出现 Kafka 已经认为事务超时了，丢弃了预提交的数据；而 Sink 任务认为还可以继续等待。如果接下来检查点保存成功，发生故障后回滚到这个检查点的状态，这部分数据就被真正丢掉了。所以配置这两个超时时间，前者应该小于等于后者。