干货分享 | MatrixOne系统架构

原创金海 MatrixOrigin服务号 2022-09-09 17:00

分享嘉宾：金海 矩阵起源研发VP

讲稿整理：张德通 DataFun志愿者

导读

随着数字化转型的深入，数据价值也愈发凸显，数据技术出现了融合的趋势，基于客户的真实痛点，矩阵起源推出了超融合MatrixOne HSTAP数据库。MatrixOne使数据数据库同时具备 TP、AP和Streaming三种能力，帮助客户彻底打破数据孤岛问题，成为企业智能化核心的数据基础设施。

今天的分享将解析新一代超融合异构云原生数据库MatrixOne刚刚发布的0.5版本和后续版本的架构设计。主要围绕以下四点展开，最后还有精彩Q&A环节实录奉上：

•

数据存储与处理面临的挑战

•

MatrixOne架构、各组件和设计方面的取舍

•

MatrixOne事务相关的读写流程

•

MatrixOne研发路线图

本文字数：7400字+

阅读时间：20分钟+

Part 1

数据存储与处理面临的挑战

企业数字化中台组件众多

企业在做数字化转型时，常常会构建技术中台，基于Hadoop搭建出一整套数据处理系统。下图展示了Hadoop体系内常见的数据组件，如图所示组件众多，有版本、兼容性等大量细节问题需要处理，学习成本、维护成本和维护难度高。

构建和维护整个数据中台就像在搭积木，各个模块搭建起来，虽然可以工作，但整个系统很脆弱，可靠性不好。任何一个组件有问题都会导致整个系统不可用，整个系统的可维护性差，构建成本高。

数据库

纵观数据库发展历程，70年代开始关系型数据库开始出现，80年代OLTP数据库经历了大规模发展，出现了Oracle、DB2、Sybase、SQL Server；90年代出现了AP型数据库，2000年随着互联网发展，出现了NoSQL数据库，支持更好的横向扩展性。NoSQL数据库对事务的支持能力不强，但其良好的可扩展性对数据存储非常友好，2000年左右出现了一批支持KV、文档、JSON的NoSQL数据库。

2000年开始逐渐出现了以Hadoop为代表的大数据系统，其中包括Hive、Spark等。同一时期，为解决传统数据库横向扩展能力受限问题出现了多种解决方案，一种方案是在传统MySQL数据库等基础之上做分库分表中间件，在中间件的基础上实现分布式事务能力；另一种方案则是以Google Spanner为代表的NewSQL，此外还包括OceanBase、TiDB、CockroachDB、YugaByte等，其特点是底层基于KV引擎，上层利用Raft、Paxos共识协议保证日志和状态机的可靠。

从2010年开始，云计算得到大规模发展，数据库本身也面临着上云挑战。第一代数据库上云的问题是如何在云上部署数据库，第二代、第三代云原生数据库则会利用云上的组件，如S3等对象存储降低成本提升可扩展性。云原生数据库中比较有代表性的TP型数据库是Aurora，AP型数据库是Snowflake，架构方面特征是存算分离，计算节点、存储节点可以独立扩缩容，节省用户成本，无需部署，运维简单，用户开箱即用。

大数据经过多年发展，从纯粹地用批处理解决问题变为可以用流处理做计算，对实时性要求高的场景支持得更好，与AP型数据库融合后大数据系统的分析能力也可以得到提升。

数据湖通常用于解决非结构化的数据，可以存储各类型数据，而数据仓库处理的是结构化的数据，两者的结合就是湖仓一体。湖仓一体比较有代表性的开源系统是Hudi、Iceberg，非开源的有Lakehouse。

由上面的发展可以看到，融合是大数据系统近年来的发展趋势：TP和AP的融合、湖和仓的融合等等。MarixOne正是这样的超融合数据库，类似集成了相机、电话、电子书、闹钟等功能的智能手机，具备着使用多种数据库才能实现的能力，能极大地降低数据库运维成本。

Part 2

MatrixOne架构总览

图中是MatrixOne的架构图，自上而下共分为5部分：

首先最上层的是计算层，计算层的计算节点我们称为CN。计算节点本身可以做分析、流计算和后台任务处理。用户层面和前台业务是感知不到计算节点的后台处理任务的。

