巨人肩膀
改版通知

巨人肩膀网站已全新改版。若您仍依赖旧站功能或数据，欢迎联系我们，我们会协助处理。联系我们

← 返回专栏列表

StarRocks实时更新技术内部实现方案解析

ckckck2025年1月10日139 浏览

编者荐语

不错的文章

以下文章来源于 DataFunSummit，作者常冰琳

Image

DataFunSummit
DataFun社区旗下账号，专注于分享大数据、人工智能领域行业峰会信息和嘉宾演讲内容，定期提供资料合集下载。

Image

导读

本次分享是对 StarRocks 内部实时更新技术方面的介绍。首先介绍为什么 OLAP 系统里面越来越需要实时更新，后面会介绍当前市面上一些 OLAP 数据库是如何去做实时更新的。

分享大纲

  • 实时更新需求
  • StarRocks 实时更新实现
  • 当前和未来工作

01 实时更新需求

1. AP系统里面为什么需要实时更新

传统的OLAP系统：

  • 最常使用的就是 T+1 批量 ETL 方式导入数据，同时通过 Overwrite 方式实现更新数据的导入。这种方式延迟会比较高，不太适合实时更新的需求。
  • 增量不停的导入数据，数据只能追加（append only），不能更新（update），只能插入（insert）。这在实时日志分析、广告等场景都会用到。
  • 追加更新数据，底层使用 MOR（merge-on-read）方式来支持更新（类似 LSM 结构【日志结构合并树】）。这种方式能完美解决实时更新的需求，但是 merge-on-read 对查询的性能影响比较高，不太适合实时查询场景。

对实时分析的新需求：

  • 摄入的数据基本是实时数据，也就是热数据，通常也会是可变的数据（volatile data），经常会被更新到。
  • 很多 HTAP 的系统都支持实时数据的更新，把 TP 的数据实时的同步到 AP 系统里面（一般通过 CDC 的方式）。

另外，目前 ELT 的方式越来越流行，就是原始数据抽取（extract）出来直接导入（load）到数仓里面，然后在数仓内部进行一些高级的转换（transform），同时可以在数仓内部进行查询。大大简化了整个数据的 pipeline，并且能够很方便的、对用户非常友好转化数据，使之能够更高效的被查询。涉及到复杂的 DML 的更新，比如 update、delete 或者是 merge into 等批量修改数据的语句，在 AP 系统越来越流行。

2. 实时更新的场景

（1）场景: full row upsert / delete

对数据库里的一行（整行）进行 upsert 或者 delete：

Image
  • 在 mysql 下，插入一条数据，如果 key（unique key）重复就覆盖数据行。
Image
  • 在 StarRocks 下，有 Unique key 表支持 upsert 语义，另外 StartRocks 新加了一个 Primary key 的表模型，它支持 upsert 和 delete 语义。

有了这两个语义之后就能完美支持从 OLTP 系统到 OLAP 系统 CDC 数据的同步。

（2）场景实现方式

① Merge-on-Read

当获取 CDC 数据，排序后直接写入新的文件，不做重复键检查。读取时通过 Key 值比较进行 Merge，合并多个版本的数据，仅保留最新版本的数据返回。Merge-on-Read 方式 写入数据非常快，但是 读数据比较慢，对查询的影响比较大。它参考的是存储引擎比较典型的、应用广泛的 LSM 树的数据结构。

Image

Hudi 的 MOR 表以及 StarRocks 的 Unique key 表都是使用这种方式是实现的。

② Copy-on-Write

当获取 CDC 数据后，新的数据和原来的记录进行 full join，检查新的数据中每条记录跟原数据中的记录是否有冲突（检查 Key 值相同的记录）。对有冲突的文件，重新写一份新的、包含了更新后数据的。读取数据时直接读取最新数据文件即可，无需任何合并或者其他操作，查询性能最优，但是 写入的代价很大，因此适合 T+1 的不会频繁更新的场景，不适合实时更新场景。

Image

现在的数据湖方案里像 Delta Lake，Hudi 的 COW 表，Iceberg 以及商用的 Snowflake 都是使用这种方式达到数据更新目标。

③ DeltaStore

