TiFlink：使用TiKV和Flink实现强一致的物化视图

发布时间：2025-11-05 13:38:03 来源：码上建站 作者：IT科技类资讯

在本年初的使k实视图TiDB Hackathon上，我和一众队友尝试使用Flink为TiDB添加物化视图功能，现强并摘得了“最佳人气奖”。物化可以说，使k实视图物化视图在这届比赛中可谓是现强一个热点。单单是物化结合Flink实现相关功能的队伍就有三四个。必须承认的使k实视图是，在比赛结束时我们项目的现强完成度很低，虽然基本思路已经定型，物化最终呈现的使k实视图结果却远没达到预期。经过半年多断断续续的现强修补，在今天终于可以发布一个 预览版本 给大家试用。物化这篇文章就是使k实视图对我们思路和成果的一个介绍。

相比其他队伍，现强我们的物化主要目标是实现强一致的物化视图构建。也就是保证查询时的物化视图可以达到接近快照隔离（Snapshot Isolation）的隔离级别，而不是亿华云一般流处理系统的最终一致性（Eventual Consistency）。关于实现一致性的讨论在下文有详细介绍。

使用简介

尽管是一个实验性的项目，我们仍然探索了一些方便实用的特性，包括：

关于TiFlink实用的详细信息，请参考 README 。下面是快速启动一个任务的代码片段：

物化视图（流处理系统）的一致性

目前主流的物化视图（流处理）系统主要使用最终一致性。也就是说尽管最终结果会收敛到一致的状态，但在处理期间终端用户仍可能查询到一些不一致的结果。最终一致性在很多应用中被证明是足够的，那么更强的一致性是否真的需要呢？这里的一致性和Flink的Exact Once语义又有什么关系呢？有必要进行一些介绍。

ACID

ACID是数据库的站群服务器一个基本的概念。一般来说，作为CDC日志来源的数据库已经保证了这四条要求。但是在使用CDC数据进行流式处理的时候，其中的某些约束却有可能被破坏。

最典型的情况是失去Atomic特性。这是因为在CDC 日志中，一个事务的修改可能覆盖多条记录，流处理系统如果以行为单位进行处理，就有可能破坏原子性。也就是说，在结果集上进行查询的用户看到的事务是不完整的。

一个典型的案例如下：

在上述案例中，我们有一个账户表，账户表之间会有转账操作，由于转账操作涉及多行修改，因此往往会产生多条记录。b2b供应网假设我们有如下一条SQL定义的物化视图，计算所有账户余额的总和：

显然，如果我们只存在表内账户之间的转账，这个查询返回的结果应该恒为某一常数。但是由于目前一般的流处理系统不能处理事务的原子性，这条查询产生的结果却可能是不断波动的。实际上，在一个不断并发修改的源表上，其波动甚至可能是无界的。

尽管在最终一致的模型下，上述查询的结果在经过一段时间之后将会收敛到正确值，但没有原子性保证的物化视图仍然限制的应用场景：假设我想实现一个当上述查询结果偏差过大时进行报警的工具，我就有可能会接收到很多虚假报警。也就是说此时在数据库端并没有任何异常，数值的偏差只是来源于流处理系统内部。

在分布式系统中，还有另一种破坏原子性的情况，就是当一个事务修改产生的副作用分布在多个不同的节点处。如果在这时不使用2PC等方法进行分布式提交，则也会破坏原子性：部分节点（分区）上的修改先于其他节点生效，从而出现不一致。

线性一致性

不同于由单机数据库产生的CDC日志（如MySQL的Binlog），TiDB这类分布式数据库产生的日志会有线性一致性的问题。在我们的场景下，线性一致性的问题可以描述为：从用户的角度先后执行的一些操作，其产生的副作用（日志）由于消息系统传递的延迟，以不同的先后顺序被流处理系统处理。

假设我们有订单表（ORDERS）和付款信息表（PAYMENTS）两个表，用户必须先创建订单才能进行支付，因此下列查询的结果必然是正数：

但是由于ORDERS表和PAYMENTS表在分别存储在不同的节点上，因此流处理系统消费他们的速度可能是不一致的。也就是说，流处理系统可能已经看到了支付信息的记录，但是其对应的订单信息还没到达。因此就可能观察到上述查询出现负数的结果。

在流处理系统中，有一个Watermark的概念可以用来同步不同表的数据的处理进度，但是它并不能避免上述线性一致性问题。这是因为Watermark只要求时间戳小于其的所有记录都已经到达，不要求时间戳大于其的记录都没有到达。也就是说，尽管ORDERS表和PAYMENTS表现在拥有相同的Watermark，后者仍然可能会有一些先到的记录已经生效。

