支持流式处理ACID事务！Flink团队开源新作Streaming Ledger

Posted 2021-04-13 AI前线

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了支持流式处理ACID事务！Flink团队开源新作Streaming Ledger相关的知识，希望对你有一定的参考价值。

策划编辑 | Natalie

作者 | Stephan Ewen

译者 | 无明

编辑 | Natalie

AI 前线导读：近日，由 Apache Flink 创始人创立的 data Artisans 宣布开源 Streaming Ledger——一项激动人心的新技术，为基于流式架构的应用程序带来了串行化 ACID 事务支持。在这篇文章中，我们将解释为什么串行化 ACID 事务对于现代企业来说是一个巨大的挑战、data Artisans 如何找到解决方案以及如何将这项技术用于流式应用程序。

更多优质内容请关注微信公众号“AI 前线”（ID：ai-front）

流式处理的下一个演化步骤

在 data Artisans，我们目睹了数据流式处理的疯狂式增长，从早期阶段到快速增长的市场，预计到 2025 年将达到近 500 亿美元的规模。从 Apache Flink 创建之初，我们就坚信流式处理是一项可以为企业的任务关键型应用程序提供强大支持的技术。

随着越来越多的行业采用流式处理，并为越来越多的任务关键型应用程序提供支持，这项技术本身也在不断发展，以便为数据和计算的正确性提供更好的保证。

第 1 步：数据分析的分布式流式处理（“至少一次保证”）

第一个分布式流式处理器的主要目标是数据分析应用程序，提供了一种实时非精确和异步精确的分析方法。这种方式被称为“Lambda 架构”，流式处理器基于到达的数据提供不精确的分析结果，同时批处理器以小时或天为单位提供精确的分析结果。这种流式处理保证被称为“至少一次处理”。这是流式处理系统能够提供的最弱的正确性保证，是流式处理技术的第一步。

第 2 步：单键应用程序的分布式流式处理（“恰好一次保证”）

Apache Flink 率先采用了真正的有状态流式处理，提供了大规模的“恰好一次保证”。这样就可以通过能过提供强正确性保证的流式处理技术来构建分析类型和事务类型的应用程序，减少对 Lambda 架构和批处理器的需求。

如今，市场上有很多流式处理器都提供了强大的一致性保证，但仅适用于特定类型的应用程序，这些应用程序每次只更新单个键。也就是说，如果一个应用程序每次只更新单个银行账户余额，那就可以通过流式处理来实现，但如果应用程序要将资金从一个银行账户转移到另一个银行账户，就难以实现强一致性。

第 3 步：一般应用程序的分布式流式处理（“ACID 保证”）

随着 Streaming Ledger 成为 data Artisans 平台的一部分，用户现在可以构建同时读取或更新多条记录和多张表的应用程序，并实现 ACID 事务支持。

Streaming Ledger 在提供这些保证的同时，还能保持流式处理（恰好一次保证）的伸缩能力，而且不会影响应用程序的速度、性能、可伸缩性或可用性。

我们可以将 Lambda 架构的至少一次保证视为最终一致性的一种形式（因为批处理系统最终会赶上来）。Flink 提供的“恰好一次保证”类似于分布式键值存储系统为单键操作提供的一致性保证，而 Streaming Ledger 提供的保证类似于关系型数据库提供的 ACID 保证。

我们相信这是流式处理的下一个演化步骤，它为以正确、可伸缩和灵活的方式实现基于流式架构的应用程序打开了大门。

多键和多表事务简介

大部分关系型数据库管理系统会执行 ACID 事务，每个事务通过 ACID 语义在串行化的隔离级别下修改数据表，以此来实现完整的数据一致性。根据 ACID 语义，所有事务都是 Atomic、Consistent、Isolated 和 Durable 的。ACID 语义在金融服务或电子商务等行业中扮演着重要的角色。

让我们举一个经典的例子，假设有一个帐户余额表，基于这个表将一个帐户的钱转到另一个帐户上。为了确保事务的正确性，转账操作必须同时修改两个帐户或都不做修改（原子性），而且只有在源帐户有足够资金时才能进行转账（一致性），并且不存在其他可能导致错误结果的操作（隔离、无异常）。任何违反这些条件的操作都会导致资金丢失，最终导致账户余额不正确。