基本思想是牺牲写入性能，换取更高的写性能。当获取 CDC 数据后，通过主键索引，可以定位到这条记录原来所在的位置、文件（或者 Block），然后在这个 Block 旁边放一个 Delta Store，用来保存对于这个 Block 的增量修改。这个 Delta Store 里保存的不是主键而是数据记录在 Block 里的行号（RowId）。查询时读到原始 Block，然后根据 RowId 与 Delta 数据进行合并更新并返回最新数据给上层引擎。由于 Delta 数据是按行号组织的，与 Merge-on-Read 按照 Key 进行合并比，查询性能好很多。不过这种方式会破坏二级索引，因为对 Block 做修改后，他的索引相当于失效了，想要在更新后再维护索引复杂度会很高。这种方式 写入性能差（要查询索引，定位原数据），但 读取的性能好 很多。另外因为引入了主键索引和 Delta Store，复杂性也较高。

Image

采用这种方案较典型的是 Apache Kudu，许多 TP 系统甚至 HTAP 数据库也会到 Delta Store 存储方式。

④ Delete-and-Insert

思路也是牺牲部分写性能，极大的优化读的性能。原理也是引入主键索引，通过主键索引定位原来这条记录所在位置，找到记录后只需要给这条记录打个删除标记（Delete Bitmap），表示这条记录是被删除的，然后所有其它 update 记录可以当成新增数据插入到新的 Block 里面。这样的好处是读取时直接把所有的 Block 都可以并行的加载，然后只需要根据 Delete Bitmap 标记过滤已经删除的记录。

Image

StarRocks 新的支持实时更新的 Primary Key 表就是用到这个 Delete+Insert 的方式实现的；另外还有阿里云的 ADB 和 Hologres 也用到在这种方式。

02 StarRocks的实时更新

1. 基本系统架构

StarRocks 集群由 FE 和很多 BE 组成：

  • FE：FrontEnd，StarRocks 的前端节点，负责集群管理元数据，管理客户端连接，进行查询解析规划，生成查询执行计划，查询调度（把查询下发给 BE 执行）等工作。
  • BE：BackEnd，StarRocks 的后端节点，负责数据存储，查询执行计划的执行，compaction，副本管理等工作。表数据存储这块，每个表的数据根据分区或分桶机制被打散分成多个 Tablet，为了容错 Tablet 会做多个副本，这些 Tablet 最终分布在不同的 BE 上。

如下图所示，一个导入事务可以分为两个阶段：Write, Commit.

Image
  • Write阶段：在第一阶段，将写入事务的数据按照分区分桶规则分散地写入到很多的 Tablet 里，每个 Tablet 收到数据后把数据写成内部的列存格式，形成一个 Rowset；
  • Commit阶段：所有数据都写入成功后，FE 向所有参与的 Tablet 发起事务、提交 Commit，每个 Commit 会携带一个 Version 版本号，代表 Tablet 数据的最新版本。Commit 成功后，整个事务就完成了。

2. Tablet内部结构

Tablet 是 StarRocks 底层存储引擎的基本单元，每个 Tablet 内部可以看作一个单机列存引擎，负责对数据的读写操作。Tablet 内部分为以下四个部分：

Image
  • Rowset：一个 Tablet 的所有数据被切分成多个 Rowset，每个 Rowset 以列存文件的形式存储。
  • Meta（元数据）：保存 Tablet 的版本历史以及每个版本的信息，比如包含哪些 Rowset 等。序列化后存储在 RocksDB 中，为了快速访问会缓存在内存中。
  • DelVector：记录每个 Rowset 中被标记为删除的行，同样保存在 RocksDB 中，也会缓存在内存中以便能够快速访问。
  • PrimaryIndex（主键索引）：保存主键与该记录所在位置的映射。目前主要维护在内存中，正在实现把主键索引持久化到磁盘的功能，以节约内存。

3. 元数据组织

和 Delta lake 或者其它数据湖保存元数据的方式类似，也是一个 MVCC 多版本方式。每次写事务或者每次 Compaction 都会生成新的元数据版本。版本里面会记录这个版本包含哪些 Rowset、相对上一个版本它的增量数据是什么（就是 Delta）。

Image

