处理海量数据：列式存储综述（系统篇）

发表于 更新于 Disqus：

在上一篇文章《处理海量数据：列式存储综述（存储篇）》 中，我们介绍了几种 Apache ORC、Dremel 等几种典型列式存储的数据组织格式。实践中，很多数据系统构建在 HDFS 等分布式文件系统之上，使用这些规范的格式保存数据，方便各方的业务进行分析查询。

本文将介绍若干个典型的列式存储数据库系统。作为完整的 OLAP 或 HTAP 数据库系统，他们大多使用了自主设计的存储方式，运行在多台机器节点上，使用网络进行通讯协作。

C-Store (2005) / Vertica

大多数 DBMS 都是为写优化，而 C-Store 是第一个为读优化的 OLTP 数据库系统，虽然从今天的视角看它应当算作 HTAP 。在 ad-hoc 的分析型查询、ORM 的在线查询等场景中，大多数操作都是查询而非写入，在这些场景中列式存储能取得更好的性能。像主流的 DBMS 一样，C-Store 支持标准的关系型模型。

在上一篇文章中，我们已经介绍了 C-Store 特有的 projection 数据模型。这里只做一个简单的回顾：每个 projection 可以包含一个或多个列，完整的表视图需要通过若干 projection JOIN 得到。Projection 水平拆分成若干 segments。

C-Store 的设计考虑到企业级应用的使用模式，在优化 AP 查询的同时兼顾了大多数 DBMS 具有的 TP 查询功能。在 ACID 事务方面同样提供了完整的支持，支持快照（snapshot）读事务和一般的 2PC 读写事务。

通常而言，互联网应用对 DBMS 有较高的并发写入需求，对一致性读、分析型查询的需求不那么强烈。而企业级应用（例如 CRM 系统）的并发写入需求不大，但需要对一致性读、分析型查询等。

系统设计

C-Store 将其物理存储也就是每个 projection 分成两层，分别是为写优化的 Writeable Store (WS) 和为读优化的 Read-optimized Store (RS)。RS 即是基线数据，WS 上暂存了对 RS 数据的变更，二者在读取时需要 merge 得到最新的数据。在上一篇文章的 Apache ORC 格式种，我们也看到了类似的做法（基线数据叠加增量数据）。

RS 是一个为读优化的列式存储。RS 中采用之前提到的 projection 数据模型，对不同的列采用了不同的编码方式，根据它是否是 projection 的排序列、以及该列的取值个数，来决定采取何种编码方式。

WS 用于暂存高性能的写入操作，例如 INSERT、UPDATE 等。为了简化系统的设计，WS 逻辑上仍然按照 projection 的列式模型存储，但是物理上使用 B 树以满足快速的写入要求。WS 基于 BerkeleyDB 构建。

对于某一列中的某个值 v，会有两个映射关系存在：一是 (storage_key -> v)，在 RS 中 storage_key 就是 segment 中的行号，但在 WS 中显式的记录下来；二是 (sort_key -> storage_key)，用于满足主键查询的需求。

值得一提的是，WS 是一个 MVCC 的存储——它的每个数据都保存了对应的写入事务编号，同一行可能有多个版本同时存在。而 RS 是没有 MVCC 的，你可以将它看作过去某个时间点的快照。

Tuple Mover 周期性地将 WS 中的数据移动到 RS 中。与大多数 MVCC 系统一样，C-Store 中的更新是通过一个删除加一个插入实现的，Tuple Mover 的主要工作是根据 WS 的数据更新 RS：删掉被删除的行、添加新的行。

事务支持

C-Store 认为大多数事务是只读事务，因此采用了 Snapshot Isolation。C-Store 维护两个全局的时间戳：低水位（Low Water Mark, LWM）和高水位（High Water Mark, HWM），允许用户查询介于二者之间的任意时间戳的 Snapshot。时间戳来自中心化的 Time Authority (TA)。

LWM 对应 RS 即基线数据的版本。Tuple Mover 会保证任何高于 LWM 的修改都不会被移动到 RS 中，因为一旦移动到 RS 也就失去了多版本。

HWM 由中心的 TA 维护，时间被分成固定长度的 epoch。当各个节点确认 epoch e 中开始的写入事务完成时，就会发送一个 Complete(e) 的消息给 TA，当 TA 收集到所有节点的 Complete(e) 将 HWM 置为 e。换句话说，HWM 以前的事务一定是已经完成提交的。

对于读写事务，C-Store 采用了传统的 2PC。

MonetDB (2012) / VectorWise

MonetDB 是一个面向 OLAP 的内存数据。区别于大多数 DBMS 使用的 Valcano 执行模式，MonetDB 使用一种独特的 full materialization 的列式（向量）执行模型，也因此设计了对应的一系列算子以及查询优化器。

BAT Algebra

