Data Artisans Streaming Ledger ——流数据处理中串行化的ACID事务

介绍

Data Artisans Streaming Ledger，在data Artisans的River Edition上已经可用，提供串行化（一致性事务处理机制的最高级别）的ACID的语义，作为一个依赖库来处理事件流上多个共享的状态/表。

在data Artisans Streaming Ledger的串行化事务并行机制下，用户可以定义多张表与事务驱动的数据流相连接，并可以定义灵活的事务处理逻辑来处理事务事件，读取或者更新表中不同的数据行。相对于，在一个时间点内只能修改一个表中的一个单独的键（类似于Apache Flink或者其他流处理组件），streaming ledger的事务性的函数可以一次性操作多张表中的多个键。相同的表可以被不同的数据流共享，所有的这些都不会影响到数据处理的性能和一致性。

想象从键值存储（单行数据更新具有一致性）到关系型数据库中都是通过全串行化的事务机制来操作数据行和数据表。同样的方式，data Artisans Streaming Ledger在当前的exactly-once的有状态的流处理模型的基础上引入了针对于多行，多状态，多流交叉处理的事务处理机制；这在当前其他的流处理框架上都是不曾实现的。

通过data Artisans Streaming Leger，用户现在可以将一类新的应用迁移到流处理上：比如说那些需要依赖于关系型数据数据库的ACID事务特性进行数据处理的应用。这些应用可以平滑的迁移到流处理的框架中，并充分利用流处理的优点。由于data Artisans Streaming Leger提供很高的吞吐量，大量的应用可以直接有效的转为成一致性的流处理应用。

流数据，有状态的流处理，以及它的局限性

数据流处理（或者说事件流处理）就是在一系列的事件数据上进行处理的模型。典型的是和一些商业行为有关（比如说：一个顾客在购物车买了一件商品，或者一笔转账交易）。这些事件流可以被实时的处理或者事后处理，从而使得流数据处理成为线上实时交易的一个强有力的工具，当然它也可以用于对有延迟的数据或者历史数据的分析。“流处理引擎”就是驱动流式数据应用的系统，Apache Flink就是这样一个强大的流数据处理引擎，可以运行最大规模的，可以满足需求最多的流处理的应用。

像Apache Flink这样强大的流处理引擎，可以用来将数据处理状态保存持久化并进行管理，这些状态是从事件流中获取的。这些状态其实就是流数据处理所发生的一切中的视图。对应这些状态的样例包括：

1） 用户对每件产品在过去5分钟内的关注次数（分析）

2） 反欺诈交易中在线模式识别（复杂事件处理）；

3） 每个网上用户当前的交互行为的状态（应用状态）；

4） 资金交易中帐目计算（准确数据计算）

这些应用中，流处理引擎处理的数据通常都是存放在数据库中。流处理的输入是一系列持久化的消息事件，可以被存放在日志中，消息队列，文件系统，或者这几者的组合中。你可以在这里找出它与关系数据库管理系统的相似之处，输入事件流对应是事务的write-ahead日志，流处理引擎的状态对应于数据库中的表 ——当前所有处理事物的视图。

基于流计算来实现这个架构，而不是基于数据库管理系统来实现，优点在于：

1） 流计算比SQL更加灵活，它可以进行实时计算和数据洞察，这些是数据库做不了的； 并且它可以和其他的框架或者依赖库进行交互，比如说统计或者机器学习相关的库；

2） 流计算的全异步式的以及事件驱动的本质使得它在大多数情况下，相对于数据库的请求/响应机制来说，可以获得更高的吞吐量；

3） 当启动一个新的流计算应用消费相同的数据源，流计算的视图可以被不同的租户应用到流程中的任意位置。新应用不需要和其他的应用交互，也不需要提前规划就可以创建；

4） 流计算可以很灵活的解耦出来，成为一个简单的并且模块化的基础架构；

并行的分布式状态化的流处理

流计算应用的状态（当前被计算的视图），一般被保存到key/value存储的表格中，可以增长到很大的量。在Apache Flink中，某些应用在流处理中保存了数十TB的状态数据。

Apache Flink共享这些状态，是为了能够在多台服务器上进行并行计算，这个策略和大多数据库管理系统以及key/value存储类似。表的行键决定了数据在那一台服务器/处理引擎上存储。

状态相关的并行操作（转换操作）一般都是和状态数据在同一位置：转换算子并行的各个实例在状态数据的对应分片上执行，事件数据被发送到不同的算子function上，每个算子的function接收到事件数据后都会和它对应的状态分片交互。因而，所有的状态的写入和修改都是分片本地化的。