由此可见，单纯依靠Watermark本身是无法处理线性一致性问题的，必须和源数据库的时间产生系统和消息系统配合。

更强一致性的需求

尽管最终一致性在很多场景下是够用的，但其依然存在很多问题：

实际上，更强一致性的缺乏还导致了一些运维操作，特别是DDL类的操作难以利用之前计算好的结果。参考关系型数据库和NoSQL数据库的发展历史，我们相信目前主流的最终一致性只是受限于技术发展的权宜之计，随着相关理论和技术研究的进步，更强的一致性将会慢慢成为流处理系统的主流。

技术方案简介

这里详细介绍一下TiFlink在技术方案上的考虑，以及如何实现了强一致的物化视图（StreamSQL）维护。

TiKV和Flink

尽管这是一个TiDB Hackthon项目，因此必然会选择TiDB/TiKV相关的组件，但是在我看来TiKV作为物化视图系统的中间存储方案具备很多突出的优势：

我们最初的想法本来是直接将计算功能集成进TiKV，选择Flink则是在比赛过程中进一步思考后得到的结论。选择Flink的主要优势有：

当然，选择TiKV+Flink这样的异构架构也会引入一些问题，比如SQL语法的不匹配，UDF无法共享等问题。在TiFlink中，我们以Flink的SQL系统和UDF为准，将其作为TiKV的一个外挂系统使用，但同时提供了方便的建表功能。

强一致的物化视图的实现思路

这一部分将介绍TiFlink如何在TiDB/TiKV的基础上实现一个比较强的一致性级别：延迟快照隔离（Stale Snapshot Isolation）。在这种隔离级别下，查询者总是查询到历史上一个一致的快照状态。在传统的快照隔离中，要求查询者在$T$时间能且只能观察到Commit时间小于$T$的所有事务。而延迟快照隔离只能保证观察到$T-\Delta t$之前所有已提交的事务。

在TiDB这样支持事务的分布式数据库上实现强一致的物化视图，最简单的思路就是使用一个接一个的事务来更新视图。事务在开始时读取到的是一个一致的快照，而使用分布式事务对物化视图进行更新，本身也是一个强一致的操作，且具有ACID的特性，因此得以保证一致性。

为了将Flink和这样的机制结合起来且实现增量维护，我们利用了TiKV本身已经提供的一些特性：

因此TiFlink的基本实现思路就是：

以上几点的关键在于协调各个节点一起完成分布式事务，因此有必要介绍一下TiKV的分布式事务执行原理。

TiKV的分布式事务

TiKV的分布式事务基于著名的Percolator模型。Percolator模型本身要求存储层的KV Store有MVCC的支持和单行读写的原子性和乐观锁（OCC）。在此基础上它采用以下步骤完成一次事务：

上述分布式事务之所以可行，是因为对主键的Commit是原子的，分布在不同节点的副键是否提交成功完全依赖于主键，因此其他的读取者在读到Prewrite后但还没Commit的行时，会去检查主键是否已Commit。读取者也会根据Commit时间戳判断某一行数据是否可见。Cleanup操作如果中途故障，在之后的读取者也可以代行。

为了实现快照隔离，Percolator要求写入者在写入时检查并发的Prewrite记录，保证他们的时间戳符合一定的要求才能提交事务。本质上是要求写入集重叠的事务不能同时提交。在我们的场景中假设物化视图只有一个写入者且事务是连续的，因此无需担心这点。

在了解了TiKV的分布式事务原理之后，要考虑的就是如何将其与Flink结合起来。在TiFlink里，我们利用Checkpoint的机制来实现全局一致的事务提交。

使用Flink进行分布式事务提交

从上面的介绍可以看出，TiKV的分布式事务提交可以抽象为一次2PC。Flink本身有提供实现2PC的Sink，然而并不能直接用在我们的场景下。原因是Percolator模型在提交时需要有全局一致的事务开始时间戳和提交时间戳。而且仅仅是在Sink端实现2PC是不足以实现强一致隔离级别的：我们还需要在Source端配合，使得每个事务恰好读入所需的增量日志。

幸运的是，Flink的2PC提交机制实际上是由Checkpoint驱动的：当Sink接收到Checkpoint请求时，会完成必要的任务以进行提交。受此启发，我们可以实现一对Source和Sink，让他们使用Checkpoint的ID共享Transaction的信息，并配合Checkpoint的过程完成2PC。而为了使不同节点可以对事务的信息（时间戳，主键）等达成一致，需要引入一个全局协调器。事务和全局协调器的接口定义如下：