举个例子：

  1. 一开始创建一个 tablet，版本是 1.0，Rowset 是空的，Delta 也是空的。
  2. 新来一个数据，写入数据后版本号变成 2.0，Rowsets 包含 Rowset 0，Delta 相比上一个版本增加了 Rowset 0。
  3. 又做一个新的 commit，加进去一个 Rowset 1，版本号变成 3.0。
  4. 假设两个 Rowset 比较小，触发了一个 compaction。输入 Rowset 0 和 1，目标是合并这两个 Rowset。compaction 需要时间，但系统还会继续接收数据写入。
  5. 又 commit 一个新的版本 4.0，新加入 Rowset 2，这时候 Rowsets 里面包含 Rowset 0、1、2 三个 Rowset。
  6. 这时 compaction 完成，把 Rowset 0 和 1 合并为 Rowset 3，Rowsets 里就是 Rowset 2 和 3，版本变成 4.1。
  7. 最后再写入一次数据，版本变成 5.0，Rowsets 里有 Rowset 2、3、4，Delta 是 Rowset 4。
Image

从上可以看出这是一个 MVCC 多版本的元数据组织，同时支持并行的 Compaction 和写入操作。随着数据不断导入，元数据版本会逐渐积累，这个版本不能一直这样积累（每分钟可能产生好几个导入，半个小时可能上百个版本，一直下去元数据大小会很恐怖）。我们可以把一些老的版本，不太可能会被用到的版本元数据 GC 掉。其原理是：每个版本都有时间戳，当时间已经比较老了，就认为不太可能被查询到，就可以把它删除掉，删除后整个元数据大小就变小了，同时被 compaction 过清理掉的 Rowset 也可以删除，回收存储空间。

Image

4. 写入流程

目前数据库里的写事务基本上是导入的方式来完成。导入就是把一批数据通过 Broker Load 或者 StreamLoad 的方式整个写入到系统里面。从前面可以知道写入数据是由每个 Tablet 负责，每个 Tablet 接收到数据后，会执行一个写入流程。

具体流程如下：

  1. Tablet 收到批次数据后，先写到 MemTable 里面（在 MemBuffer 里保存），MemBuffer 满了之后再 Flush 到存储引擎上面。
  2. 在 Flush 之前会把这批导入中 Upsert 操作和 Delete 操作拆分开，同时还要按主键去重，最终效果是这批导入里的 Upsert 操作和 Delete 操作的 key 不再重复，相当于老版本被新的版本覆盖。Flush 包含 Sort、Merge、Split 三个操作，Sort 先按主键对数据排序，Merge 对主键相同的多条记录进行合并，只保留最新版本的数据。Split 把这批操作中的 Upsert 和 Delete 操作拆分开来，写入不同的文件。
  3. Flush 完成后会生成一个包含多个文件的新 Rowset。
  4. 这批数据发送完毕并 Flush 后，FE 发起 Commit 流程。BE 以一个单独的事务提交，同时查询 Primary Index 找到所有被更新的记录标记为删除，并生成 DelVector，与新版本 Meta 一起提交到 RocksDB 中，达到事务的原子性。
Image

简单样例说明：

  1. 假设刚开始有一个 Rowset 0 写入到系统，里面有 1,2,3,4 四条记录。
  2. 又一次写入，主键是 1,3,5，发现 1,3 已存在，它们就是 upsert 更新的操作。通过查询主键索引能够得到这两条记录原来在哪个位置，在原来的位置上生成 delVector。
  3. 假设又新来一个事务，要对 6 进行插入，同时删除主键为 3 的记录。同样查询主键索引获取位置并生成 delVector。
  4. Rowset 3 commit 后，也是查询主键索引，增量的生成 delVector 整个过程完成之后，也生成了 MVCC 的结构，查询引擎可以查询任何版本得到相应的数据。
Image

5. 写事务并发控制

StarRocks 支持并发写操作，系统中可能同时存在多个并发读事务和并发写事务执行。对同一个表的写事务并发：

  • 写入流程可以并发执行
  • 但是 Commit 需要序列化（顺序）执行
Image

序列化执行过程中，Commit 需要做的事情是要更新 Primary Index、同时生成 DelVector，然后把 DelVector 和 Meta 写入 RocksDB。

Image

6. 主键索引