执行模型“将计算移动到数据上”: 触发计算的事件被发送到对应的状态/表的分片上。对比于“将状态数据拉取到计算任务上”的模型而言，这个模型是非常高效的，提升了事件处理的吞吐量，并提供了强一致性。每个分片上都有一个高效的reader/writer, 并且和本地分片单独的事件交互都能保证一致性。也就是说，在数据流中进行“单行事务操作”（像多数key/value存储以及关系数据库中）是非常高效的。

当前流处理引擎的局限

流数据处理上串行化的多键多表事务机制

data Artisans Streaming Ledger在Apache Flink上进行了扩展，克服了这些局限性：使得flink可以通过串行化的事务机制来处理多流多表的情况。可以把它看作是在key/value存储，甚至跨多key/value存储上进行多行事务处理。类似于在关系数据库上通过ACID事务处理机制处理事务一样，每一个function也是通过ACID的语义在串行化的事务机制上对流计算中的各个表进行修改，这样就保证了类似于关系数据库上的数据一致性。这使得可以直接将一些事务处理的应用直接迁移到流处理的框架上，正如我们所知道的，目前还没有其他的流处理框架有这种能力。

data Artisans streaming ledger无缝的植入到了apache flink中，它使用了Flink的状态来保存表，所以不需要构造其他的存储系统。不像传统的基于锁的事务机制，data streaming ledger的事务机制是没有使用分布式锁，它处理数据速度快，并且易于可扩展；不像基于时间戳的乐观锁并发机制，当出现冲突时事务不会失败，它不会在出现失败时候进行不断重试；相反，在该实现中通过将时间戳，watermark，以及乱序调度算法相结合，满足了事务处理的隔离性。

表，事件流，和事务函数

构建data Artisans streaming ledger应用的核心概念是：表，事件流，事务函数，以及结果输出流。为了能够很好的解释这些概念，我们列举一个在账户之间进行转账交易的例子。

表

Streaming ledger应用的状态数据都是保存一个或者多个key/value的表中，它们是通过事务函数来进行更新。一张表是由多行组成，每一行被一个唯一的键标识。

表被存放在Apache Flink的状态中：表以检查点的方式被持久化。这些数据可以被全部存放在内存或者RocksDB中，根据配置的state backend的不同。表中的键和值可以是任意类型。

如上所示中有两张表：账目和资产，表中的每一行都通过一个唯一的ID进行标识。

事务事件流

根据流计算处理模型，事务通过事件驱动，这里称之为事务事件。事件在并行的数据流中流动，触发相应的事务。Streaming ledger中可以同时运行多个事件流，并包含多个事件类型。在我们的样例中，数据流中包含了转账事件和储蓄事件。事件将参数传递给事务，例如：需要被修改的行，或者修改所涉及到的信息。在以上的样例中，转账事件中包含了源账户和目的账户的ID，ledger表中的有这些ID的行将会被更新，加上被转入的金额，或者可能进行转账前提条件判断（例如：最小账户余额等等）。

事务函数

每一个事务事件流都有一个事务函数，事务函数中包含了事务处理的逻辑。事务函数就像是Apache Flink中的转换函数，当接收到对应的事件时就会被触发，对数据行进行操作。和Apache Flink的函数的主要不同点在于不同的事件函数共享对数据表的连接，并且可以在同一时间对同一张表中的多行多key进行修改，并保证严格的一致性。

事务函数可以包括复杂的业务逻辑，可以通过Java/Scala来实现。这就使得streaming ledger在复杂的业务逻辑中如需要将事件和状态联合处理的情况下，可以被用到。可以把事务函数认为是一个强大的存储过程，直接在系统中执行数据管理，而不是将数据拉出来，更新完再放回去。

在我们的样例中，共有两个事务函数：一个用于进行账户储蓄，另一个用于进行转账。这些事务函数根据不同的事件类型触发，并根据业务逻辑对表中的行进行调整。转账函数也有可能不对数据表进行任何修改，例如源账户没有一个最小的额度可以转出，或者资产不可用。

结果输出流

事务函数可以选择性的将事件发送到某个结果输出流中，这样可以用于显示处理成功或者失败。如，是否满足事务函数检查逻辑的前提条件。

结果输出流也可以将更新完的结果输出到外部系统中，从而来创建一个输出结果镜像表。当这个外部系统（比如一个key/value存储）同时也被其他用户使用时，Apache Flink以及data artisans streaming ledger做了的很多工作来保证事务处理的一致性。

