概述
最近Flink母公司Data Artisans发布了一篇博客关于一个新的组件Streaming Ledger，给出了流数据的事务解决方案（就是常说的数据库的事务，满足ACID，隔离级别为Serializable）。
使用姿势
- 举例使用经典的转账和存款问题
- 它是基于Flink的，关于Flink任务初始化的一些内容就不放在这里了
- 首先创建StreamingLedger。
// start building the transactional streams
StreamingLedger tradeLedger = StreamingLedger.create("simple trade example");
- 第二定义所需要用到的状态和相应的KV类型，这里分别是账户和账单明细。
// define transactors on states
tradeLedger.usingStream(deposits, "deposits")
.apply(new DepositHandler())
.on(accounts, DepositEvent::getAccountId, "account", READ_WRITE)
.on(books, DepositEvent::getBookEntryId, "asset", READ_WRITE);
- 第三步是分别将输入流，具体的事务操作，操作的状态、从事件中获取key的方法、别名（会在事务操作，即此处的TxnHandler中体现）、和权限分别注入StreamLedger并输出为一个sideOutput。
// produce transactions stream
DataStream<TransactionEvent> transfers = env.addSource(new TransactionsGenerator(1));
OutputTag<TransactionResult> transactionResults = tradeLedger.usingStream(transfers, "transactions")
.apply(new TxnHandler())
.on(accounts, TransactionEvent::getSourceAccountId, "source-account", READ_WRITE)
.on(accounts, TransactionEvent::getTargetAccountId, "target-account", READ_WRITE)
.on(books, TransactionEvent::getSourceBookEntryId, "source-asset", READ_WRITE)
.on(books, TransactionEvent::getTargetBookEntryId, "target-asset", READ_WRITE)
.output();
- 第四步是根据sideOuput的OutputTag输出结果，到这里，除了TxnHandler需要去实现以外，主干逻辑已经完成了。
// compute the resulting streams.
ResultStreams resultsStreams = tradeLedger.resultStreams();
// output to the console
resultsStreams.getResultStream(transactionResults).print();
- 最后就是实现TxnHandler, 具体的转账和写入明细的逻辑都在这里。值得注意的是状态的获取依赖于上一步中在StreamLedger注入的别名，更新完状态之后再输出。
private static final class TxnHandler extends TransactionProcessFunction<TransactionEvent, TransactionResult> {
private static final long serialVersionUID = 1;
@ProcessTransaction
public void process(
final TransactionEvent txn,
final Context<TransactionResult> ctx,
final @State("source-account") StateAccess<Long> sourceAccount,
final @State("target-account") StateAccess<Long> targetAccount,
final @State("source-asset") StateAccess<Long> sourceAsset,
final @State("target-asset") StateAccess<Long> targetAsset) {
final long sourceAccountBalance = sourceAccount.readOr(ZERO);
final long sourceAssetValue = sourceAsset.readOr(ZERO);
final long targetAccountBalance = targetAccount.readOr(ZERO);
final long targetAssetValue = targetAsset.readOr(ZERO);
// check the preconditions
if (sourceAccountBalance > txn.getMinAccountBalance()
&& sourceAccountBalance > txn.getAccountTransfer()
&& sourceAssetValue > txn.getBookEntryTransfer()) {
// compute the new balances
final long newSourceBalance = sourceAccountBalance - txn.getAccountTransfer();
final long newTargetBalance = targetAccountBalance + txn.getAccountTransfer();
final long newSourceAssets = sourceAssetValue - txn.getBookEntryTransfer();
final long newTargetAssets = targetAssetValue + txn.getBookEntryTransfer();
// write back the updated values
sourceAccount.write(newSourceBalance);
targetAccount.write(newTargetBalance);
sourceAsset.write(newSourceAssets);
targetAsset.write(newTargetAssets);
// emit result event with updated balances and flag to mark transaction as processed
ctx.emit(new TransactionResult(txn, true, newSourceBalance, newTargetBalance));
}
else {
// emit result with unchanged balances and a flag to mark transaction as rejected
ctx.emit(new TransactionResult(txn, false, sourceAccountBalance, targetAccountBalance));
}
}
}
原理
- 其实我的第一想法是，卧槽好牛逼，这得涉及到分布式事务。把repo clone下来之后发现包含例子只有2000多行代码，一下子震惊了。但是实际的实现还是比较简单地，当然也肯定会带来一些问题。
- 实际上上面这些API会转换为一个source，一个sink，两个map和一个包含了SerialTransactor（ProcessFunction的实现）的算子。
- 在这边展示几行代码应该就能明白是如何做到的。关键在于forceNonParallel，这就让所有事情都变得明了了，事实上就是把状态全部都托管到一个并行度为1的算子上，处理的时候也是串行的，这里我才反应过来关键在于隔离级别是Serializable。这里带来的问题就是所有状态都保存在一个节点，并且不能支持水平扩展，所能支撑的吞吐量也不能通过加机器来提升。
SingleOutputStreamOperator<Void> resultStream = input
.process(new SerialTransactor(specs(streamLedgerSpecs), sideOutputTags))
.name(serialTransactorName)
.uid(serialTransactorName + "___SERIAL_TX")
.forceNonParallel()
.returns(Void.class);
感想
其实看到这个功能的第一感觉是很牛逼，但是仔细看过了它的实现觉得真正应用上可能会有不少问题。因为对于最重要的处理事务的那个算子来说，本质上它并不是Scalable的，没有办法横向扩展。不过从功能上来说，确实引出了一个新的发展方向，希望以后还能看到有更优的解决方案，比如针对另外两种隔离级别Read Committed和Repeatable read。
