Flink 详解（二）：核心篇 Ⅲ

29、Flink 通过什么实现可靠的容错机制？

Flink 使用 轻量级分布式快照，设计检查点（checkpoint）实现可靠容错。

30、什么是 Checkpoin 检查点？

Checkpoint 被叫做 检查点，是 Flink 实现容错机制最核心的功能，是 Flink 可靠性的基石，它能够根据配置周期性地基于 Stream 中各个 Operator 的 状态 来生成 Snapshot 快照，从而将这些状态数据定期持久化存储下来，当 Flink 程序一旦意外崩溃时，重新运行程序时可以有选择地从这些 Snapshot 进行恢复，从而修正因为故障带来的程序数据状态中断。

Flink 的 Checkpoint 机制原理来自 Chandy-Lamport algorithm 算法（一种分布式快照算法）。

注意区分 State 和 Checkpoint：

1.State

• 一般指一个具体的 Task/Operator 的状态（Operator 的状态表示一些算子在运行的过程中会产生的一些中间结果）。
• State 数据默认保存在 Java 的堆内存中 / TaskManage 节点的内存中。
• State 可以被记录，在失败的情况下数据还可以恢复。

2.Checkpoint

• 表示了一个 FlinkJob 在一个特定时刻的一份全局状态快照，即包含了所有 Task/Operator 的状态。

• 可以理解为 Checkpoint 是把 State 数据定时持久化存储了。

• 比如 KafkaConsumer 算子中维护的 Offset 状态，当任务重新恢复的时候可以从 Checkpoint 中获取。

31、什么是 Savepoint 保存点？

保存点 在 Flink 中叫作 Savepoint，是基于 Flink 检查点机制的应用完整快照备份机制，用来保存状态，可以在另一个集群或者另一个时间点从保存的状态中将作业恢复回来。适用于 应用升级、集群迁移、 Flink 集群版本更新、A/B 测试 以及 假定场景、暂停和重启、归档 等场景。保存点可以视为一个（算子 ID → State）的 Map，对于每一个有状态的算子，Key 是算子 ID，Value 是算子 State。

32、什么是 CheckpointCoordinator 检查点协调器？

Flink 中检查点协调器叫作 CheckpointCoordinator，负责协调 Flink 算子的 State 的分布式快照。当触发快照的时候，CheckpointCoordinator 向 Source 算子中注入 Barrier 消息 ，然后等待所有的 Task 通知检查点确认完成，同时持有所有 Task 在确认完成消息中上报的 State 句柄。

33、Checkpoint 中保存的是什么信息？

检查点里面到底保存着什么信息呢？我们以 Flink 消费 Kafka 数据 Wordcount 为例：

1、我们从 Kafka 读取到一条条的日志，从日志中解析出 app_id，然后将统计的结果放到内存中一个 Map 集合，app_id 做为 key，对应的 pv 做为 value，每次只需要将相应 app_id 的 pv 值 $+1$ 后 put 到 Map 中即可；

2、kafka topic：test；

3、Flink 运算流程如下：

kafka topic 有且只有一个分区。

假设 kafka 的 topic-test 只有一个分区，flink 的 Source task 记录了当前消费到 kafka test topic 的所有 partition 的 offset。

例：（0，1000）表示 0 号 partition 目前消费到 offset 为 1000 的数据。

Flink 的 pv task 记录了当前计算的各 app 的 pv 值，为了方便讲解，我这里有两个 app：app1、app2。

例：（app1，50000）（app2，10000）
表示 app1 当前 pv 值为 50000
表示 app2 当前 pv 值为 10000
每来一条数据，只需要确定相应 app_id，将相应的 value 值 +1 后 put 到 map 中即可。

该案例中，CheckPoint 保存的其实就是 第 n 次 CheckPoint 消费的 offset 信息和各 app 的 pv 值信息，记录一下发生 CheckPoint 当前的状态信息，并将该状态信息保存到相应的状态后端。图下代码：（注：状态后端是保存状态的地方，决定状态如何保存，如何保障状态高可用，我们只需要知道，我们能从状态后端拿到 offset 信息和 pv 信息即可。状态后端必须是高可用的，否则我们的状态后端经常出现故障，会导致无法通过 checkpoint 来恢复我们的应用程序）。

chk-100
offset：（0，1000）
pv：（app1，50000）（app2，10000）
该状态信息表示第 100 次 CheckPoint 的时候， partition 0 offset 消费到了 1000，pv 统计。

34、当作业失败后，检查点如何恢复作业？

Flink 提供了 应用自动恢复机制 和 手动作业恢复机制。

1、应用自动恢复机制

Flink 设置有作业失败重启策略，包含三种：

