Flink 的容错机制

1.概述

介绍 Flink 中的容错机制是如何保证故障后的恢复。

2. 检查点

将某个时间点之前所有状态保存下来，这份存档就是检查点（checkpoint）。遇到故障重启的时候，从检查点中“读档”，可以恢复出之前的状态。

2.1 检查点保存的流程以及从检查点恢复状态

检查点的保存，最关键的就是要等所有任务将同一个数据处理完毕。当所有任务处理完同一条数据后，对状态做个快照保存下来。例如输入数据为：“hello”，“world”，“hello”，“flink”，“hello”，“world”，“hello”，“flink”...。所以每个任务都处理完“hello”后，保存状态，标记此处source 偏移量是3。

当发生故障时，就需要找到最近一次成功保存的检查点来恢复状态。最近检查点就是 source 偏移量是3的一次快照。

接下来就是需要从检查点恢复状态。

（1）重启应用，读取检查点（source 偏移量为3），恢复状态

（2）重置偏移量

为了不丢数据，应该从保存检查点后开始重新读取数据，可以通过 Source 任务向外部数据源重新提交偏移量（offset，source 偏移量为3）来实现。

（3）继续处理数据（开始处理source 偏移量为4的数据）

2.2 检查点算法

2.2.1 检查点分界线

在数据流中插入一个特殊的数据结构，专门用来表示触发检查点保存的时间点。收到保存检查点的指令后，Source任务可以在当前数据流中插入这个结构；之后的所有任务只要遇到它就开始对状态做持久化快照保存。由于数据流是保持顺序依次处理的，因此遇到这个标识就代表之前的数据都处理完了，可以保存一个检查点；而在它之后的数据，引起的状态改变就不会体现在这个检查点中，而需要保存到下一个检查点。这种特殊的数据形式，把一条流上的数据按照不同的检查点分隔开，所以就叫做检查点的“分界线”（Checkpoint Barrier）。

2.2.2 分布式快照算法

1）Barrier 对齐的精准一次

（1）触发检查点保存

JobManager 中的检查点协调器向 TaskManager 发动指令，触发检查点的保存；Source 任务插入一个分界线，并将偏移量保存到远程的持久化存储中。

（2）状态快照保存完成，分界线向下游传递

状态存入持久化存储后，会返回通知给 Source 任务；Source 任务就会向 JobManager 确认检查点完成，然后分界线跟随数据向下游传递。由于 Map 算子没有状态，所以进行 keyBy 分区。

（3）向下游多个并行子任务广播分界线，执行分界线对齐

sum 任务需要接收到上游 Map 广播的所有分界线后才能保存状态，所以需要执行分界线对齐。

sum 任务接收到 map1 的分界线，如果map1 继续有数据传递，不应立即处理，而是要缓存起来，等接收到 map2 的分界线和保存 sum 状态之后再做处理。

（3）分界线对齐后，保存状态到持久化存储

2）Barrier 对齐的至少一次

和 Barrier 对齐的精准一次相比，差别是从开始分界线对齐到分界线对齐完成这段时间，从已经广播过分界线的上游子任务传递过来的数据，是否纳入算子计算和存入状态。如果从已经广播分界线的上游传递过来的数据，不存入状态，则是 Barrier 对齐的精准一次，否则就是 Barrier 对齐的至少一次。对于 Barrier 对齐的至少一次，如果重新启动加载检查点，这时这块数据就会被再次计算。

3）非 Barrier 对齐的精准一次

和 Barrier 对齐的精准一次相比，差别是从开始分界线对齐到分界线对齐完成这段时间，接收到第一个上游子任务传递过来的分界线，就将状态缓存起来。然后对未广播过分界线的上游子任务传递过来的数据也进行缓存，直到完成分界线对齐，再将缓存的数据存入持久化存储。

2.3 检查点配置

检查点的作用是为了故障恢复，不能因为保存检查点占据了大量时间、导致数据处理性能明显降低。为了兼顾容错性和处理性能，就需要在代码中对检查点进行各种配置。

2.3.1 启动检查点

默认情况下，Flink程序是禁用检查点的。如果想要为Flink应用开启自动保存快照的功能，需要在代码中显式地调用执行环境的.enableCheckpointing()方法：

