所谓的流式处理其实就是对Stream的读取-处理-写入（ETL）操作，应用从Stream中读取数据，再对数据进行相应的处理分析，最后将结果写入另一个Stream中。其中仅一次语义保证了哪怕系统发生故障，每一个ETL操作也仅会被执行一次，不会产生数据的丢失或者重复。这样的可靠性保证对于一些交易、金融类的应用来说至关重要，这就需要Pravega作为流存储与流计算引擎共同努力来完成。

通常来说，对于单独的消息系统而言，语义分为如下三种：

至多一次（At most once）：不管Writer在等待ACK时是否发生超时或者得到错误异常，Writer都不会重新发送Event，因此会有数据丢失的风险。在具体的实现过程中，这一种语义无需做任何额外的控制，实现起来最为简单，因此也通常有着最优的性能。在某些特定的场景中，我们只希望追求极致的性能而不关心数据的丢失，可能会选用此方案。

至少一次（At least once）：如果Writer在等待ACK时发生超时或者得到错误异常，Writer将会重新发送消息，这样能保证每个Event至少被处理一次，保证了数据不会丢失，从而提高了系统的可靠性，但同时会带来数据重复的问题，例如，当Writer往Stream中成功写入一个Event，但是当系统尝试给Writer返回ACK的时候出现网络异常，Writer因没有收到ACK而判断为写入Event失败，因此Writer还是会重新发送此Event，导致数据重复。

仅一次（Exactly once）：在系统发生异常时，Writer可以尝试多次重新发送Event，同时能保证最终每个Event只被写入一次。一些对数据准确性要求非常高的系统需要保证exactly-once语义，譬如支付系统，当用户在移动端付款时，很有可能会因为网络原因导致延时较长甚至超时，用户可能会手动进行刷新操作，如果没有exactly-once的语义支持，很有可能会发生两次扣费，我们绝对不希望此类错误发生。

仅一次语义是实现流处理系统正确性(correctness)的基石，因此也是流存储Pravega自从设计之初就规划好的设计目标。但是，exactly-once的实现也面临着诸多挑战，例如Kafka也直到0.11版本引入了KIP-98之后才完成了仅一次的支持。这种更强的语义不仅使编写应用程序更容易，而且使Pravega有了更为广泛的应用空间。这一篇文章我们将介绍Pravega实现这一特性的设计细节，以及和Flink社区合作开发的端到端(end-to-end)的exactly-once的实现。

Pravega自身的exactly-once语义

从之前对三种语义的描述可以看出，要满足exactly-once的语义，需要对于可能发生的故障具有足够的恢复机制，来保证最终结果的每一条Event仅被写入一次。Pravega通过实现了读写端的以下两个性质，增强了Pravega的读写语义，完成了这一目标:

数据的可恢复性
Writer的幂等性

其中数据的可恢复性保证了在Pravega读取数据过程中，发生故障之后数据进行正确重放的可能性，需要在读客户端优化。而幂等性保证了单一一条数据在数据写入的过程不会出现重复，需要在写客户端优化。

数据的可恢复性

发生故障时，数据的恢复需要有以下三个性质的保障，才能有数据恢复的前提。

持久性 Durability

持久性是指一旦写操作确认成功，即使系统在发生故障的情况下数据都不会丢失，保证了数据的可恢复性。Pravega利用了Bookkeeper以及分层的存储实现了流式存储系统的持久性，帮助用户解决了数据可能会丢失的问题。用户只需要直接使用Pravega，而不需要考虑额外的数据备份工作。

Pravega的持久化详细内容请参阅本系列之前的文章：

https://www.infoq.cn/article/VXA51t57pKphQ3d*7ZST

一致性 Consistency

一致性是指，不管系统是否发生异常，所有的Reader读到的相同键值下的数据都是一致的、有序的。对Pravega来说，不管是tail read还是catch up read都满足一致性。

有序性 Ordering

有序性是指Reader读取Event的顺序要和Event被写入的顺序保持一致。对Pravega来说，数据与应用定义的路由键(routing key)一并写入Stream，Pravega根据路由键的哈希值将写入操作分配至不同的Segment。Pravega保证在路由键内部的有序性，即针对在同一个路由键的数据写入是保证有序的。Pravega的有序性同样保证了读客户端在数据读取发生异常时，仍然能够进行有序的恢复重放。