使用上述接口，各个Source和Sink节点可以使用CheckpointID开启事务或获得事务ID，协调器会负责分配主键并维护事务的状态。为了方便起见，事务Commit时对主键的提交操作也放在协调器中执行。协调器的实现有很多方法，目前TiFlink使用最简单的实现：在JobManager所在进程中启动一个GRPC服务。基于TiKV的PD（ETCD）或TiKV本身实现分布式的协调器也是可能的。

上图展示了在Flink中执行分布式事务和Checkpoint之间的协调关系。一次事务的具体过程如下：

注意到，在第一个Checkpoint开始前，Sink可能已经开始接收到写入的数据了，而此时它还没有事务的信息。为了解决这一问题，TiFlink在任务开始时会直接启动一个初始事务，其对应的CheckpointID是0，用于提交最初的一些写入。这样的话，在 CheckpointID=1 的Checkpoint完成时，实际上提交的是这个0事务。事务和Checkpoint以这样的一种错位的方式协调执行。

下图展示了包含协调器在内的整个TiFlink任务的架构：

基于以上的系统设计，我们就得到了一个在TiKV上实现延迟快照隔离的物化视图。

其他设计考虑

众所周知，KSQL是Flink之外另一个流行的流处理系统，它直接与Kafka消息队列系统结合，用户无需部署两套处理系统，因此受到一些用户的青睐。很多用户也使用KSQL实现类似物化视图这样的需求。然而在我看来，这种强耦合于消息队列的流处理系统并不适合物化视图的使用场景。

KSQL可以说是Log Oriented数据处理系统的的代表，在这种系统中，数据的本源在于日志信息，所有的表都是为了方便查询而消费日志信息从而构建出来的视图。这种系统具有模型简单、容易实现、可以长时间保存日志记录等优点。

与之相对是Table Oriented数据处理系统，MySQL、TiDB/TiKV都属于这一类系统。这一类系统的所有修改操作都作用于表数据结构，虽然期间也会有日志生成，但往往对表数据结构和日志的修改是一起协调进行的。这里日志的主要是为持久化和事务服务，往往不会留存太长时间。相比于Log Oriented数据处理系统，这类系统对写入和事务的处理都更为复杂一点，然而却拥有更强可扩展性的要求。

归根结底，这是因为Log Oriented系统中的数据是以日志的形式存储，因此在扩展时往往需要进行成本较高的Rehash，也更难实现再平衡。而Table Oriented的系统，数据主要以表的形式存储，因此可以以某些列进行有序排列，从而方便在一致性Hash的支持下实现Range的切分、合并和再平衡。

个人认为，在批流一体的物化视图场景下，长时间保存日志并无太大的意义（因为总是可以从源表的快照恢复数据）。相反，随着业务的发展不断扩展数据处理任务和视图是一件比较重要的事。从这个角度来看Table Oriented系统似乎更适合作为物化视图需求的存储承载介质。

当然，在实时消费增量Log时发生的分区合并或分裂是一个比较难处理的问题。TiKV在这种情况下会抛出一个GRPC错误。TiFlink目前使用的是比较简单的静态映射方法处理任务和分区之间的关系，在未来可以考虑更为合理的解决方案。

总结

本文介绍了使用Flink在TiKV上实现强一致的物化视图的基本原理。以上原理已经基本上在TiFlink系统中实现，欢迎各位读者试用。以上所有的讨论都基于Flink的最终一致模型的保证，即：流计算的结果只与消费的Event和他们在自己流中的顺序有关，与他们到达系统的顺序以及不同流之间的相对顺序无关。

目前的TiFlink系统还有很多值得提高的点，如：

如果各位读者对TiFlink感兴趣的话，欢迎试用并提出反馈意见，如果能够贡献代码帮助完善这个系统那就再好不过了。

关于物化视图系统一致性的思考是我今年最主要的收获之一。实际上，最初我们并没有重视这一方面，而是在不断地交流当中才认识到这是一个有价值且很有挑战性的问题。通过TiFlink的实现，可以说是基本上验证了上述方法实现延迟快照一致性的可行性。当然，由于个人的能力水平有限，如果存在什么纰漏，也欢迎各位提出讨论。

最后，如果我们假设上述延迟快照一致性的论述是正确的，那么实现真正的快照隔离的方法也就呼之欲出。不知道各位读者能否想到呢？

相关文章