env.enableCheckpointing(1000); // 每隔 1 秒启动执行一次检查点保存。检查点的间隔时间是对处理性能和故障恢复速度的一个权衡。

2.3.2 检查点存储

env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));

2.3.3 其他高级配置

CheckpointConfig checkpointConfig = env.getCheckpointConfig();

1）常用高级配置

检查点模式（CheckpointingMode）：设置检查点一致性的保证级别，有“精确一次”（exactly-once）和“至少一次”（at-least-once）两个选项，默认级别为exactly-once。

超时时间（checkpointTimeout）：用于指定检查点保存的超时时间，超时没完成就会被丢弃掉。传入一个长整型毫秒数作为参数，表示超时时间。

最小间隔时间（minPauseBetweenCheckpoints）：用于指定在上一个检查点完成之后，检查点协调器最快等多久可以出发保存下一个检查点的指令。这就意味着即使已经达到了周期触发的时间点，只要距离上一个检查点完成的间隔不够，就依然不能开启下一次检查点的保存。这就为正常处理数据留下了充足的间隙。当指定这个参数时，实际并发为1。

最大并发检查点数量（maxConcurrentCheckpoints）：用于指定运行中的检查点最多可以有多少个。由于每个任务的处理进度不同，完全可能出现后面的任务还没完成前一个检查点的保存、前面任务已经开始保存下一个检查点了。这个参数就是限制同时进行的最大数量。

开启外部持久化存储（enableExternalizedCheckpoints）：用于开启检查点的外部持久化，而且默认在作业失败的时候不会自动清理，如果想释放空间需要自己手工清理。里面传入的参数 ExternalizedCheckpointCleanup 指定了当作业取消的时候外部的检查点该如何清理。

DELETE_ON_CANCELLATION：在作业取消的时候会自动删除外部检查点，但是如果是作业失败退出，则会保留检查点。

RETAIN_ON_CANCELLATION：作业取消的时候也会保留外部检查点。

检查点连续失败次数（tolerableCheckpointFailureNumber）：用于指定检查点连续失败的次数，当达到这个次数，作业就失败退出。默认为0，这意味着不能容忍检查点失败，并且作业将在第一次报告检查点失败时失败。

2）开启非对齐检查点

非对齐检查点（enableUnalignedCheckpoints）：不再执行检查点的分界线对齐操作，启用之后可以大大减少检查点对性能的影响。这个设置要求检查点模式（CheckpointingMode）必须为exctly-once，并且最大并发的检查点个数为1。

对齐检查点超时时间（alignedCheckpointTimeout）：该参数只有在启用非对齐检查点的时候有效。参数默认是0，表示一开始就直接用非对齐检查点。如果设置大于0，一开始会使用对齐的检查点，当对齐时间超过该参数设定的时间，则会自动切换成非对齐检查点。

2.3.4 通用增量 checkpoint （changelog）

不同于产生一个包含所有数据的全量备份，增量快照中只包含自上一次快照完成之后被修改的记录，因此可以显著减少快照完成的耗时。

1）执行过程

（1）带状态的算子任务将状态更改写入变更日志（记录状态）

（2）状态物化：状态表定期保存，独立于检查点

（3）状态物化完成后，状态变更日志就可以被截断到相应的点

2）使用方式

（1）方式一：配置文件指定

state.backend.changelog.enabled: true

state.backend.changelog.storage: filesystem

# 存储 changelog 数据

dstl.dfs.base-path: hdfs://hadoop102:8020/changelog

execution.checkpointing.max-concurrent-checkpoints: 1

execution.savepoint-restore-mode: CLAIM

（2）在代码中设置

env.enableChangelogStateBackend(true);

3）优缺点对比

4）总结

2.4 保存点（Savepoint）

保存点与检查点最大的区别，就是触发的时机。检查点是由 Flink 自动管理的，定期创建，发生故障之后自动读取进行恢复，这是一个“自动存盘”的功能；而保存点不会自动创建，必须由用户明确地手动触发保存操作，所以就是“手动存盘”。可以在需要的时候创建一个保存点，然后停止应用，做一些处理调整之后再从保存点重启。需要注意的是，保存点能够在程序更改的时候依然兼容，前提是状态的拓扑结构和数据类型不变。保存点中状态都是以算子ID-状态名称这样的key-value组织起来的，算子ID可以在代码中直接调用SingleOutputStreamOperator的.uid()方法来进行指定。为了方便后续的维护，强烈建议在程序中为每一个算子手动指定ID。