MonetDB 独有的列式计算是通过 BAT（Binary Association Table）的运算组成的，BAT 之间通过算子产生新的 BAT，最终生成查询结果。每个 BAT 可以简单地理解为一列带有编号的数据 <oid, value>，有些 BAT 来自用户的逻辑表，其他则是运算的结果。每个算子被设计地很紧凑、高效，能充分利用 CPU 流水线的计算能力，这和 CPU 设计的 RISC 思想颇为相似，所以被称为“数据库查询的 RISC 方案”。

如上图，对于用户一条 SELECT 查询，MonetDB 先将其分解为多次 BAT 的运算，执行计划中的每一步的输入和输出都是 BAT。图中蓝框中为输入的 BAT，其他则是执行产生的运算结果。

MonetDB 的设计决定了它的计算过程十分耗费内存。MonetDB 利用操作系统的 Memory Mapped File 进行内存管理，不使用的页面可以被换出内存，为执行查询腾出空间。但显然这并不是一个彻底的解决方案。

VectorWise 使用类似的向量化执行模型，但它尝试在 full materialization 和 Valcano 模型中间寻求一个平衡——它将整个列划分成较小的 block，对 block 进行上述的 column algebra 计算。

Apache Kudu (2015)

Kudu 是 Cloudera 研发的处理实时数据的 OLAP 数据库。上文提到的 Parquet / ORC 是开源界常用的处理静态数据的方式，为什么说是静态数据呢？因为这些紧凑的格式对数据修改很不友好，且随机读写性能极差，通常只能用于后台 OLAP。

所以我们看到，很多数据系统都采用动态、静态两套数据，例如：把在线业务数据放在 HBase 中，定期通过 ETL 程序产生Parquet 格式文件放到 HDFS 上，再对其进行统计、归档等。这种定期导入的方式不可避免地会带来小时级的延迟，而且，众所周知维护 ETL 代码是一件费时费力的事情。

Kudu 试图在 OLAP 与 OLTP 之间寻求一个平衡点——在保持同一份数据的情况下，既能提供在线业务实时写入的能力，又能支持高效的 OLAP 查询。

Kudu 采用我们熟悉的半关系型模型，允许用户定义 schema，但是目前并不支持二级索引。

事务方面，Kudu 默认使用 Snapshot Isolation 一致性模型。此外，如果用户需要一个更强的一致性保证（例如 read own's writes），Kudu 也允许用户指定特定的时间戳，读取这个时间戳的 snapshot。这项功能被集成在 Kudu 的 API 层面，用户可以方便地获得因果（causality）一致性保证。

系统设计

Kudu 采用了类似 HBase 的 master-slave 架构：中心节点被称作 Kudu Master，数据节点被称作 Tablet Server。一个表的数据被分割成多个 tablets，由它们对应的 Tablet Server 来提供数据读写服务。

与 HBase 相比，中心节点 Kudu Master 除了存放了 Tablet 的分布信息，还身兼了如下角色：

  • Catalog 管理：同步各个库、表的 schema 等元信息、负责协调完成建表等 DDL 操作
  • 集群协调者：各个 Tablet Server 向其汇报自己的状态、replica 变更等

Kudu 底层数据文件并没有存储在 HDFS 这样的分布式文件系统上，而是基于 Raft 算法实现了一套副本同步机制，保障数据不丢失及高可用性。其中 Raft 算法用于同步数据修改操作的 log，这点和大多数 shared-nothing 架构分布式数据库并无二致。对 Raft 算法有兴趣的同学可以参考原论文。

作为列式 OLAP 数据库，Kudu 的磁盘存储是常见的列式方案，很多地方直接复用了 Parquet 的代码。我们知道，紧凑的列式存储难以实现高效的更新操作。Kudu 为了提供实时写入功能，采用了类似 C-Store 中的方案——在不可变的基线数据上，叠加后续的更新数据。

具体来说，Tablet 由 RowSet 组成，而 RowSet 既可以是内存中的 MemRowSet，也可以是存储在磁盘上的 DiskRowSet。一个 RowSet 包含两部分数据：基础数据通常以 DiskRowSet 形式保存在磁盘上；而变更数据先以 MemRowSet 的形式暂存在内存中，后续再异步地刷写到磁盘上。和 C-Store 类似，内存中的数据使用 B 树存储。

与 C-Store 不同的是，Delta 数据并不会立即和磁盘上的基线数据进行合并，而是由后台的 compaction 线程异步完成。值得注意的是，为了保证 compaction 操作不影响过去的 snapshot read，被覆盖的旧数据也会以 UNDO 记录的形式保存在另外的文件中。

PowerDrill (2012)

PowerDrill 是 Google 研发用于快速处理 ad-hoc 查询的 OLAP 数据库，为前端的 Web 交互式分析软件提供支持。PowerDrill 的数据放在内存中，为了尽可能节约空间，PowerDrill 引入一种全新的分区的存储格式，在节省内存占用的同时提供了类似索引的功能，能过滤掉无关的分区、避免全表扫描。

