本系列包含：
Flink 使用 轻量级分布式快照，设计检查点（checkpoint）实现可靠容错。
Checkpoint 被叫做 检查点，是 Flink 实现容错机制最核心的功能，是 Flink 可靠性的基石，它能够根据配置周期性地基于 Stream 中各个 Operator 的 状态 来生成 Snapshot 快照，从而将这些状态数据定期持久化存储下来，当 Flink 程序一旦意外崩溃时，重新运行程序时可以有选择地从这些 Snapshot 进行恢复，从而修正因为故障带来的程序数据状态中断。
Flink 的 Checkpoint 机制原理来自 Chandy-Lamport algorithm 算法（一种分布式快照算法）。
注意区分 State 和 Checkpoint：
1.State
2.Checkpoint
表示了一个 FlinkJob 在一个特定时刻的一份全局状态快照，即包含了所有 Task/Operator 的状态。
可以理解为 Checkpoint 是把 State 数据定时持久化存储了。
比如 KafkaConsumer 算子中维护的 Offset 状态，当任务重新恢复的时候可以从 Checkpoint 中获取。
保存点 在 Flink 中叫作 Savepoint，是基于 Flink 检查点机制的应用完整快照备份机制，用来保存状态，可以在另一个集群或者另一个时间点从保存的状态中将作业恢复回来。适用于 应用升级、集群迁移、 Flink 集群版本更新、A/B 测试 以及 假定场景、暂停和重启、归档 等场景。保存点可以视为一个（算子 ID → State）的 Map，对于每一个有状态的算子，Key 是算子 ID，Value 是算子 State。
Flink 中检查点协调器叫作 CheckpointCoordinator，负责协调 Flink 算子的 State 的分布式快照。当触发快照的时候，CheckpointCoordinator 向 Source 算子中注入 Barrier 消息 ，然后等待所有的 Task 通知检查点确认完成，同时持有所有 Task 在确认完成消息中上报的 State 句柄。
检查点里面到底保存着什么信息呢？我们以 Flink 消费 Kafka 数据 Wordcount 为例：
1、我们从 Kafka 读取到一条条的日志，从日志中解析出 app_id，然后将统计的结果放到内存中一个 Map 集合，app_id 做为 key，对应的 pv 做为 value，每次只需要将相应 app_id 的 pv 值
+
1
+1
+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 统计。
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
从保存点恢复作业并不简单，尤其是在作业变更（如修改逻辑、修复 Bug）的情况下， 需要考虑如下几点:
--allowNonReStoredSlale(short: -n) 跳过无法恢复的算子 。要实现分布式快照，最关键的是能够将数据流切分。Flink 中使用 屏障（Barrier）来切分数据流。 Barrierr 会周期性地注入数据流中，作为数据流的一部分，从上游到下游被算子处理。Barrier 会严格保证顺序，不会超过其前边的数据。Barrier 将记录分割成记录集，两个 Barrier 之间的数据流中的数据隶属于同一个检查点。每一个 Barrier 都携带一个其所属快照的 ID 编号。Barrier 随着数据向下流动，不会打断数据流，因此非常轻量。 在一个数据流中，可能会存在多个隶属于不同快照的 Barrier ，并发异步地执行分布式快照，如下图所示：
Barrier 会在数据流源头被注人并行数据流中。Barrier
n
n
n 所在的位置就是恢复时数据重新处理的起始位置。 例如，在 Kafka 中，这个位置就是最后一个记录在分区内的偏移量（offset），作业恢复时，会根据这个位置从这个偏移量之后向 Kafka 请求数据，这个偏移量就是 State 中保存的内容之一。
Barrier 接着向下游传递。当一个非数据源算子从所有的输入流中收到了快照 n n n 的 Barrier 时，该算子就会对自己的 State 保存快照，并向自己的下游 广播发送 快照 n n n 的 Barrier。一旦 Sink 算子接收到 Barrier，有两种情况：
CheckpointCoordinator）。当所有的算子都向检查点协调器汇报成功之后，检查点协调器向所有的算子确认本次快照完成。CheckpointCoordinator），检查点协调器向所有的算子确认本次快照完成，Sink 算子 提交事务（两阶段提交的第二阶段），本次事务完成。我们接着 33 33 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 之前的数据都处理了，所以说，不会出现丢数据的情况。
上图从左至右分别表示：开始对齐，对齐，执行检查点，继续处理数据。
一旦 operator 从输入流接收到 checkpoint barrier n n n，它就不能处理来自该流的任何数据记录，直到它从其他所有输入接收到 barrier n n n 为止。否则，它会混合属于快照 n n n 的记录和属于快照 n + 1 n + 1 n+1 的记录；
如上图所示：
Checkpoint 是要等到所有的 Barrier 全部都到才算完成。
上述图 2 2 2 中，当还有其他输入流的 Barrier 还没有到达时，会把已到达的 Barrier 之后的数据 1 、 2 、 3 1、2、3 1、2、3 搁置在缓冲区，等待其他流的 Barrier 到达后才能处理。
Barrier 不对齐：就是指当还有其他流的 Barrier 还没到达时，为了不影响性能，也不用理会，直接处理 Barrier 之后的数据。等到所有流的 Barrier 的都到达后，就可以对该 Operator 做 Checkpoint 了。
Exactly Once 时必须 Barrier 对齐，如果 Barrier 不对齐就变成了 At Least Once。
Checkpoint 的目的就是为了保存快照，如果不对齐，那么在 chk-100 快照之前，已经处理了一些 chk-100 对应的 offset 之后的数据，当程序从 chk-100 恢复任务时，chk-100 对应的 offset 之后的数据还会被处理一次，所以就出现了重复消费。
Exactly-Once 语义: 指端到端的一致性，从 数据读取、引擎计算、写入外部存储 的整个过程中，即使机器或软件出现故障，都确保数据仅处理一次，不会重复、也不会丢失。
流系统要实现 Exactly-Once，需要保证上游 Source 层、中间计算层和下游 Sink 层三部分同时满足端到端严格一次处理，如下图：
Source 端：数据从上游进入 Flink，必须保证消息严格一次消费。同时 Source 端必须满足可重放（replay）。否则 Flink 计算层收到消息后未计算，却发生 failure 而重启，消息就会丢失。
Flink 计算层：利用 Checkpoint 机制，把状态数据定期持久化存储下来，Flink 程序一旦发生故障的时候，可以选择状态点恢复，避免数据的丢失、重复。
Sink 端：Flink 将处理完的数据发送到 Sink 端时，通过 两阶段提交协议 ，即 TwoPhaseCommitSinkFunction 函数。该 SinkFunction 提取并封装了两阶段提交协议中的公共逻辑，保证 Flink 发送 Sink 端时实现严格一次处理语义。同时，Sink 端必须支持事务机制，能够进行数据回滚或者满足幂等性。
两阶段提交协议（Two-Phase Commit，2PC）是解决分布式事务问题最常用的方法，它可以保证在分布式事务中，要么所有参与进程都提交事务，要么都取消，即实现
A
C
I
D
ACID
ACID 中的
A
A
A（原子性）。
两阶段提交协议中有两个重要角色，协调者（Coordinator）和 参与者（Participant），其中协调者只有一个，起到分布式事务的协调管理作用，参与者有多个。
两阶段提交阶段分为两个阶段：投票阶段（Voting）和 提交阶段（Commit）。
（1）投票阶段
prepare 请求和事务内容，询问是否可以准备事务提交，等待参与者的响应。undo 日志（用于回滚）和 redo 日志（用于重放），但不真正提交。yes，失败返回 no。（2）提交阶段
yes，说明事务可以提交：
commit 请求。commit 请求后，将事务真正地提交上去，并释放占用的事务资源，并向协调者返回 ack。ack 消息，事务成功完成，如下图：
no 或者超时未返回，说明事务中断，需要回滚：
rollback 请求。rollback 请求后，根据 undo 日志回滚到事务执行前的状态，释放占用的事务资源，并向协调者返回 ack。ack 消息，事务回滚完成。
Flink 通过 两阶段提交协议 来保证 Exactly-Once 语义。
对于 Source 端：Source 端严格一次处理比较简单，因为数据要进入 Flink 中，所以 Flink 只需要保存消费数据的偏移量（offset）即可。如果 Source 端为 Kafka，Flink 将 Kafka Consumer 作为 Source，可以将偏移量保存下来，如果后续任务出现了故障，恢复的时候可以由连接器重置偏移量，重新消费数据，保证一致性。
对于 Sink 端：Sink 端是最复杂的，因为数据是落地到其他系统上的，数据一旦离开 Flink 之后，Flink 就监控不到这些数据了，所以严格一次处理语义必须也要应用于 Flink 写入数据的外部系统，故这些外部系统必须提供一种手段允许提交或回滚这些写入操作，同时还要保证与 Flink Checkpoint 能够协调使用（Kafka 0.11 0.11 0.11 版本已经实现精确一次处理语义）。
我们以 Kafka - Flink - Kafka 为例 说明如何保证 Exactly-Once 语义。
如上图所示：Flink 作业包含以下算子。
KafkaConsumer）window+window 函数）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 是内存级的，也可以改为文件级的进行持久化保存。