在以上的样例中，转账函数会将结果事件输出，从而表示这个转账交易是成功发生了还是被拒绝了。

ACID语义

Streaming ledger的事务处理机制的实现基于串行化的事务隔离机制，并遵循ACID语义

A – 原子性： 事务将所有的修改都在一个原子操作内完成。事务中所涉及的修改动作要么全部执行，要么都不执行；

C — 一致性： 事务将表中的数据从一个一致状态转换到另外一个一致状态；

I— 隔离性： 每一个事务的执行就像是它是唯一的一个操作表的事务一样。数据库有不同的隔离级别，通过不同的机制来保证。data artisans streaming ledger是通过最高级别的隔离机制：串行化。

D—持久性：事务对表数据的修改是持久的并且不会丢失。持久性在Flink的其他应用中也是使用同样的方式来保证 — 通过持久化的数据源和checkpoints。由于流计算天然的异步处理的机制，输出结果的持久性只能在checkpoints之后来实现（详细的, 请看下一章节关于一致性模型的介绍）。

用例

我们在data streaming artisans ledger上应用了一个现实中的用例：在流处理中进行join操作处理的原型。Data streaming artisans ledger解决了流数据相交处理，以及状态一致性整合的问题。接下来我们将选择这方面的用例来介绍。

多表事务处理

就像上面列举的一样，data artisans streaming ledeger的典型的操作就是在流处理的应用中对多个状态进行处理。就像数据库中的ACID事务处理机制一样，数据库的操作要求保证原子性，事务中所看到状态的视图应该保持一致性和隔离性，这样的话就不会发生不遵守规则（ACID）的事情了。

这些在具备ACID的关系型数据库中是很容易保证的，但是在当前的流处理引擎中却没有保证。streaming ledger给你展现的就像它的核心概念中描述的：多张表在流处理中被事件进行事务性的修改，每一个事务的修改动作都是和其他事务相隔离的，并可以保证可串行的一致性。

样例：在账户，帐目上进行资产转移

在这个样例中， streaming ledger一共处理两张表：“acounts”和“ledger”。streaming ledger消费事件流，事件流描述了账户，帐目资产，或者两者都有，可能也包括一些其他前提条件。

当一个事件进来后，事务函数读取到相关的数据行，检查前提条件，并决定是否处理该事件或者拒绝处理。在前面的样例中，它更新了数据表中的相关的行。这些案例中，事务函数可能选择性的产生结果事件，用于判断转账交易是成功还是失败。

这个样例可以很容易扩展到在多条数据流中进行不同的操作：储蓄，转账，查帐。streaming ledger消费不同的事件流，针对事件流中的不同事件类型进行不同的处理，并根据不同的事件类型输出到不同的结果流中。比如对于查账事件来说，不会触发对数据的修改，只是将帐目从数据表中读取，并将结果写入到输出流中。

位置保持/数据整合

streaming ledger可以用来将操作以及独立的算子的操作视图统一到一个源状态表中。这些操作被当作事件流来接收，结果保存到streaming ledger的表中，并可以选择性的通过输出流进行镜像保存。

事务处理逻辑遵守将事件以及当前的处理状态进行整合的规则。串行化的一致性事务语义保证了没有操作会丢失，并且没有任何修改会导致不一致性状态。

样例：流处理事件交易中的实时位置保持

不同的应用（如交易台，客户机）都会基于位置数据子集生成一个快速的本地视图，通过这些视图来展示处理性能，流畅度。影响位置的交易事件或者其他事件都会写入到流中。在stream ledger中对流事件进行处理前，这些事件可以被前置处理，这些在Apache Flink中也是同样的。

streaming ledger将位置信息数据源保存到它管理的表中。它从事件流中读取数据进而对表更新。事务函数中的应用逻辑决定怎样去更新表中的位置信息——在应用逻辑中检查前提条件（例如是否根据前提条件拒绝或者取消交易，或者根据事件以及位置信息的一致性视图决定怎样更新表中数据）。同时更新表中的多行数据或者多张表，保持一致性，是保证位置状态可靠的关键。事务逻辑可以通过java/scala的应用代码来灵活实现，从而支持高度复杂业务逻辑，以及对各种库的运用。

每次更新的结果可以被写入到一个结果输出流中，然后被一个外部系统消费到，从而将标定好的结果进行“镜像”保存，也使得处理结果可以被另外一个不支持可串行化的事务的系统（如数据库）访问到，从而用于支持简单灵活的业务逻辑，或进行大吞吐量的事务更新。