计算层下面的事务层主要节点是DN，这些节点本身是share nothing、可横向扩容的。每个DN之间和内部的数据分片是互不相交的。目前的方案是根据数据主键做数据的横向分布扩展。每个DN之间处理的数据范围互相没有交集，不会有冲突检测的需要。

再下层是Log service和File service。DN节点写入的数据的日志会存在Log Service里。Log Service内部使用Multi-Raft，构建可靠的存储状态机。异步的compaction任务把写入的日志合并后写入File Service。File Service是一个通用的接口，底层可对接本地磁盘、NFS、S3、HDFS等多种数据存储系统。

HA Keeper组件不是大数据里常用的Zookeeper，而是一个集中式的集群管理组件。目前的实现方案是用单Raft组实现。HA Keeper维护的是计算集群、DN集群、Log Service集群的状态和可靠性。HA Keeper 发现集群内节点挂掉后会把节点拉起。

File Service

MatrixOne对自身的预期是可支持云原生和私有化部署，无论云上还是私有化全部使用一套方案进行部署。在私有化环境下，用户的实际存储环境多种多样，用户可能选择使用HDFS、Ceph等等做存储。云原生场景下也有多种选择，S3兼容的协议，阿里云OSS等。

在此背景下我们对存储进行了通用抽象，这一服务即File Service，作为存储接口提供给DN节点和CN节点进行数据交互。同时，File Service本身也承担了数据缓存的工作。

Log Service

Log Service是一个Multi-Raft组。由于存储的是Log tail，数据量不大。但我们对Log Service的吞吐量要求非常高。因此，在部署时候需要给Log Service配置更好的硬件设备，如硬盘需要用SSD盘。

Transaction DN节点

每个DN节点是shared-nothing的，每个DN节点之间互相访问的数据是不重叠的。目前版本DN组件的数据划分使用对主键做哈希的方式实现，不使用range的方式实现，因为我们认为哈希方式会让分布更加均匀。

DN节点的横向扩容难度比CN节点的横向扩容大，传统的NewSQL扩容节点可以利用Raft组的特性、加减副本即可完成扩容；但DN节点近期几个Release版本暂时没有计划DN的横向扩容，后续版本会增加扩容方面的优化。

计算节点CN

MatrixOne是存算分离的架构，计算节点CN节点的横向扩展能力很强，可以任意地扩缩容。

CN节点除了前台查询任务，还处理流任务和后台任务。后台任务包括异步的Compaction任务，本身会对数据进行修改，与前台任务数据会发生冲突，这些冲突需要被检测和处理。

后台任务方案由DN节点发起。DN节点维护状态机，状态机发现节点后发起异步任务给CN节点调起后台任务。这个过程不需要很高可靠性，异步后台任务并不需要很高的实时响应能力和幂等性。

HA Keeper

HA Keeper维护了集群各节点的状态，和集群内各节点保持心跳，节点挂到后把节点拉起；此外HA Keeper还和外部K8S资源池交互，把新增节点的上下文建立起来。HA Keeper是一个单Raft组的可靠集群，本身并发性不高，只能承载不高的心跳频率。

Part 3

MatrixOne事务相关的读写流程

事务流程

Matrixone使用的两阶段提交（2PC）实现事务和目前NewSQL方案略有区别。具体写入流程如下：

客户端使用write接口写入数据，请求到达CN节点后在CN节点上保存事务的读写空间workspace。数据只有到达DN节点才开始冲突检测。

一个事务从begin到commit中间发生的数据交互都会存储在CN节点的workspace上，一旦客户端发起commit，CN节点会进行两阶段提交把数据推到DN节点。Workspace数据可能分布在多个DN节点上，因此我们设计了DN节点内部两个阶段的处理流程：第一阶段是prepare，第二阶段commit。

为了保证两个阶段本身的可靠性，我们保证多个DN节点里选取一个作为coordinator，在一些系统中这个角色也被称作transaction record。第一个DN节点我们会做transaction commit，并把所有transaction参与者都记录在transaction coordinator里。

发生事务时，首先要把prepare log写入Log service进行持久化，包括transaction的commit信息、DN上的prepare信息等都做持久化存储。prepare的事务信息会返回给CN，CN节点收到事务参与者的response即整个事务成功，随后返回用户提交已经成功；或在rollback后再返回用户。

两阶段事务是一个异步的过程，DN节点commit过程是与prepare后返回给用户提交已经成功是异步进行的。两阶段事务具一些特殊性：首先workspace存储在CN节点上，冲突检测在DN节点上进行；第二个特点是分布式事务使用Clock SI方式分配timestamp。