有了以上三个特性，Pravega得以进行有效的数据重放，保证了at-least-once语义。

Pravega是如何防止数据重复的

Pravega实现了at-least-once语义，但是为了要更进一步得满足exactly-once语义，我们还需要避免数据重复。所有的写入操作需要保证重复数据的消除，也称为幂等性(idempotency)。由于分布式环境下，数据交互复杂导致故障的发生位置众多，实现需要考虑诸多细节，这也是exactly-once实现的难点。这需要从读写两方面进行控制。

Pravega中的Reader每成功消费Segment中的一条数据后都会将数据的位置信息以SegmentID+offset的形式写入State Synchronizer（可参阅之前的文章）中进行持久化。这样的SegmentID+offset的对应关系构成了Pravega的StreamCut（可参考http://pravega.io/docs/latest/streamcuts/），代表了当前Stream读取位置的状态信息。对于一个ReaderGroup中的所有读客户端而言，ReaderGroup实现了Checkpoint机制，以Map<Stream, StreamCut>的形式保存了Reader读取的所有Stream的一致性的状态信息。确定了一致的恢复位置，我们就保证了故障发生时，Segment中的数据也仅被读取一次。

Pravega的Writer内部自带一个ID，在Writer与Pravega服务发生重新连接时可以通过ID定位到最近一次成功写入Stream中的Event，Writer会以block的形式将Event批量追加写入到Stream中，当block成功写入后，Writer会发送一个block end指令，指令中包含number of events written和last event number。当Writer与Segment Store之间发生断开重连时，Segment Store通过Writer ID并通过握手将last event number传递给该Writer，这样Writer就能知道应该从哪里开始发送Event。这也是分布式系统中最常用的保证幂等性的方法之一。

然而Pravega中Writer的ID是无法保障永远不变的，一旦Writer发生异常崩溃了，新起的Writer将会生产一个新的Writer ID，所以在考虑Writer发生故障的情况下，我们就需要将Writer ID与Segment Store解耦。

因此，Pravega使用事务来保证数据写入时的exactly-once。Event将会被批量地写入事务中，这些Event要么同时被提交，要么同时被丢弃。

Pravega的事务（Transaction）

接下来让我们一起来看一下Pravega中的事务。Pravega提供了Transaction API，支持事务性地数据写入，代码示例如下。

先回顾一下将普通Event写入Pravega Stream的操作：