另外，基于streaming ledger的应用也可以接受交易事务处理过程中更晚阶段的保存在其他系统中的处理结果，并且为整个交易的链路中保持一个一致性视图，显示整个交易是成功了还是失败了。

多方式join处理&特征向量装配

data Artisans streaming ledger支持在不同的数据流中共享表以及对不同的数据行进行处理，从而使得对于共享表的多种方式的join操作有了一个完美的解决方案。

一些数据流可以用来构建或者更新表，其他的数据流可以对表中不同数据行进行join操作，产生join的结果。

样例：机器学习应用中的流式特征向量装配

机器学习模型对不同的事件进行区分（例如标记信用卡事务是欺诈或者正常），一般情况下需要对这些事件进行特征标记，从而作为模型预测的输入。在信用卡反欺诈中，这些特征可以是，比如说：近几天内的使用频率，一天内信用卡使用次数，使用信用卡的地理位置，或者信用卡交易的对方帐户。对这些特征进行实时计算，以及将这些特征取值装配到特征向量里面，用streaming ledger来进行处理是非常适合的。

在这个样例中，streaming ledger通过三张表保存了地理位置，帐目，以及源账户和目的账户对应关系。这些信息可以通过信用卡事务事件进行更新，也可以通过另外一个系统产生的数据流更新。每个事件都携带了相关特征的key，这些数据将被收集起来，并对所有的数据表进行join操作，最终产生特征向量作为机器学习模型的输入。

架构以及API设计

基于data Artisans streaming ledger开发应用是非常简单的事情。它的API对于之前使用过流计算框架或者使用过数据库的用户来说十分自然。

我们引用的样例和上面“跨表流事务处理”的例子十分相似：

两张表：Accounts 和 Book Enties；

两个事件流：Despoits 和 Transfers；

Despoits事件流将数值放入到Accounts和Book中；

Transfers事件流在满足前提的条件下，自动的将数值在account和book entries中移动；

1. 创建一个Flink DataStream程序并建立事件源

Data Artisans streaming ledger 的API 完全兼容Apache Flink DataStream API。第一步是创建一个DataStream API的应用程序，包含执行环境变量以及事件源数据流。

2. 定义事件范围和事件表

下一步是定义事件范围和表。简单来说，我们的表使用字符串类型来作为keys（account ID, entry ID）， 长整型来作为当前的value；

3. 定义事务函数以及键

接下来，我们指定对于某张表以及指定的行键，那个事务函数将会被应用，

‘apply（..）’函数用于包含事务函数的业务逻辑。

对于数据表中的每一行，我们通过方法’.on(table, key, name, type)’来调用操作：

‘table’用于指明要操作的数据表；

‘key’是指事务函数从输入事件中获取来的；

‘name’是指特定行的逻辑名字；

‘type’用于定义对行的“只读”,“只写”,“读写”等操作类型，这是一个可选项。“读写”是最常用的选项；

4. 实现事务函数的逻辑

事务函数包含了业务处理逻辑: 是否更新或者怎样更新表中的行，以及怎样发送计算结果等等。

这些事务函数对于每一个被读取或者更新的数据行，都会完成一个状态操作。为了将对状态操作与数据行，键联合在一起，我们在上一步根据它们的名字进行了定义声明。

我们仅展示下对于TransferFunction的实现：

5. 选择性获取的结果输出流

作为一个可选的步骤，我们可以抓取“TransferFunction”函数处理结果的事件流，通过骤3中定义和创建的result标记。

一致性模型

Data atrisans streaming ledger 通过串行化的事务隔离机制实现事务的ACID语义。这是数据库管理系统中最强的隔离机制。这些语义使得事务就像串行化一样执行，每个事务都是单独的执行，只有当前一个事务完成，下一个事务才会开始执行，后面一个事务将看到前面事务的所有修改。

现在的挑战问题是在不基于事务一个一个执行机制的前提下，提供事务的ACID语义，这并不是不可实现。data Artisans Streaming Ledger通过将事务中的逻辑时间戳机制和单个操作的无冲突调度结合起来达到这一目的。

持久性

流计算天然是异步处理的，很多流计算系统考虑到性能上的因素，也都执行异步的处理。Apache flink，执行异步的快照机制来进行持久化以及恢复；

由于data Artisans streaming ledger是在Apache flink上面的一个库，它的状态在flink的快照完成时将被更新持久化，因此持久化的语义依赖于streaming ledger应用中sink算子的类型：

1）对于extract-once sink的算子，一旦结果事件从结果流中读出，强持久化机制就能得到保证。但是这样的sink算子一般会导致额外的延迟；