时钟

Clock SI本身定义如上图红框圈中内容。任何一个事务会开启一个一致性快照，快照的开始时间是由一个快照时间戳确定的，在这个时间戳以前，所有已经commit的事务在这个快照里都可见。提交的时间戳是全序的。如果发生并发写写冲突，事务会被取消。

Clock SI主要解决了没有中心节点情况下的时间戳分发这一问题，也就是使用每个节点自己的时间戳。但节点和节点之间存在始终时钟漂移的问题，时钟漂移会面临两个错误错误：

第一个错误是快照不可用问题，左侧Fig.1展示了处在不同节点上的P1、P2两个事务参与者，他们之间存在时钟漂移，这会导致P2到达t之前快照不可用的问题。

第二个错误是Fig.2展示的，T2如果在T1提交时候读，拿到的数据需要等到T1提交完成后才能读到T1提交后的数据，不等待则读不到数据快照。

对于第二个错误，Clock SI使用上图中的算法1解决这两个问题。当T.SnapshotTime超过了当前提交时间戳才能读到数据，否则进行等待；也就是当前如果有事务在做prepare和commit，需要等待prepare和commit都完成后才可以进行冲突检测、做提交。

MatrixOne把Clocks SI和HLC结合，HLC是混合逻辑时钟，两个事务参与者发生时钟漂移时使用混合逻辑时钟校准，从而解决第一个错误。

读

对数据一致性要求较高的读请求，读请求到CN节点后需要从DN拉取数据的最新分布，DN把最新的元数据和meta都返回给CN节点。CN节点根据这些信息从File Service接口拉取到真正的数据。

MatrixOne使用单一的列存存储数据，每个column的block和segment的树形结构和bloom filter、minmax index等信息都保存在meta、index zone里，所有数据写入都是append only的，不论更新、插入还是删除都只是增加新文件。查询时用Merge on read的方式做合并。

DN内保存最新的meta，使用树形结构保存在DN内。查询的时候，CN会问DN要当前需要的快照和Log tail。CN根据快照里记录的对SQL进行裁剪。比如做SQL查询时，可用bloom filter做runtime filter。使得，真正需要读取进行计算的数据量比较小。

此外，DN节点会返回Log tail数据给CN节点，这部分日志数据相对较小，因此问题不大。

对于TP类查询，CN在拿到最新的数据后即可做一致性读。而AP类查询对数据一致性要求更低，可以使用CN内存储的meta读取，对DN负载压力较小，可以承载很高吞吐量。

异步 compaction

我们既可以用CN扫描数据决定是否做compaction，也可以由DN节点判断是否进行compaction。下图展示了使用DN做compaction的操作流程。

当DN节点发现数据删除过多，数据零散，这时会进行数据合并。数据合并时触发一个专用的用来做compaction节点跑合并，CN节点会把要发生compaction的数据进行收集，在CN节点内合并后提交。

Compaction会对数据产生产生修改，这个过程本身也是一个事务，会向DN提交数据并进行数据检测。

Compaction过程中修改的数据，前台任务也有可能修改了这部分数据，此时可能发生写写冲突，发生写写冲突时会abort后台compaction。此时要重跑compaction，独立的CN节点跑compaction任务也不会对用户体验造成影响。

Streaming方案实现计划

目前由两种方案做streaming，一种是数据发生修改，根据上一次产生的snapshot和当前的snapshot产生的delta snapshot推到CN节点上，CN节点本身根据streaming任务生成DAG图，在DAG图上做增量计算。增量计算也会取得上一次查询结果，以此为依据做增量计算。

最终查询结果是上图中的delta查询结果和基础查询结果的组合。存储中间结果的部分用推的模式。

另一个方式是定期地、在用户做streaming查询时从CN节点拉取base result、拉取delta snapshot，再做增量查询。查询结束后把最新的base result存储到S3、HDFS等可靠存储上，下次做增量计算时可以用到存储的base result。

Part 4

MatrixOne研发路线图

MatrixOne 0.5版本实现了列存带事务的分析引擎，第三季度即将发布的0.6版MatrixOne将支持分布式事务，TPC-H和TPC-C性能都有大幅提升。

2022年底计划发布的0.7版本将支持streaming，也会支持更多高级SQL、window function等。0.7版本MatrixOne计划上线云服务。

