Flink端到端严格一次Exactly-Once

1. 什么是Exactly-Once

Exactly-Once语义 : 指端到端的一致性，从数据读取、引擎计算、写入外部存储的整个过程中，即使机器或软件出现故障，都确保数据仅处理一次，不会重复、也不会丢失。

2. 要实现Exactly-Once,需具备什么条件？

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

Source端：数据从上游进入Flink，必须保证消息严格一次消费。同时Source 端必须满足可重放（replay）。否则 Flink 计算层收到消息后未计算，却发生 failure 而重启，消息就会丢失。
Flink计算层：利用 Checkpoint 机制，把状态数据定期持久化存储下来，Flink程序一旦发生故障的时候，可以选择状态点恢复，避免数据的丢失、重复。
Sink端：Flink将处理完的数据发送到Sink端时，通过 两阶段提交协议 ，即 TwoPhaseCommitSinkFunction 函数。该 SinkFunction 提取并封装了两阶段提交协议中的公共逻辑，保证Flink 发送Sink端时实现严格一次处理语义。 同时：Sink端必须支持事务机制，能够进行数据回滚或者满足幂等性。

回滚机制：即当作业失败后，能够将部分写入的结果回滚到之前写入的状态。**

幂等性：就是一个相同的操作，无论重复多少次，造成的结果和只操作一次相等。即当作业失败后，写入部分结果，但是当重新写入全部结果时，不会带来负面结果，重复写入不会带来错误结果。

3. 什么是两阶段提交协议？

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

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

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

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

提交阶段：
分为成功与失败两种情况。
若所有参与者都返回 yes，说明事务可以提交：

• 协调者向所有参与者发送 commit 请求。
• 参与者收到 commit 请求后，将事务真正地提交上去，并释放占用的事务资源，并向协调者返回 ack 。
• 协调者收到所有参与者的 ack 消息，事务成功完成，如下图：

若有参与者返回 no 或者超时未返回，说明事务中断，需要回滚：

• 协调者向所有参与者发送rollback请求。
• 参与者收到rollback请求后，根据undo日志回滚到事务执行前的状态，释放占用的事务资源，并向协调者返回ack。
• 协调者收到所有参与者的ack消息，事务回滚完成。

4. 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作业包含以下算子。
（1）一个Source算子，从Kafka中读取数据（即KafkaConsumer）
（2）一个窗口算子，基于时间窗口化的聚合运算（即window+window函数）
（3）一个Sink算子，将结果写会到Kafka（即kafkaProducer）

Flink使用两阶段提交协议 预提交（Pre-commit）阶段和 提交（Commit）阶段保证端到端严格一次。
（1）预提交阶段
1、当Checkpoint 启动时，进入预提交阶段，JobManager 向Source Task 注入检查点分界线（CheckpointBarrier）,Source Task 将 CheckpointBarrier 插入数据流，向下游广播开启本次快照，如下图所示：

预处理阶段： Checkpoint 启动
2、Source 端：Flink Data Source 负责保存 KafkaTopic 的 offset偏移量，当 Checkpoint 成功时 Flink 负责提交这些写入，否则就终止取消掉它们，当 Checkpoint 完成位移保存，它会将 checkpoint barrier（检查点分界线） 传给下一个 Operator，然后每个算子会对当前的状态做个快照，保存到状态后端（State Backend）。
对于 Source 任务而言，就会把当前的 offset 作为状态保存起来。下次从 Checkpoint 恢复时，Source 任务可以重新提交偏移量，从上次保存的位置开始重新消费数据，如下图所示：

预处理阶段：checkpoint barrier传递 及 offset 保存
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 是内存级的，也可以改为文件级的进行持久化保存。

出处：原文