2）对于at-least-once的sink算子，事务的结果都会基于一个一致性视图，在重复计算中一个结果可能会被另外一个结果包含（在at-least-once中有时候会出现重复计算）。新的结果也是基于一个一致性视图，但是可能和上一个结果有所不同，因为它来自一个不同的串行化调度。

串行化化，线性串行化，严格串行化

Data artisans streaming ledger实现了通过“串行化”的隔离机制，在一般条件下，用户可以通过严格串行化实现并发语义。

严格串行化将“串行化”的特性以及线性化的语义结合起来，线性化这里意味着如果事件B是在事件A的处理结果从事件流中接收到之后才进入到事件流中，那么事件B对数据的修改一定是在事务A之后。简单的说，事务的执行顺序严格的遵守真实情况下触发动作；

由于流计算天然的异步处理，线性化得以实现主要依赖于以下两个条件： 1. 事务A的更改是持久化的，在事务B进入到数据流源之前。这个可以很容易的通过对结果输出流的extactly-once sink机制来实现；

2. 事务A和B可以在流计算恢复的时候进行按序重放。这可以通过log类型的数据源来实现，例如：Kinesis或者Kafka。当在两个事务上使用这种线性化的语义的时候，触发他们的事件将被添加到相同的数据分区或者分片中，从而保证了失败后进行恢复时可以按序重放。

性能测试

Streaming Ledger对比关系数据库管理系统

相对于关系数据库中ACID的事务处理机制，data Artisans Streaming ledger的串行化的事务处理机制有很多相似的地方，但是也有着很多的不同：

1. 事务不是被客户端通过请求/响应的方式触发。相反，在streaming ledger中事件都被放到了事务事件数据流中被消费， 这遵循了流计算的异步/响应的模式，从而使得streaming ledger获取更高的吞吐量，从而和流计算管道天然兼容。

在此基础上也可以建立一个请求/响应的客户端，通过一个请求/响应的消息队列，以及与之关联的ID来处理请求事件，并给出响应。

2. 事务并不是在SQL中指定，而是在代码程序中。这使得streaming ledger和传统的关系型数据库不同，当涉及复杂的业务逻辑的时候，仍然可以通过异步/响应的机制来处理事务事件以及表数据，

3. 事务类型需要被提前定义好，这和数据库系统类似，需要提前把事务通过“存储过程”的方式定义好。

4. 事务读取或者更新的数据行的键必须可以从事件中获取。这些键不能被事务的业务逻辑动态计算。也就是说事务的业务逻辑不能看到和依赖于某些数据行/值，当它要去操作别的数据行/值的时候。为了满足这种动态性，事务函数需要指定哪些数据行需要被读取/更新，那些是不需要的。

这个限制对于异步调度/执行模型来说是至关重要的，从而也使得streaming ledger可以在高并发的情况下取得高吞吐量。

Streaming Ledger是如何取得这样的性能的？

data artisans streaming ledger背后处理数据的技术和关系数据库系统中事务处理的技术是不同的，streaming ledger事务机制的实现没有使用分布式锁，也没有通过多版本并发控制，在事务冲突的时候。分布式锁机制是非常消耗性能的，并且基于锁的实现需要额外的守护来预防死锁。乐观锁的通过时间戳和冲突检查来进行并发控制，可能会导致事务被丢弃，并不断重试。在这种解决方案中，重试会消耗掉系统资源的大多数时间，使得系统在事务的压力下只能取得较低的效率。

data artisans streaming ledger是基于流处理模型，尤其是在Apache Flink的基础上，来设计获取高吞吐量的： 1）事务被作为函数放入到系统中，在流处理中被事务事件触发。这个过程都是异步的，事件驱动的；并且和多线程，多连接处理不同，需要花费大多数时间去等待，就像是在典型的数据库管理系统中通过请求/响应的方式进行会话交互一样。

2）数据行的读取/修改并不是通过分布式锁的方式来管理，而是通过调度的方式来避免冲突。

3）整个实现使用了一种轻量级的同步化的逻辑时钟。事务处理的准确性不会因为延迟或者时钟迁移而受到影响，只有数据行读取的时延会被影响。当进行数据操作调度时，Flink的watermark机制被用来建立最小的时间保证。

4）Flink灵活的检查点机制以及状态管理，被用来实现对乱序的数据的处理，从而在保证持久化和extractly-once的语义上的基础上，保证了事务的串行化。

这种实现的可行性是对Apache Flink作为流处理引擎的能力的一种尝试。