它适用的具体场景有：

• 版本管理和归档存储
• 更新Flink版本
• 更新应用程序
• 调整并行度
• 暂停应用程序

2.4.1 使用保存点

（1）创建保存点

命令行为运行的作业创建一个保存点镜像：bin/flink savepoint :jobId [:targetDirectory]

通过配置文件flink-conf.yaml 设置保存点的默认路径：state.savepoints.dir: hdfs:///flink/savepoints

程序代码中通过执行环境来设置：env.setDefaultSavepointDir("hdfs:///flink/savepoints");

停掉一个作业时直接创建保存点：bin/flink stop --savepointPath [:targetDirectory] :jobId

（2）从保存点重启应用

bin/flink run -s :savepointPath [:runArgs]（如果是基于yarn的运行模式还需要加上 -yid application-id）

2.4.2 使用保存点切换状态后端

（1）提交 flink 作业

bin/flink run-application -d -t yarn-application -Dstate.backend=hashmap -c com.atguigu.checkpoint.SavepointDemo FlinkTutorial-1.0-SNAPSHOT.jar

（2）停止flink作业时，触发保存点

方式一：stop优雅停止并触发保存点，要求source实现StoppableFunction接口

bin/flink stop -p savepoint路径 job-id -yid application-id

方式二：cancel立即停止并触发保存点

bin/flink cancel -s savepoint路径 job-id -yid application-id

（3）从savepoint恢复作业，同时修改状态后端

bin/flink run-application -d -t yarn-application -s hdfs://hadoop102:8020/sp/savepoint-267cc0-47a214b019d5 -Dstate.backend=rocksdb -c com.atguigu.checkpoint.SavepointDemo FlinkTutorial-1.0-SNAPSHOT.jar

（4）从保存下来的checkpoint恢复作业

bin/flink run-application -d -t yarn-application -Dstate.backend=rocksdb -s hdfs://hadoop102:8020/chk/532f87ef4146b2a2968a1c137d33d4a6/chk-175 -c com.atguigu.checkpoint.SavepointDemo ./FlinkTutorial-1.0-SNAPSHOT.jar

3. 端到端精确一致性

3.1 输入端保证

想要在故障恢复后不丢数据，外部数据源就必须拥有重放数据的能力。常见的做法就是对数据进行持久化保存，并且可以重设数据的读取位置。一个最经典的应用就是Kafka。在Flink的Source任务中将数据读取的偏移量保存为状态，这样就可以在故障恢复时从检查点中读取出来，对数据源重置偏移量，重新获取数据。

3.2 输出端保证

为了实现端到端 exactly-once，需要对外部存储系统、以及Sink连接器有额外的要求。能够保证 exactly-once 一致性的写入方式有两种：

• 幂等写入
• 事务写入

（1）幂等（Idempotent）写入

所谓“幂等”操作，就是说一个操作可以重复执行很多次，但只导致一次结果更改。也就是说，后面再重复执行就不会对结果起作用了。

（2）事务（Transactional）写入

事务写入的基本思想就是：用一个事务来进行数据向外部系统的写入，这个事务是与检查点绑定在一起的。当 Sink 任务遇到 barrier 时，开始保存状态的同时就开启一个事务，接下来所有数据的写入都在这个事务中；待到当前检查点保存完毕时，将事务提交，所有写入的数据就真正可用了。如果中间过程出现故障，状态会回退到上一个检查点，而当前事务没有正常关闭（因为当前检查点没有保存完），所以也会回滚，写入到外部的数据就被撤销了。

具体来说，又有两种实现方式：预写日志（WAL）和两阶段提交（2PC）

1）预写日志（write-ahead-log，WAL）

①先把结果数据作为日志（log）状态保存起来

②进行检查点保存时，也会将这些结果数据一并做持久化存储

③在收到检查点完成的通知时，将所有结果一次性写入外部系统

④在成功写入所有数据后，在内部再次确认相应的检查点，将确认信息也进行持久化保存，这才代表着检查点的真正完成

在 Flink中DataStream API提供了一个模板类 GenericWriteAheadSink，用来实现这种事务型的写入方式。