Part 5

精彩Q&A实录

※ 点击问号查看答案

扩容是冷扩容还是热扩容？

我们期望实现热扩容。但会先支持冷扩容，冷扩容指的是在线的事务先abort，执行扩容，扩容完毕后再重新开启事务。整个过程速度很快，因为只扩容DN节点和对应的Log tail状态机。Log tail存储的数据量少，扩容快。我们计划实现热扩容，可能会以追日志的方式实现热扩容。

元数据没有存Zookeeper？

元数据存储在HA Keeper。表结构我们成为Catalog。每个表数据索引的位置、数据的多个版本被认为是元数据，也叫做meta。HA Keeper本身存储当前集群是否挂掉，和每个DN分区负责的哈希桶。Catalog表信息存储在可靠存储File Service上，最新的可靠元数据信息是存储在Log Service上。

CN层和存储层是Share Disk架构吗，这样是否在扫描数据的时候影响性能？

File Service存储数据，访问数据也是CN节点通过File Service拉取数据，DN节点处理数据只做事务的冲突检测。DN节点冲突检测时访问数据有限，会先过滤，不会全量访问数据。且冲突检测通常只存在于一两个分片上。CN在进行Scan时也会对谓词做下推，Join时用Key决定访问数据范围，没有where条件的select * from table的请求对数据库性能影响很大。

CN节点对Scan类查询做谓词下推，最后CN节点真正要访问的数据范围有限，使用zone map过滤后确定部分数据。Join时维度表相对事实表更小，对性能影响不大；事实表通过维度表构建好Hash Table，根据hash key建立block、再下到事实表，用bloom filter过滤。最终访问的数据块也很小，减小数据访问的范围。

元数据、zone map、bloom filter这类索引信息都会在CN节点上有缓存。

DN内部的副本复制也用raft吗？

DN 在现在的方案内没有副本，意味重启一个 DN 时，需要从 log service 回放日志来恢复 DN 的状态机，此时当前 DN 不可用。我们可以把DN做成主备和三副本的形式，从log service拉取log tail，现在使用的DN没有stand by节点。

Clock SI里时钟方案相比HLC和TSO的优势是？

TSO有中心节点，HLC每次访问时虽然需要同步时钟，但同步可以发生在节点和节点之间。ClockSI使用本地时钟，延迟低，但节点之间可能存在时钟漂移问题。

读取会不会随着del和update增多性能下降，会有异步的合并吗？

MatrixOne是后台任务异步合并，而大量delete和update带来的垃圾数据多问题，会通过后台合并任务merge掉这部分数据而得到解决。

纯列存是指内存和磁盘都是列存吗？查询时会区分TP类查询还是AP类查询吗？

纯列存是磁盘和内存都是列存，查询时候会通过optimizer区分是AP还是TP。

资源隔离有什么考虑?

首先是TP和AP之间的资源隔离，TP和AP的CN节点分开。根据AP查询的一致性要求高低决定从CN、DN还是File service服务拉取元数据信息。

其次是租户之间的资源隔离。目前的方案是VIP用户部署单独的一套K8s deployment，普通用户根据schema进行资源隔离。后续可能会实现cgroup类似的方式限制租户隔离。

DN有开源的原型吗?

MatrixOne实现比较偏向NewSQL，DN也是share nothing实现的。我们参考过duckdb，但duckdb的update 和 delete是用version chain方式实现的，0.6版本后我们不再使用这一方案。

有考虑过不基于类似lsm的思想做存储层，而是采用copy on write或者类似kudu的delta store又或者adb采用的delta and insert 方案吗？

MatrixOne现在的方式也是LSM Tree的模式。LSM可以认为是分层的架构，以RokcsDB为例，它一共7层，最底层一定是全序的。MatrixOne为了避免写放大，没有全序的层，每层之间都有overlap。

Copy on write的实现方式存在一些问题，本质上对原有的数据块进行了修改、用新数据块替代了原有数据块，对元数据的修改比较复杂。LSM Tree是append only的，对元数据修改也是append only的，对于过去数据的compaction也是append only解决的。

MatixOne就是delta insert的实现，元数据管理复杂度大大降低。

流处理和分布式事务处理的结合，是否考虑过采用类似iceberg的元数据管理的思想？

Iceberg的元数据管理和我们的管理方式没有太多区别，但Iceberg把元数据都存储在S3上，而MatrixOne有DN加log service做元数据缓存，性能更好。