同是 Google 家的产品，和 Dremel 相比，PowerDrill 有以下几点差异：

  • 定位不同：Dremel 用于查询“大量的大数据集”（数据集的规模都大，数据集很多），PowerDrill 用于查询“少量的大数据集”（数据集的规模大，但数据集不多）
  • Dremel 用全表扫描（full scan）处理查询，而 PowerDrill 对数据做了分区，并能根据查询只扫描用到的分区。
  • Dremel 使用类似 Protobuf 的嵌套数据模型；PowerDrill 使用关系模型
  • Dremel 的数据直接放在分布式文件系统上，而 PowerDrill 需要一个 load 过程将数据载入内存

数据分区

Ad-hoc 查询常常包含 GROUP BY 子句，在这些 group key 上进行分区，能很好的过滤掉不需要的数据。PowerDrill 需要 DBA 根据自己对数据的理解，选出用于用于分区的一组属性 Key1 Key2 Key3 ...（优先级依次递减）。分区是一个递归的过程：一开始把整个数据集视为一个分区（Chunk），如果 Key1 能将数据分开就用 Key1，否则用 Key2、Key3—……直到分区大小小于一个阈值。

以下是一个分区的例子，第一次使用 Age 分区、第二次使用 Salary 分区。

数据结构

PowerDrill 的数据组织以列为单位。对于每个列有一个全局字典表，列的每个分区有一个分区字典表：

  • 全局字典表（global dictionary）存储列中所有 distinct 的字符串，按字典顺序排序。字典结构是双向的，既能将 string 映射到 global-id，也能从 global-id 查 string。
  • 分区字典表（chunk dictionary）存储一个分区中 chunk-id 到 global-id 的双向映射。相应地，数据列（elements）存储 chunk-id 而不是 global-id。

如果要将 chunk 中的一个 element 也就是 chunk-id 还原成数据，第一步需要查分区字典表，得到 global-id；第二步查全局字典表，得到原本的字符串数据。以上图举例而言：

  1. Chunk 0 存储的 chunk-id 数据 [3, 2, 0, ...]
  2. 根据分区字典表，查出 global-id：[5, 4, 1, ...]
  3. 根据全局字典表，查出 search string: ['ebay', 'cheap flights', 'amazon', ...]

这样的两层映射保证 chunk-id 尽可能的小，所以可以用更紧凑的编码，比如用 8bit、16bit 整数存储。这不仅能节省空间，也能加快扫描速度。

此外，相同的数据只会在全局字典表中存一份。而且全局字典表中的字符串数据已经被排序，相比不排序，排序后用 Snappy 等算法的压缩比更高。

分区索引

上述的数据结构还有一个额外的好处：它能快速算出某个分区是否包含有用的数据，帮助执行器跳过无关的分区。以下面的 SQL 为例（数据参考上一张图）：

1
2
3
4
SELECT search_string, COUNT(*) as c FROM data
WHERE search_string IN ("la redoute", "voyages sncf")
GROUP BY search_string
ORDER BY c DESC LIMIT 10;

步骤如下：

  1. 在 search_string 列的全局字典表中查找 "[la redoute", "voyages sncf"]，得到 global-id [9, 11]
  2. 在各个分区中查找 global-id [9, 11]： Chunk 0，Chunk 1 中都没有找到，所以可以直接跳过；而 Chunk 2 中出现了 [11]，对应 chunk-id 为 [4]
  3. 在 Chunk 2 中的 elements 扫描查出 chunk-id = 4 的元素数量一共有 3 次，作为 COUNT(*) 的结果返回。

总结

本文介绍了几个知名的列式存储系统。与上一篇文章不同，本文的系统大多重新设计了存储层。与此同时，系统的复杂性也大大提升。

在构建自己的数据系统时，除了存储方式本身，以下几个地方是着重需要考虑清楚的地方，上述的几个系统也给我们提供了很好的参考：

  • 系统需要处理的查询是怎样的模式？C-Store 主要服务于企业级 HTAP 场景，Kudu 在提供 OLAP 查询能力的同时保持了一定的实时写入能力，PowerDrill 着重处理 ad-hoc 的分析型查询。
  • 系统如何保证数据的持久性和高可用性？C-Store 在 projection 上保留了一定的冗余，Kudu 用 Raft 协议保持各个副本的数据一致性及可用性，PowerDrill 则直接把数据放在分布式文件系统上，因为不需要对数据作修改。
  • 系统提供怎样的数据一致性保证？对于只读的系统来说，这不是个问题。但是一旦支持写入，数据的一致性、事务隔离性都需要精心的考虑和权衡。Kudu 和 C-Store 的 Snapshot Read 实现可作为参考。

References

  1. C-store: a column-oriented DBMS - M Stonebraker, DJ Abadi, A Batkin, X Chen…
  2. The Design and Implementation of Modern Column-Oriented Database Systems - D Abadi, P Boncz, S Harizopoulos…
  3. Kudu: Storage for Fast Analytics on Fast Data - T Lipcon, D Alves…
  4. Processing a Trillion Cells per Mouse Click - A Hall, O Bachmann…