2）两阶段提交（two-phase-commit，2PC）

先做“预提交”，等检查点完成之后再正式提交。这种提交方式是真正基于事务的，它需要外部系统提供事务支持。

①当第一条数据到来时，或者收到检查点的分界线时，Sink任务都会启动一个事务。

②接下来接收到的所有数据，都通过这个事务写入外部系统；这时由于事务没有提交，所以数据尽管写入了外部系统，但是不可用，是“预提交”的状态。

③当 Sink 任务收到 JobManager 发来检查点完成的通知时，正式提交事务，写入的结果就真正可用了。

当中间发生故障时，当前未提交的事务就会回滚，于是所有写入外部系统的数据也就实现了撤回。这种两阶段提交（2PC）的方式充分利用了Flink现有的检查点机制：分界线的到来，就标志着开始一个新事务；而收到来自JobManager的checkpoint成功的消息，就是提交事务的指令。每个结果数据的写入，依然是流式的，不再有预写日志时批处理的性能问题；最终提交时，也只需要额外发送一个确认信息。所以2PC协议不仅真正意义上实现了exactly-once，而且通过搭载Flink的检查点机制来实现事务，只给系统增加了很少的开销。

Flink提供了 TwoPhaseCommitSinkFunction 接口，方便我们自定义实现两阶段提交的 SinkFunction 的实现，提供了真正端到端的 exactly-once 保证。新的Sink架构，使用的是TwoPhaseCommittingSink接口。

3.3 Flink 和 Kafka 连接时的精确一次保证

（1）Flink内部

Flink 内部可以通过检查点机制保证状态和处理结果的 exactly-once 语义。

（2）输入端

输入数据源端的 Kafka 可以对数据进行持久化保存，并可以重置偏移量（offset）。在 Source 任务（FlinkKafkaConsumer）中将当前读取的偏移量保存为算子状态，写入到检查点中；当发生故障时，从检查点中读取恢复状态，并由连接器 FlinkKafkaConsumer 向 Kafka 重新提交偏移量，就可以重新消费数据、保证结果的一致性了。

（3）输出端

写入 Kafka 的过程实际上是一个两段式的提交：处理完毕得到结果，写入Kafka时是基于事务的“预提交”；等到检查点保存完毕，才会提交事务进行“正式提交”。如果中间出现故障，事务进行回滚，预提交就会被放弃；恢复状态之后，也只能恢复所有已经确认提交的操作。

（4）消费输出端 Kafka 的数据

预提交阶段数据已经写入，只是被标记为“未提交”（uncommitted），而Kafka中默认的隔离级别 isolation.level 是 read_uncommitted，也就是可以读取未提交的数据。外部应用就可以直接消费未提交的数据，对于事务性的保证就失效了。所以应该将隔离级别配置为read_committed，表示消费者遇到未提交的消息时，会停止从分区中消费数据，直到消息被标记为已提交才会再次恢复消费。

.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")

（4）代码实践

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.time.Duration;

public class SourceKafkaSinkKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 代码中用到 hdfs，需要导入 hadoop 依赖、指定访问 hdfs 的用户名
        System.setProperty("HADOOP_USER_NAME", "xxx");

        // 1、启用检查点,设置为精准一次
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointStorage("hdfs://vm2:8020/chk");
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.读取kafka
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("vm1:9092,vm2:9092,vm3:9092")
                .setGroupId("g1")
                .setTopics("t1")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> kafkasource = env
                .fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource");

        /**
         * 3.写出到Kafka
         * 精准一次 写入Kafka，需要满足以下条件，缺一不可
         * 3.1 开启checkpoint
         * 3.2 sink设置保证级别为 精准一次
         * 3.3 sink设置事务前缀
         * 3.4 sink设置事务超时时间： checkpoint间隔 <  事务超时时间  < max的15分钟
         */
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                // 指定 kafka 的地址和端口
                .setBootstrapServers("vm1:9092,vm2:9092,vm3:9092")
                // 指定序列化器：指定Topic名称、具体的序列化
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("t2")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setTransactionalIdPrefix("transactional-")
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10*60*1000+"")
                .build();
        kafkasource.sinkTo(kafkaSink);

        env.execute();
    }
}