• 定期恢复策略（fixed-delay）：固定延迟重启策略会尝试一个给定的次数来重启 Job，如果超过最大的重启次数，Job 最终将失败，在连续两次重启尝试之间，重启策略会等待一个固定时间，默认 Integer.MAX_VALUE 次。

• 失败比率策略（failure-rate）：失败比率重启策略在 Job 失败后重启，但是超过失败率后，Job 会最终被认定失败，在两个连续的重启尝试之间，重启策略会等待一个固定的时间。

• 直接失败策略（None）：失败不重启。

2、手动作业恢复机制

因为 Flink 检查点目录分别对应的是 JobId，每通过 flink run 方式 / 页面提交方式恢复都会重新生成 JobId，Flink 提供了在启动之时通过设置 -s 参数指定检查点目录的功能，让新的 Jobld 读取该检查点元文件信息和状态信息，从而达到指定时间节点启动作业的目的。

启动方式如下：

/bin/flink -s /flink/checkpoints/03112312a12398740a87393/chk-50/_metadata

35、当作业失败后，从保存点如何恢复作业？

从保存点恢复作业并不简单，尤其是在作业变更（如修改逻辑、修复 Bug）的情况下， 需要考虑如下几点:

• 算子的顺序改变。如果对应的 UID 没变，则可以恢复；如果对应的 UID 变了，恢复失败。
• 作业中添加了新的算子。如果是无状态算子，没有影响，可以正常恢复；如果是有状态的算子，跟无状态的算子一样处理。
• 从作业中删除了一个有状态的算子。默认需要恢复保存点中所记录的所有算子的状态，如果删除了一个有状态的算子，从保存点恢复的时候被删除的 OperatorID 找不到，所以会报错，可以通过在命令中添加 --allowNonReStoredSlale(short: -n) 跳过无法恢复的算子 。
• 添加和删除无状态的算子。如果手动设置了 UID 则可以恢复，保存点中不记录无状态的算子。如果是自动分配的 UID ，那么有状态算子的可能会变（Flink 一个单调递增的计数器生成 UID，DAG 改变，计数器极有可能会变）很有可能恢复失败。

36、Flink 如何实现轻量级异步分布式快照？

要实现分布式快照，最关键的是能够将数据流切分。Flink 中使用 屏障（Barrier）来切分数据流。 Barrierr 会周期性地注入数据流中，作为数据流的一部分，从上游到下游被算子处理。Barrier 会严格保证顺序，不会超过其前边的数据。Barrier 将记录分割成记录集，两个 Barrier 之间的数据流中的数据隶属于同一个检查点。每一个 Barrier 都携带一个其所属快照的 ID 编号。Barrier 随着数据向下流动，不会打断数据流，因此非常轻量。 在一个数据流中，可能会存在多个隶属于不同快照的 Barrier ，并发异步地执行分布式快照，如下图所示：

Barrier 会在数据流源头被注人并行数据流中。Barrier $n$ 所在的位置就是恢复时数据重新处理的起始位置。 例如，在 Kafka 中，这个位置就是最后一个记录在分区内的偏移量（offset），作业恢复时，会根据这个位置从这个偏移量之后向 Kafka 请求数据，这个偏移量就是 State 中保存的内容之一。

Barrier 接着向下游传递。当一个非数据源算子从所有的输入流中收到了快照 $n$ 的 Barrier 时，该算子就会对自己的 State 保存快照，并向自己的下游 广播发送 快照 $n$ 的 Barrier。一旦 Sink 算子接收到 Barrier，有两种情况：

• 如果是引擎内严格一次处理保证，当 Sink 算子已经收到了所有上游的 Barrie $n$ 时， Sink 算子对自己的 State 进行快照，然后通知检查点协调器（CheckpointCoordinator）。当所有的算子都向检查点协调器汇报成功之后，检查点协调器向所有的算子确认本次快照完成。
• 如果是端到端严格一次处理保证，当 Sink 算子已经收到了所有上游的 Barrie $n$ 时， Sink 算子对自己的 State 进行快照，并 预提交事务（两阶段提交的第一阶段），再通知检查点协调器（CheckpointCoordinator），检查点协调器向所有的算子确认本次快照完成，Sink 算子 提交事务（两阶段提交的第二阶段），本次事务完成。

我们接着 $33$ 的案例来具体说一下如何执行分布式快照：

对应到 pv 案例中就是，Source Task 接收到 JobManager 的编号为 chk-100（从最近一次恢复）的 CheckPoint 触发请求后，发现自己恰好接收到 Kafka offset(0,1000) 处的数据，所以会往 offset(0,1000) 数据之后，offset(0,1001) 数据之前，安插一个 Barrier，然后自己开始做快照，也就是将 offset(0,1000) 保存到状态后端 chk-100 中。然后 Barrier 接着往下游发送，当统计 pv 的 task 接收到 Barrier 后，也会暂停处理数据，将自己内存中保存的 pv 信息 (app1，50000)、(app2，10000) 保存到状态后端 chk-100 中。OK，Flink 大概就是通过这个原理来保存快照的。