Commit 阶段的最主要工作是查找和更新主键索引，约占整个 Commit 过程 90% 以上的时间。怎样让这个关键一步尽量的快？目前主键索引使用的是高性能内存 HashMap。

  • Key 是主键通过二进制编码后形成的 Binary String 串
  • Value 是一个 unit64 位的行位置信息，前面 32 位是 rowset_id，后面 32 位是 rowset 内部的 rowid
  • HashMap 使用的是高性能的开源 phmap库，每次操作耗时仅 20ns~200ns，单核可以达到 5M-50M op/s。如果一个导入事务有1千万行，要写入10个bucket，每个bucket 进行100万的hashmap的操作，这100万的操作在Primary index里面只需要0.1左右。这个commit过程是非常快的，也能保证整个系统多个写事务的并发操作。

主键索引优化：

  1. 由于支持的都是一些批量操作，可以通过预取的方式加速 Hashmap 操作，可以缓解 Cache miss 的开销。
  2. 主键索引非常占内存，针对定长和变长类型的主键分别做了一些优化，以降低内存占用：
    • 定长主键使用定长的字符串类型（如：FixSlice）保存到 hashmap 里面，就不需要保存字符串长度。
    • 变长主键设计一个 shard by length 的 hashmap，类似 clickhouse 里的 hashmap，可以节约 1/2 ~ 1/3 的内存。
    • cashe on-demand loading，如果一个表暂时没有更新操作（6 分钟内没有 load），就不加载到内存。
  3. 目前在做的工作：
    • 优化常量和基数低的主键列，减少内存占用。
    • 使用原始列 128bit hash 做为主键，风险是有可能有冲突，但是冲突概率非常低，对特定应用可用。

可持久化主键（PersistentPrimaryIndex）： 这项工作正在进行，大概思想还是用 hashmap 的方式实现。很多索引要么是用 LSM 的方式做，要么用 B tree 的方式做，都是有序索引，StarRocks 中没有有序的需求，所以使用 hashmap 的方式。

Image

组织方式是 LSM Tree 的思想，分多层：

  • L0 层映射在内存里，有一个 buffer，同时写 WAL（预写日志系统，一种高效的日志算法）或者 FlushSnapshot。
  • L1 层是 Shard HashMap，保存在磁盘上，每个 shard 大概 1M 左右大小。如果有批量的读操作，可以根据 shard 把批量读操作也跟 L1 shard 大小一样分成很多 bucket。

比如一批读操作有 100 条，分 8 个 shard，每个 shard 大约 10 几条的读操作，每次读操作批量的把 shard load 上来，操作完成后，shard 内存可以再释放掉，读取下一个 shard。

7. 场景：冷热数据

主键索引对内存要求比较高，比较适用两个场景，一个是数据按时间分区，数据按时间有冷热特征，比较老的数据基本上不会被更新，被更新的数据基本是最近几天的数据。因为我们的主键是 lazy 加载的方式，只有最近几天的表分区的加载到主键索引占用内存，整个表的内存就可空。

Image

很多场景都是这个场景，比如：

  • 在线电商订单表
  • 共享打车和共享单车的骑行订单数据
  • 移动端 App 和 IOT 设备的 sessions

8. 场景：大宽表

Column 非常多（大于 100 列），主键只占整个存储数据量的很小部分，可以允许整个主键索引放在内存里，对整个系统的消耗有限。

Image

像 User profile，用 user_id 做主键，占内存也不是特别大。

9. Compaction

对数据的更新和删除会使得 Rowset 中的大量记录被标记为删除，另外数据持续实时导入会导致存在大量小的 Rowset 文件，这些会影响查询的读取性能，并且会浪费存储空间。StarRocks 的 Compaction 作用是删除已经标记删除的行、把实时导入产生的大量小文件合并成大文件。LSM compaction 是 merge 一些 sst 文件生成一个新的 sst 文件，在元数据里面原子的对 sst 文件进行替换。StarRocks compaction 与 LSM 不同点在于：

  • 没有重复记录，不需要考虑相同 Key 的 merge。
  • 没有 delete / range delete 的记录，rocksdb 曾经出行过 range delete bug。
  • 由于引入了主键索引，Compaction 会使得记录的实际位置发生变化，这个变化需要同时反应到 Primary Index 里面，这个数据就增加了复杂度。