支持流式处理ACID事务！Flink团队开源新作Streaming Ledger

上图显示了如何在有状态流式处理器中实现这个示例。其中的帐户 ID 就是键，而且两个帐户位于不同的分片中。这两个分片对对方都没有访问权限或者对对方的状态持有一致的视图，这使得实现这一的框架变得相当复杂，因为需要通过下列方式在两个分片之间传递状态：

提供一致的状态视图
能够管理并发修改
确保修改的原子性

超越流式处理框架的恰好一次语义

data Artisans Streaming Ledger 基于 Apache Flink 构建，为跨多个表和单表多行的多个数据流提供执行串行化事务的能力。它可以被视为等效于键值存储系统（甚至是跨多个键值存储系统）的多行事务。Streaming Ledger 使用 Flink 的状态来存储表，所以就不需要额外的存储或系统配置。Streaming Ledger 应用程序的构建块由表、事务事件流、事务函数和可选的结果流组成。

有关更多信息，请下载白皮书（https://data-artisans.com/download-the-data-artisans-streaming-ledger-whitepaper ）。

data Artisans Streaming Ledger 为使用流式处理构建新的应用程序类型打开了一个大门，这类应用程序在以前只能依赖关系型数据库。现在，数据密集型实时应用程序（如欺诈检测、机器学习和实时交易定价）可以毫不费力地迁移到流式处理平台上。在下一节中，我们将通过具体的示例演示使用 data Artisans Streaming Ledger 进行开发的必要步骤。

Streaming Ledger 用例演示

data Artisans Streaming Ledger 非常适合用于处理涉及多个状态的用例，它支持对多个状态进行事务性的修改，这些修改彼此隔离，并遵循串行化一致性原则。

我们假设有一个实时应用程序，它负责在帐户和分类帐条目之间识别汇款模式。

这个应用程序需要维护两张 Flink 状态表：第一张表叫作“Accounts”，第二张叫作“Asset Ledger”。应用程序消费事务事件流，例如帐户之间、分类帐条目之间或二者之间的转账。当有事件进入时，不同的事务类型会被应用在每个事件类型上，然后访问相关行，检查前置条件，并决定是处理还是拒绝转账操作。对于账户之间的转账，它会更新表中的各个行。对于账户和分类账之间的转账，它会生成结果事件，表示转账是被接受还是被拒绝。下图显示了架构细节：

支持流式处理ACID事务！Flink团队开源新作Streaming Ledger

Streaming Ledger 公开了一个易于使用的 API，对于有过流式处理使用经验的用户和熟悉关系型数据库的用户来说，这个 API 可以轻松上手。在我们的例子中，我们使用了以下假设：

两个表：Accounts 和 Asset Ledger
三个事件流：存款、转账和余额查询
存款时将值写入 Accounts 和 Asset Ledger 表中
转账原子操作在 Accounts 和 Asset Ledger 之间转移值。

接下来，我们将逐步使用 Streaming Ledger API 来完成这个示例。Ledger 的 API 是开源的，还包含了一个串行（单节点）的实现（https://github.com/dataArtisans/da-streamingledger ）。

以下是通过 Flink DataStream 来创建事件源的方法：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<DepositEvent> deposits = env.addSource(…); 
DataStream<TransferEvent> transfers = env.addSource(…);

And this is how you can define the scope and tables of your programme:

// 定义事务范围
TransactionalStreams txStreams =
TransactionalStreams.create(“simple example”);

// 定义事务表
TransactionalStreams.State<String, Long> accounts = txStreams.declareState(“accounts”)
         .withKeyType(String.class)
         .withValueType(Long.class);

TransactionalStreams.State<String, Long> asset ledger = 
txStreams.declareState(“AssetLedger”)
        .withKeyType(String.class)
        .withValueType(Long.class);

接下来，你可以为每个流和表指定事务函数，以及选择将要访问到的数据行的键。