统计 pv 的 task 接收到 Barrier，就意味着 Barrier 之前的数据都处理了，所以说，不会出现丢数据的情况。

37、什么是 Barrier 对齐？

上图从左至右分别表示：开始对齐，对齐，执行检查点，继续处理数据。

一旦 operator 从输入流接收到 checkpoint barrier $n$，它就不能处理来自该流的任何数据记录，直到它从其他所有输入接收到 barrier $n$ 为止。否则，它会混合属于快照 $n$ 的记录和属于快照 $n+1$ 的记录；

如上图所示：

• 图 $1$：算子收到数字流的 Barrier，字母流对应的 Barrier 尚未到达。
• 图 $2$：算子收到数字流的 Barrier，会继续从数字流中接收数据，但这些流只能被搁置，记录不能被处理，而是放入缓存中，等待字母流 Barrier 到达。在字母流到达前， $1、2、3$ 数据已经被缓存。
• 图 $3$：字母流到达，算子开始对齐 State 进行异步快照，并将 Barrier 向下游广播，并不等待快照执行完毕。
• 图 $4$：算子做异步快照，首先处理缓存中积压数据，然后再从输入通道中获取数据。

38、什么是 Barrier 不对齐？

Checkpoint 是要等到所有的 Barrier 全部都到才算完成。

上述图 $2$ 中，当还有其他输入流的 Barrier 还没有到达时，会把已到达的 Barrier 之后的数据 $1、2、3$ 搁置在缓冲区，等待其他流的 Barrier 到达后才能处理。

Barrier 不对齐：就是指当还有其他流的 Barrier 还没到达时，为了不影响性能，也不用理会，直接处理 Barrier 之后的数据。等到所有流的 Barrier 的都到达后，就可以对该 Operator 做 Checkpoint 了。

39、为什么要进行 Barrier 对齐？不对齐到底行不行？

Exactly Once 时必须 Barrier 对齐，如果 Barrier 不对齐就变成了 At Least Once。

Checkpoint 的目的就是为了保存快照，如果不对齐，那么在 chk-100 快照之前，已经处理了一些 chk-100 对应的 offset 之后的数据，当程序从 chk-100 恢复任务时，chk-100 对应的 offset 之后的数据还会被处理一次，所以就出现了重复消费。

41、要实现 Exactly-Once 需具备什么条件？

流系统要实现 Exactly-Once，需要保证上游 Source 层、中间计算层和下游 Sink 层三部分同时满足端到端严格一次处理，如下图：

Source 端：数据从上游进入 Flink，必须保证消息严格一次消费。同时 Source 端必须满足可重放（replay）。否则 Flink 计算层收到消息后未计算，却发生 failure 而重启，消息就会丢失。

Flink 计算层：利用 Checkpoint 机制，把状态数据定期持久化存储下来，Flink 程序一旦发生故障的时候，可以选择状态点恢复，避免数据的丢失、重复。

Sink 端：Flink 将处理完的数据发送到 Sink 端时，通过 两阶段提交协议 ，即 TwoPhaseCommitSinkFunction 函数。该 SinkFunction 提取并封装了两阶段提交协议中的公共逻辑，保证 Flink 发送 Sink 端时实现严格一次处理语义。同时，Sink 端必须支持事务机制，能够进行数据回滚或者满足幂等性。

• 回滚机制：即当作业失败后，能够将部分写入的结果回滚到之前写入的状态。
• 幂等性：就是一个相同的操作，无论重复多少次，造成的结果和只操作一次相等。即当作业失败后，写入部分结果，但是当重新写入全部结果时，不会带来负面结果，重复写入不会带来错误结果。

42、什么是两阶段提交协议？

两阶段提交协议（Two-Phase Commit，2PC）是解决分布式事务问题最常用的方法，它可以保证在分布式事务中，要么所有参与进程都提交事务，要么都取消，即实现 $ACID$ 中的 $A$（原子性）。

两阶段提交协议中有两个重要角色，协调者（Coordinator）和 参与者（Participant），其中协调者只有一个，起到分布式事务的协调管理作用，参与者有多个。

两阶段提交阶段分为两个阶段：投票阶段（Voting）和 提交阶段（Commit）。

（1）投票阶段

• 协调者向所有参与者发送 prepare 请求和事务内容，询问是否可以准备事务提交，等待参与者的响应。
• 参与者执行事务中包含的操作，并记录 undo 日志（用于回滚）和 redo 日志（用于重放），但不真正提交。
• 参与者向协调者返回事务操作的执行结果，执行成功返回 yes，失败返回 no。

（2）提交阶段

