流计算引擎对分布式状态一致性的保证：Exactly Once, At least Once, At most once

流计算引擎对状态一致性的保证：Exactly Once, At least Once, At most once

参考: 《Stream processing with Apache Flink》、《Streaming Systems》

流计算引擎一般提供的是系统内部的状态一致性。这其中不包括输入和输出的一致性，因为输入端和输出端通常并不属于流引擎，而会对接其他成熟的系统比如Kafka或数据库。输入端的一致性取决于它是否能始终以不变的顺序重放数据（比如从文件中读取）。输出结果的一致性跟数据库的一致性类似，主要取决于sink的输出是否具备幂等性（idempotency）或支持事务（transaction）。当然，在应用中，我们关注的一定是端到端(end-to-end)的一致性，也就是包括输入、系统内部、和输出的整体一致性。那么，流计算引擎要如何提供它这部分的保证呢？

分布式系统中常用这么几种语义（semantics）来描述系统在经历了故障恢复后，内部各个组件之间状态的一致性。严格程度从高到低为：Exactly Once（准确一次）, At least Once（至少一次）, At most once（最多一次）。

怎么界定每个部分的状态是一致的？我们来看几个例子：

数据源S发送给算子A数据，由A进行累加，也就是sum。A中的状态就是累加的和，S的状态就是当前已发送的流（因为A的当前结果和S之前发送的所有数据都有关）。一开始，S发出了1,2,3。A的状态依次更新为1,3,6。当S发出4的时候，A出现了故障。现在S和A的状态分别是S：1,2,3,4，A:6。接下来可能有这么几种情况：

• at most once: S不知道A发生了故障。随后系统从故障中恢复，A收到了S发送的下一条数据“5”，数据“4”实际上被丢失了。这时S的发送历史是1,2,3,4,5，A的状态是11。而“1,2,3,4,5”所对应的累加和应该是15，所以S和A就出现了状态不一致。S认为A的状态包含了“4”，而A实际上没有。
• exactly once: 回到S：1,2,3,4，A：6的时间点。如果这时S发现A挂了，并且A没有处理“4”，系统从故障中恢复之后，S和A都退回到上一个一致的状态，然后重新开始运算。比如上一个一致的状态是：S:1,2,3, A:6，那么S回退之后会重发“4”。如果A成功处理“4”，现在的状态就变成：S：1,2,3,4， A：10。S和A的状态就是一致的。如果上一个一致的状态是：S:1,2, A:3，那么S回退之后会从“3”开始重发。
• exactly once: 同样是回到S: 1,2,3,4, A: 6。如果这时A已经处理了“4”，那么状态变为S: 1,2,3,4, A: 10。然后A在通知S“4已处理”之前挂了。因为S不知道A已经处理了“4”，所以为了确保一致性，S必须假设A没有处理"4"。因此，S和A仍然要一起退回上一个一致的状态，并且从一致的状态重新开始运算。比如说：S:1,2,3, A:6。S重发“4”并且被A处理之后，虽然A实际上处理了“4”两次，但A第一次处理“4”之后的状态在故障之后弃用了。所以在A当前的状态中，“4”只被处理了一次，所以是exactly once。
• at least once: 与上面的例子情况相同，但是S和A没有一起退回上一个一致的状态，比如说在系统恢复之后，两者的状态为：S:1,2,3, A: 10。这时，S重发“4”，状态变为S:1,2,3,4 A: 14，当前状态下，“4”被处理了两次，且状态不一致。

我们再来看一看就算重复处理数据，状态依然一致的情况。也就是所谓的幂等(idempotent)操作（执行多次与执行一次的效果相同）。比如update就是一个幂等操作，将一个值update为1,这个操作不管执行几次，结果都是相同的。

• at least once/exactly once：数据源S2发送给算子A2数据，A2将求出流中的最大值。一开始S发出了1,2,3。A的状态依次更新为1,2,3。与上面一样，当S发出4的时候，A在处理完4之后、通知S之前挂了，但A保存下了处理“4”之后的状态。系统恢复后S和A的状态分别为S:1,2,3 A:4。S重新发送4时，S和A的状态为S:1,2,3,4 A:4。这个时候，虽然A实际上处理了两次4，但S和A的状态是一致的。这个例子中虽然A处理了两次“4”，但A处理两次和处理一次导致的状态是一样的，所以结果上讲，他也实现了exactly once的保证。

回到语义上，如果不管故障是否发生，一个数据一定只会被处理一次（exactly once)，就好像故障没有发生一样。那么不管是不是幂等操作，我们都能保证状态的一致性，这也是最严格的一致性保证。如果一个数据可能被处理不止一次(at least once)，则仅能在幂等操作或能够分辨重复数据的前提下保证一致性。如果一个数据最多被处理一次（at most once），那就等于没有提供任何保证。毕竟就算所有数据都丢失，也符合这个“最多一次”的保证。

需要注意的是，exactly-once和at-least-once的保证都有一个前提：能够可靠地、按原来的顺序重放没有被处理的数据。比如上面的例子中，S和A一起倒退回上一个一致的状态，然后重放数据。一些常作为数据源的框架，比如Kafka，就有这个能力。如果数据源不提供这样的便利，我们也可以在系统内部实现它：我们可以将这些数据写入日志中，以便重放时查询。也可以存入缓冲区，当下游返回了ack，就将对应的数据从缓冲区中删掉（比如S收到A发出的“4已处理”的消息后，将“4”从缓冲区中删掉）。当需要重放时，只重发缓冲区中剩下的数据。