Compaction 可以理解为一个写入的事务，和外部写入事务类似：

  • 根据策略，挑选一些 Rowsets。
  • merge Rowsets，生成新的 Rowset。
  • 作为一个 Commit 把元数据写入到 rocksdb 里面，同时生成 DelVector，在元数据中原子替换掉输入的 Rowsets 也写入到 rocksdb 里面。
Image

10. Merge-on-Read vs Delete+Insert模式

Delete+Insert 模式为什么对性能提升那么明显，这里与 Merge-on-Read 对比来看，对订单做统计查询：

使用 merge-on-read 方式，首先需要多个 Rowset 需要做一次 merge 操作，得到 merge 表后才应用 filter，过滤后再对剩下的数据进行汇总统计得到结果。

使用 delete + insert 方式，不需要 merge 操作，filter 可以下推到存储引擎最下面一层，这就可以利用 bitmap 索引加速查询。而 merge-on-read 方式下索引就失效了，另外 merge-on-read 中需要把 order_id 和 state 列读取出来。在 delete + insert 这边 order_id 和 state 都不需要读，这极大的减少了 IO 量。

总结来看优势有以下几点：

  • 不需要 merge 操作
  • 不需要读 order_id，省 IO
  • 过滤条件下推，利用 bitmap 索引加速查询
  • 只扫描返回 revenue 列
Image

11. Benchmark

（1）Simple Query on Single Table

这是内部一个测试，在实时导入的情况下，模拟订单表，数据量大概 2 亿行左右，5 分钟一个 batch。实际是 2~3 秒导入一次，很短的时间内把 2 亿条数据导入到系统里。一边导入一边查询，和原来 Merge-on-Read 方式对比，在简单查询的情况下有 10 到几十倍的性能提升。

Image

（2）TPC-H1T

利用标准的 TPC-H 1T 数据和市面上商用的 MPP 支持更新的数据库进行对比，基本上有 2 倍甚至 2 倍以上的性能提升。

Image

03 当前和未来工作

1. 支持PartialUpdate（部分列更新）

StarRocks 现有支持的操作类型包括 Upsert 和 Delete，就是整行的更新和删除。Partial update 就是只更新其中的某一列或者某几列而不是所有的列。这个在 Merge-on-Read 里面很容易实现，但是在 Delete + Insert 的方式下有写挑战。最大的挑战是，比如一个表有 100 列，只更新其中一列，在做 Insert 过程中要把原来版本的数据先读上来拼成一整行之后才能写入。但是在列存情况下，99 列可能需要做 99 次随机 IO 才能一行拼出来，对整个存储系统的 IO ps 数压力特别大。

Image

目前只实现了功能，如果是一个 NVMe 或者 SSD 的磁盘基本能够杠住一定的流量，再高的流量还不太适合这个场景，后面也会持续优化。对一些隔离级别比较高的场景，在最后写入 commit 时需要检查冲突，如果写入事务发生冲突，还需要解决冲突操作，这里不做详述。

适用场景：

  • 固定列的更新，比如订单表，第一次写入时大部分列已经固化，只会一些固定列比如订单状态。
  • 某列的批量更新，比如 user_profile 表，上游有些像机器学习的任务批量更新用户标签。
Image
  • 大宽表，维护一个大宽表查询时不在需要 Join，可以提高查询性能。对大宽表一般都是由两个或两个以上的 Job 来导入，每个 Job 只更新部分列。
Image

2. 支持 Conditional Update（条件更新）

导入数据满足一定条件才更新，典型场景是：上游数据乱续到达，需要通过一个时间戳来判断是否需要更新。

Image

3. 支持 Merge Update

对嵌套数据类型比如 Array 或者 Map/Set 新增加一笔记录。

  • array append
  • map/set add

4. 通用 Read-Write 事务

做一些通用的修复工作，一般是多行的事务。比如要修补一个用户的订单数据，先需要把这个用户的数据删掉，再把新数据导入进来，而且要做到原子生效。

Image

5. Primary Key 表的物化视图

现在的 Primary key 表不支持物化视图，因为如果 Primary key 表一些记录被修改了，无法增量的直接反馈到物化视图中。实现思路是 update Primary key 表之后，需要生成一个 delta 数据，根据 delta 数据写到物化视图中去。

Image
Image

end