• 分为成功与失败两种情况。
• 若所有参与者都返回 yes，说明事务可以提交：
  • 协调者向所有参与者发送 commit 请求。
  • 参与者收到 commit 请求后，将事务真正地提交上去，并释放占用的事务资源，并向协调者返回 ack。
  • 协调者收到所有参与者的 ack 消息，事务成功完成，如下图：

• 若有参与者返回 no 或者超时未返回，说明事务中断，需要回滚：
  • 协调者向所有参与者发送 rollback 请求。
  • 参与者收到 rollback 请求后，根据 undo 日志回滚到事务执行前的状态，释放占用的事务资源，并向协调者返回 ack。
  • 协调者收到所有参与者的 ack 消息，事务回滚完成。

43、Flink 如何保证 Exactly-Once 语义？

Flink 通过 两阶段提交协议 来保证 Exactly-Once 语义。

对于 Source 端：Source 端严格一次处理比较简单，因为数据要进入 Flink 中，所以 Flink 只需要保存消费数据的偏移量（offset）即可。如果 Source 端为 Kafka，Flink 将 Kafka Consumer 作为 Source，可以将偏移量保存下来，如果后续任务出现了故障，恢复的时候可以由连接器重置偏移量，重新消费数据，保证一致性。

对于 Sink 端：Sink 端是最复杂的，因为数据是落地到其他系统上的，数据一旦离开 Flink 之后，Flink 就监控不到这些数据了，所以严格一次处理语义必须也要应用于 Flink 写入数据的外部系统，故这些外部系统必须提供一种手段允许提交或回滚这些写入操作，同时还要保证与 Flink Checkpoint 能够协调使用（Kafka $0.11$ 版本已经实现精确一次处理语义）。

我们以 Kafka - Flink - Kafka 为例 说明如何保证 Exactly-Once 语义。

如上图所示：Flink 作业包含以下算子。

• 一个 Source 算子，从 Kafka 中读取数据（即 KafkaConsumer）
• 一个窗口算子，基于时间窗口化的聚合运算（即 window+window 函数）
• 一个 Sink 算子，将结果写会到 Kafka（即 KafkaProducer）

Flink 使用 两阶段提交协议，即 预提交（Pre-commit）阶段和 提交（Commit）阶段保证端到端严格一次。

1、预提交阶段

（1）当 Checkpoint 启动时，进入预提交阶段，JobManager 向 Source Task 注入检查点分界线（CheckpointBarrier），Source Task 将 CheckpointBarrier 插入数据流，向下游广播开启本次快照，如下图所示：

（2）Source 端：Flink Data Source 负责保存 KafkaTopic 的 offset 偏移量，当 Checkpoint 成功时 Flink 负责提交这些写入，否则就终止取消掉它们，当 Checkpoint 完成位移保存，它会将 Checkpoint Barrier（检查点分界线） 传给下一个 Operator，然后每个算子会对当前的状态做个快照，保存到 状态后端（State Backend）。

对于 Source 任务而言，就会把当前的 offset 作为状态保存起来。下次从 Checkpoint 恢复时，Source 任务可以重新提交偏移量，从上次保存的位置开始重新消费数据，如下图所示：

（3）Slink 端：从 Source 端开始，每个内部的 Transformation 任务遇到 Checkpoint Barrier（检查点分界线）时，都会把状态存到 Checkpoint 里。数据处理完毕到 Sink 端时，Sink 任务首先把数据写入外部 Kafka，这些数据都属于预提交的事务（还不能被消费），此时的 Pre-commit 预提交阶段下 ，Data Sink 在保存状态到状态后端的同时还必须预提交它的外部事务，如下图所示：

2、提交阶段

（4）当所有算子任务的快照完成（所有创建的快照都被视为是 Checkpoint 的一部分），也就是这次的 Checkpoint 完成时，JobManager 会向所有任务发通知，确认这次 Checkpoint 完成，此时 Pre-commit 预提交阶段才算完成。才正式到两阶段提交协议的第二个阶段：Commit 阶段。该阶段中 JobManager 会为应用中每个 Operator 发起 Checkpoint 已完成的回调逻辑。

本例中的 Data Source 和窗口操作无外部状态，因此在该阶段，这两个 Opeartor 无需执行任何逻辑，但是 Data Sink 是有外部状态的，此时我们必须提交外部事务，当 Sink 任务收到确认通知，就会正式提交之前的事务，Kafka 中未确认的数据就改为 “已确认”，数据就真正可以被消费了，如下图所示：

注：Flink 由 JobManager 协调各个 TaskManager 进行 Checkpoint 存储，Checkpoint 保存在 StateBackend（状态后端） 中，默认 StateBackend 是内存级的，也可以改为文件级的进行持久化保存。

44、数的很好，很清楚，那你对 Flink 端到端 严格一次 Exactly-Once 语义做个总结。