// 创建一个client factory, 一个 writer 然后 写入event
try(ClientFactory clientFactory =
      ClientFactory.withScope(scope, controllerURI) {
    EventStreamWriter<String> writer = clientFactory
         .createEventWriter(streamName,
                            new JavaSerializer<String>(),
                            EventWriterConfig.builder().build());
    writer.writeNext("Key 1", "Hello");
    writer.writeNext("Key 2", "World!");
}

接下来是将Event事务写入Pravega Stream的操作

Transaction<String> txn = writer.beginTxn();
txn.writeEvent("Key 1", "Hello");
txn.writeEvent("Key 2", "World!");
txn.commit();

API很简单，只需调用beginTxn()开启一个Transaction，使用commit()方法进行事务提交即可。下面我们以一个例子进行说明其内部实现。

如图，Stream有3个active的Segments，当在该Stream上创建Transaction时，该Transaction会从Pravega的元数据库Zookeeper中读取相应Stream的所有的*active*的Segment集合（保证一致性），同样会分配3个对应的Transaction Segment（也称为Shadow Segment）。

当事件被写入到事务中时，它被路由并分配给与Stream相同的编号的Segment（即分配给Stream中的Segment 3的事件将被分配给Transaction中的Segment 3），并与Stream类似，事件本身将附加到事务的Segment的尾端。在事务处理的过程中，Transaction Segment并不对读客户端可见，保证了隔离性。

提交事务后，所有Transaction的Segment将自动附加到Stream中对应的Stream Segment，由于Pravega的持久化特性，数据也就被持久化了。如果事务终止，则其所有的Transaction Segment以及其中的数据都将从Pravega中删除，保证了事务操作的原子性。至此，分布式事务的ACID四大特性均得到满足。

熟悉Kafka的读者可能会发现，Pravega的Transaction和Kafka的Transaction的实现方式不同，Pravega通过副本的方式创建Transaction Segment，直到Transaction Segment提交合并入真正的Segment中后，Reader才能开始消费Transaction写入的数据，然而Kafka直接是将Transaction的Event写入Topic的Partition中，并且允许用户额外配置事务隔离级别满足不同需求。因此对比Kafka，Pravega存在以下优势：

Kafka中被终止(aborted)的Transaction会残留在topic partition中，这就导致了磁盘空间和IO带宽资源的浪费，而Pravega的Transaction使用了一个临时的Segment，当Transaction abort之后，临时的Segment就将被回收。
Kafka中Transaction的Event是直接写入partition中的，当Kafka的隔离级别为read-commit时，Reader在尝试读取一个open Transaction中的数据时会发生阻塞等待，直到Transaction完成(committed或者aborted)，而Pravega中，未提交的Transaction以一个临时的Segment表示，在提交成功前，Reader端是无法感知到Transaction Segment的存在的，因此Reader不会被阻塞。

Pravega Stream的弹性伸缩机制会对事务产生影响，在Writer端负载偏高时，Segment会相应做split操作，原先的Segment会被seal，同时生产两个新的Segment，细心的读者会发现这里会存在问题，倘若Writer正在往Transaction的Transaction Segment中写数据，如果此时Segment发生了split，就会发生与Transaction Segment不一致的情况，当我们合并Transaction Segment的时候就会发现找不到相对应的Segment。Pravega进一步实现了Rolling Transaction的机制来将Stream的伸缩和Transaction解耦，让他们同时工作互不影响。有了Rolling Transaction的支持，用户既能享受Pravega Stream的弹性伸缩机制也同时能保证exactly-once的支持。

Pravega的Transaction还包括以下API：

API	描述
getTnxId()	获取Transaction UUID
flush()	等待write操作成功
ping()	跟新Transaction的等待时间
checkStatus()	查询Transaction的状态 `Open`, `Committing`, `Committed`, `Aborting`, `Aborted`
commit()	提交
abort()	中断Transaction并且丢弃所有Events

Pravega与Flink的端到端exactly-once语义

设想文章最初提到的ETL场景，一个读客户端应用首先从Pravega中读取到数据A，对数据进行处理F(A)，此时如果在执行F(A)时读客户端应用发生故障重启，因为A已经被成功消费过了，State Synchronizer中的元信息已经更新，恢复之后将导致A数据的丢失。这一例子说明了ETL系统端到端的exactly-once仅靠流存储本身无法保证，需要配合处理端进行端到端的同步才能实现。

对于一般的端到端的exactly-once实现，ETL的三个组件要分别达到如下的要求。

输入端要支持固定位置的重放
流处理系统的容错处理保证任务只产生exactly-once的输出
输出端要有事务性的支持

为了解决这个问题，Pravega团队与Apache Flink展开了深入的合作，实现了Pravega Flink Connector连接器与Flink通信，使得用户能够在Flink中调用API，对Pravega进行数据的读写，更多详细内容可以期待Pravega系列之后的一篇文章。在这篇文章中，我们将重点介绍：如何配合Pravega提供的Checkpoint和事务机制以及Flink提供的Checkpoint，实现Pravega->Flink->Pravega的exactly-once。

熟悉Flink的读者应该知道，Flink在1.4.0版本之前通过Flink的Chandy-Lamport算法实现的Checkpoint机制，做到了Flink应用内部的exactly-once语义。然而，使用Checkpoint也意味着当时的exactly-once的应用是有条件的，只有每条消息对于Flink应用状态的影响有且只有一次才能满足exactly-once，例如数据流过Flink直接写入到数据库中的无状态应用是无法保证exactly-once的。Flink在1.4.0版本引入了TwoPhaseCommitSinkFunction，通过两阶段提交（2PC）协议对接事务性写客户端解决了写入的幂等性问题，从而支持了Kafka 0.11+以及Pravega作为输入输出端的端到端的exactly-once语义。

整体结构

从整体设计来看，Flink和Pravega实现端到端exactly-once有以下四个步骤：

当Pravega作为Source时，Flink的每个Checkpoint Event会触发Pravega ReaderGroup的Checkpoint
当Pravega作为Sink时，每两个Flink Checkpoint之间，会创建一个Pravega Transaction
当Checkpoint完成时，提交Transaction到Pravega Stream
Flink异常恢复后，尝试重新提交pending状态的Transaction或是恢复到最新Checkpoint

Pravega的PravegaFlinkWriter并没有直接继承Flink提供的TwoPhaseCommitSinkFunction，而是继承了其父类RichSinkFunction，从而用统一的入口支持了exactly-once和其它的语义。PravegaWriterMode中提供了三种写入模式：BEST_EFFORT, ATLEAST_ONCE以及EXACTLY_ONCE。其中的EXACTLY_ONCE模式即使用本文描述的事务性写入。代码如下：

protected AbstractInternalWriter createInternalWriter() {
  Preconditions.checkState(this.clientFactory != null, "clientFactory not initialized");
  if (this.writerMode == PravegaWriterMode.EXACTLY_ONCE) {
    return new TransactionalWriter(this.clientFactory);
  } else {
    ExecutorService executorService = createExecutorService();
    return new NonTransactionalWriter(this.clientFactory, executorService);
  }
}

EXACTLY_ONCE模式的实现依然遵循Flink Checkpoint的两阶段提交协议，在Flink本地算子快照时提交本地事务（进行pre-commit一阶段提交），通过Flink JobManager协调完成投票，当一切正常时，通知各算子完成Checkpoint，并最终提交事务。若存在失败则Checkpoint也将失败，视具体情况进行相应的处理，确保数据仅处理一次。

值得一提的是，投票处理的过程是异步进行的，不会影响正常的数据读写线程，对整体处理性能的影响较小。

故障恢复

针对其中的4，由于异常同样可能在流处理进行的各阶段发生，接下来将具体介绍以下三种情况出现异常时的处理方法。

Flink写入Event时发生异常

上图中，Checkpoint-1与Checkpoint-2均成功完成，但在Checkpoint-2之后的Event往Transaction-3中写的时候发生了异常。此时，Flink将从Checkpoint-2恢复，这时候由于没有Checkpoint-3因此Transaction-3也不会被提交到Pravega Stream中，因此Flink程序就可以从Checkpoint-2恢复重新在新的Transaction内写Event，不会发生数据重复，保证了从Flink到Pravega的exactly-once

Flink本地快照时发生异常

上图中，Checkpoint-1成功，当在Checkpoint-2的时候由于有些算子快照时失败。此时，程序从Checkpoint-1恢复，没有成功提交的Transaction-2和Transaction-3被丢弃，保证了exactly-once。

Flink Checkpoint成功，但是Transaction提交失败

上图中，Checkpoint-2成功，但在Transaction-2往Pravege Stream中提交的时候由于网络原因提交失败，这种情况下无需让Flink恢复到Checkpoint-1，而是只需重新提交Transaction-2即可，如下图：

实例展示

接下来我们通过一个实际的例子，展示并验证Flink写入Pravega的exactly-once。

首先我们保证Pravega的运行。Pravega可以选择部署standalone版本，具体步骤可以参考之前的文章。Flink可以按照官网推荐步骤进行部署，然后提交任务Jar包运行。也可以不进行部署，Flink会自动创建本地的ExecutionEnvironment运行，本例选择后者。
下载并构建pravega-sample，

$ git clone https://github.com/pravega/pravega-samples
$ cd pravega-samples
$ ./gradlew installDist
$ cd flink-connector-examples/build/install/pravega-flink-examples

该目录的bin/下有两个程序，exactlyOnceWriter和exactlyOnceChecker

exactlyOnceWriter是一个Flink应用，会生成一组整数的Events（默认为1～50）写入Pravega，并且会在26位置处制造一个人为的模拟异常，同时会以100毫秒为周期进行Checkpoint操作，如果出现异常应用就从最近Checkpoint恢复。

exactlyOnceChecker应用则是一个简单的Pravega Reader应用，检测写入Pravega中的数据否重复。

我们首先在一个命令行窗口中启动exactlyOnceChecker

$ bin/exactlyOnceChecker --scope examples --stream mystream --controller tcp://localhost:9090

然后在另一个窗口中运行exactlyOnceWriter开始往Pravega中写数据，我们先不开启EXACTLY_ONCE模式，将--exactlyonce 参数设置为false，此时默认为ATLEAST_ONCE模式。

$ bin/exactlyOnceWriter --controller tcp://localhost:9090 --scope examples --stream mystream --exactlyonce false

观察exactlyOnceWriter的输出，会产生类似如下内容：

...
Start checkpointing at position 6
Complete checkpointing at position 6
Artificial failure at position 26
...
Restore from checkpoint at position 6
Start checkpointing at position 50
Complete checkpointing at position 50
...

我们发现第一次Checkpoint在6的位置成功，同时数据继续往Pravega Stream中写入，直到26的位置，我们模拟了一个Flink Transaction的异常，导致了应用从最近的Checkpoint点恢复，程序又开始从6位置继续往Pravega中写数据直到结束。由于没有开启Pravega的EXACTLY_ONCE模式，7～26的数据就会重复写入Pravega。

再观察exactlyOnceChecker的输出，的确监测到了数据重复

============== Checker starts ===============
Duplicate event: 8
Duplicate event: 7
...
Duplicate event: 23
Duplicate event: 24
Found duplicates
============== Checker ends  ===============

现在以EXACTLY_ONCE的模式重新运行exactlyOnceWriter。

$ bin/exactlyOnceWriter --controller tcp://localhost:9090 --scope examples --stream mystream --exactlyonce true

exactlyOnceChecker的输出如下：

============== Checker starts ===============
No duplicate found. EXACTLY_ONCE!
============== Checker ends  ===============

在EXACTLY_ONCE模式下，Event 1～6 是一个Transaction，当Flink标记Checkpoint成功后Event 1～6才会被提交到Pravega Stream中，而在26处Flink Transaction发生了异常，这时Transaction中的Event 7-26并没有成功提交到Pravega Stream中，而是被丢弃了，因此在exactlyOnceWriter从6的Checkpoint处恢复后，重新从7开始写入数据，从而保证了端到端的exactly-once。

Pravega 系列文章计划

Pravega 根据 Apache 2.0 许可证开源，0.5 版本将于近日发布。我们欢迎对流式存储感兴趣的大咖们加入 Pravega 社区，与 Pravega 共同成长。本篇文章为 Pravega 系列第七篇，系列文章标题如下（标题根据需求可能会有更新）：

实时流处理 (Streaming) 统一批处理 (Batch) 的最后一块拼图：Pravega
开源 Pravega 架构解析：如何通过分层解决流存储的三大挑战？
Pravega 应用实战：为什么云原生特性对流存储至关重要
“ToB” 产品必备特性: Pravega 的动态弹性伸缩
取代 ZooKeeper！高并发下的分布式一致性开源组件 StateSynchronizer
分布式一致性解决方案 - 状态同步器 (StateSynchronizer) API 示例
Pravega 的仅一次语义及事务支持
与 Apache Flink 集成使用

作者简介

滕昱，就职于 DellEMC 非结构化数据存储部门 (Unstructured Data Storage) 团队并担任软件开发总监。2007 年加入 DellEMC 以后一直专注于分布式存储领域。参加并领导了中国研发团队参与两代 DellEMC 对象存储产品的研发工作并取得商业上成功。从 2017 年开始，兼任 Streaming 存储和实时计算系统的设计开发与领导工作。

周煜敏，复旦大学计算机专业研究生，从本科起就参与 DellEMC 分布式对象存储的实习工作。现参与 Flink 相关领域研发工作。

赵凯皓，现就职于 DellEMC，从事流存储和云原生相关的设计与开发工作。

参考资料

创作场景

流处理系统正确性基石：ExactlyOnce 的设计和实现