我们认为流是一个常见的后台常驻任务，任务随时根据数据源变化后推送或拉取新的数据执行分布式计算任务。分布式事务需要把流事务处理完才能返回成功，整个链条很长，返回更慢，这里就是一个取舍：流处理保证一致性还是让整个流程更快结束。我这里认为流处理和一致性不需要做强一致性，事务写完成后离开用流处理引起读，此时能拿到最新的数据，处理结束后整个流程结束，把事务分为两步进行。

AP相比主流的OLAP引擎，如Clickhouse，性能如何？

我们期望MatrixOne能和Clickhouse性能相媲美，但需要看具体场景具体分析。Clickhouse是存储计算不分离的，数据压缩、merge tree全序排列等特性会让Clickhouse更快；但MatrixOne的存算分离，非全序排列等特性也是Clickhouse无法比较的。Clickhouse的delete update是不能立刻可见的，但Matrixone是立刻可见的，这就是全序排列和非全序排列的取舍。存储全序排列主要影响主键，非主键的零散数据对数据压缩影响不大，因此AP计算引擎比较起来要具体场景具体分析，我们会做持续优化。

性能怎样同时兼容AP的大数据量和TP的高并发实时更新的？

AP和TP是用不同计算引擎做的。AP是的大数据量通常是查询的数据量大、不是写入的数据量大。为例支持AP的大查询量，数据写道S3上后最终提交时只由CN节点提交一次数据到DN上，DN上加载数据做冲突检测。CN上加载数据时已经做过一部分冲突检测，在对从S3上加载来的数据和logtail之间做检测。这个实现可以支持大批量加载。TP高并发和AP大读数据量可以兼容。

如果不需要一致性读AP可以CN直接读S3，不会影响TP性能；而需要一致性读的情况下，需要从DN读数据，DN上是高并发的实时更新的数据，此时DN会成为瓶颈，需要扩容DN节点。DN上拿到的是logtail和最新的元数据，CN上已有的快照元数据和DN上最新的快照元数据的不同也是很小，可以按批取这些数据，能极大地利用网络传输吞吐。

AP和TP底层是否有不同的存储结构？

是一样的结构，都是存储成列存格式。

AP、TP怎么调度？优化器针对AP，TP，streaming融合做了什么修改？

优化器绑定时可以拿到元数据和索引，通过索引构建直方图，能够判定数据分布，判断数据量的大小。由此判别是TP还是AP查询，随后把请求转发到AP或TP专用的CN节点上进行处理。

针对Streaming我们目前还在规划中。

在数据新鲜度上，HSTAP采取了哪些措施？

每次查询都到DN节点上读的方式一定是数据最新鲜的方式，TP数据只要写下去一致性读一定能读到最新数据，AP和TP使用一套列存，不存在数据同步等等延迟问题。

流处理如何保证最终一致性？

一旦发生了更新，会把更新的delta snapshot推到流数据计算服务上，经过计算产生流处理结果；或是在查询段流数据查询直接查，查到的数据是具备一致性的。

写流程中，第3步会客户端响应后，第4步写commit log，这是否有问题？

只要prepare全部完成，事务的提交已经成功，commit只是修改commit的时间戳。每一个事务参与者拿到事务时间戳有前有后，最后commit的时间戳需要是prepare的时间戳中最大的。这时候异步写不会有问题。

Transaction coordinator记录了事务的参与者，一旦事务中途挂掉，重新启动后会读取transaction coordinator内的log service的数据，这部分数据是可靠的。通过这个数据即可知道事务参与者，并检这些参与者的事务是否提交，没提交的话就把整个事务abort，提交了就不会有任何问题。

DN上的元数据如何做到？有何优化处理？

Table到column下每个segment其实是一个元数据数，可以理解为元数据存储本身也是一个元数据数据库。目前我们用LSM tree方式实现，内存里有状态机，写的具体内容是每一条元数据的状态机修改的日志。写完后内存的状态机会定期地刷到S3等可靠存储上。

这种实现可以比较方便地做数据的delta，CN来的查询单位一定是有事务构成的snapshot，snapshot是有每个事务的commit构成，每个commit也对应着一个数据元的修改。查询实际上是以snapshot或者commit为单位的，元数据修改所产生的delta可以拿到CN上做回放，CN就能拿到最新元数据。