gydtep

如果定义流计算的输入集合为：E，t 时刻以来的输入集合为 E(t)，输出集合为 Sink(t)，引擎此时状态为 State(t)，State(t) 包括各个算子的状态（包括上面提到的不确定性计算）、数据源的消费偏移量（或文件读取偏移等）等：

gydtep

令 O(t) = Sink(t) + State(t)，即将计算对引擎状态的更新视为一种特殊的输出，则流计算过程可简化为：

F(E(t), O(t)) = O(t+1)

结合流计算上面流计算一致性的定义，我们希望在引擎发生故障 FailOver 时，存在一种恢复函数 R 使得

gydtep

我们在上面定义了端到端一致性难题：R(E(t), O(t)) = O(t+1)。从输出结果的使用方（引擎内部和引擎下游数据消费方）的视角来看：对于记录 O(t+1)，当在故障发生的时间小于 t （数据没有输出）或者 大于 t + 1（数据已经输出了），数据肯定是一致的。

gydtep

目前流计算引擎的种类非常多，不是所有的引擎都可以实现端到端一致的流处理，在具备此能力的引擎中，从技术成本、引擎架构、能力范围考虑，会有不同的取舍和实现，如 Flink 中使用了轻量级的「分布式一致性快照」用于状态管理，Kafka Streams 为何没有使用呢？实现了幂等输出就一定能实现端到端一致么？本章节会一一解答上述问题。

gydtep

MillWheel 会对每一条记录赋予一个唯一 ID，同时基于此 ID 维护一份是否处理过当前记录的目录。对于每一条流入当前算子的记录，引擎查找此 ID 目录以确定此记录是否是已经处理过。这里会有很多技术上的挑战，这里稍微举几个例子。

gydtep

在流计算过程中，定期（epoch）以事务（2PC）的方式进行批量存储结果（分布式一致性快照 + 写外部存储）。需要注意的是，由于 Flink 会以 epoch 为周期输出结果，因此基于此构建的流处理系统会存在一定的端到端延迟。

gydtep

Kafka Streams 将上述结果定期以事务的方式进行批量存储，上述事务在 Kafka 这被称之为 Transactions API，使用这个 API 构建的流处理应用，可以在一个事务**多个主题消息进行同时提交，如果事务终止或回滚，则下游消费不会读取到相应的结果（当然下游消费者也需要配置相应的一致性级别），其过程如下图所示：

gydtep

但还不满足事务隔离性的消息（read_committed 级别），从流计算输出的角度来看，这些消息已被成功处理同时输出至下游，但从端到端的一致性来看，它们依然属于不一致的数据。又如，使用 Flink 处理 CDC（Change Data Capture） 的场景，

gydtep

同传统的批处理系统类似，流处理中也是以 RDD 构建出整个的数据血缘，当发生 FailOver 时，则重新计算整个 RDD 就可以了。

gydtep

均可以视为当前处理逻辑单元（算子或最终存储）对上游的输入（引擎状态+输出结果）进行的幂等化处理：引擎 FailOver -> 输入源的事件会进行重发 -> 前期存储的结果会用于去重/事务回滚让结果（引擎状态+输出结果）回到上一次的一致性状态 -> 下一批结果输出 -> 结果接受端只影响一次 -> 实现了端到端的一致。