.apply(…) 函数本身包含事务的业务逻辑。

对于被访问的每个数据行，为它们添加一个额外的调用：.on(table,key,name,type)：

'table'表示要访问的表
'key'是一个函数，可以通过这个函数从输入事件中获取这一行数据的键
'name'是该行数据的逻辑名称（稍后可能会用到）
'type’限定对该行的访问属性，如只读、只写或读写访问。这是一种优化，其中 READ_WRITE 是最通用的选项。

// 定义存款事务
txStreams.usingStream(deposits, “deposits”)
   .apply(new DepositsFunction())
   .on(accounts, DepositEvent::getAccountId, “account”, READ_WRITE)
   .on(assetledger,DepositEvent::getassetledgerEntryId, “asset”, READ_WRITE);

// 定义转账

// 将句柄保存在结果流中，以备后用
OutputTag<TransferResult> result = txStreams.usingStream(transfers, “transfers”)
   .apply(new TransferFunction())
   .on(accounts, TransferEvent::getSourceAccountId, “source-account”, READ_WRITE)
   .on(accounts, TransferEvent::getTargetAccountId, “target-account”, READ_WRITE)
   .on(assetledger, TransferEvent::getSourceAssetledgerEntryId, “source-asset”, READ_WRITE)
   .on(Assetledger, TransferEvent::getTargetAssetledgerEntryId, “target-asset”, READ_WRITE)
   .output();

然后实现了包含业务逻辑的事务，决定是否以及如何更新数据行，以及生成哪些结果。

我们传给这些事务函数一个状态访问对象，访问对象负责读取或跟新每一行数据。为了将状态访问与数据行和键关联起来，对这些函数进行了与前一步相同的注解。

为简单起见，我们只给出'TransferFunction'的实现。

public class TransferFunction extends 
TransactionProcessFunction<TransferEvent, TransferResult> { 
 @ProcessTransaction 
 public void process(
    TransferEvent txn,
    Context<TransferResult> ctx,
    @State(“source-account”) StateAccess<Long> sourceAccount,
    @State(“target-account”) StateAccess<Long> targetAccount,
    @State(“source-asset”) StateAccess<Long> sourceAsset,
    @State(“target-asset”) StateAccess<Long> targetAsset) {

// 访问当前值的数据行
long sourceBalance = sourceAccount.read();
long sourceAssetValue = sourceAsset.read();
long targetBalance = targetAccount.read();
long targetAssetValue = targetAsset.read();

// 检查前置条件: 正余额和最小余额
if (sourceBalance > txn.getMinAccountBalance()
&& sourceBalance > txn.getAccountTransfer()
&& sourceAssetValue > txn.getAssetledgerEntryTransfer()) {

// 计算新的余额
long newSourceBalance = sourceBalance - 
txn.getAccountTransfer();
long newTargetBalance = targetBalance + 
txn.getAccountTransfer();
long newSourceAssets = sourceAssetValue - 
txn.getAssetledgerEntryTransfer();
long newTargetAssets = targetAssetValue + 
txn.getAssetledgerEntryTransfer();

// 写入更新过的值
sourceAccount.write(newSourceBalance);
targetAccount.write(newTargetBalance);
sourceAsset.write(newSourceAssets);
targetAsset.write(newTargetAssets);

// 触发包含新余额的正结果事件
ctx.emit(new TransferResult(txn, SUCCESS, 
newSourceBalance, newTargetBalance));
}
else {
// 触发包含未更新余额的负结果事件
ctx.emit(new TransferResult(txn, REJECT, 
sourceBalance, targetBalance));
    } 
  }
}

结论

data Artisans Streaming Ledger 将之前依赖关系型数据库的应用程序带入到了流式处理时代，进一步扩展了流式处理技术的应用范围！借助 Streaming Ledger，我们正在开启流式处理的新篇章，我们很高兴现在越来越多的任务关键型应用程序可以充分利用流式处理的实时、异步、灵活等优势。

https://github.com/dataArtisans/da-streamingledger

英文原文：

https://data-artisans.com/blog/serializable-acid-transactions-on-streaming-data