Kudu

[levy5307]

https://levy5307.github.io/blog/kudu/

Kudu是一款基于Raft实现的列式分布式存储系统，可以同时满足低延迟写入和高性能分析两种场景。

结构化数据存储系统在Hadoop生态系统里面，通常分为两类：

静态数据。通常都是使用二进制存储在HDFS上面，例如Apache Avro，Apache Parquet。这类系统都是为高吞吐连续访问数据这类场景设计的，对indivial records这种随机更新支持不够好。


动态数据。通常使用半结构化方式存储，例如Apache HBase，Apache Cassandra。这些系统都能低延迟的随机读写indival records，但是对于一些像SQL分析这样需要连续大量读取数据的场景支持的不够好。

当需要上述两种场景时，通常的做法是使用data pipeline。例如，将流式数据写入HBase，随后由一些周期作业将数据导入到Parquet中，以备后续的分析使用。这样做有如下几个缺点：

需要实现很复杂的代码管理两个系统之间的数据同步


需要实现备份、安全策略、监控系统等，导致系统很庞杂


系统从实时系统中流出到离线系统才好做OLAP分析，这层转换存在延迟


系统总是会存在落后的数据，这些数据是对过去数据的修改和删除。然而当过去的数据已经被归档，这些操作需要昂贵的重写、partition swapping或者各种人工干预

Kudu弥补了高吞吐连续访问和随机读写之间的gap，官方称其为happy medium。

Kudu at a high level

table and schemas

Kudu提供了table的概念。用户可以建立多个table，每个table都有一个预先定义好的schema。Schema里面定义了这个table中的column，以及每个column的名字、类型、是否允许Null等。其中的一些columns组成了primary key。primary key强制唯一性约束，并且会作为唯一索引存在，用于高效的更新和删除。

在使用之前，用户必须首先建立一个table，并且可以使用alter table语句添加或者删除column(但不能删除包含primary key的column)。

Kudu里没有使用NoSQL中的“everything is byte”的设计理念，主要基于如下考虑：

显式类型使得用于可以针对不同类型使用不同的编码。


显式类型允许暴露出SQL-like metadata给BI报表等应用，交互体验更好。

另外，Kudu不支持二级索引，以及除了primary key之外的唯一索引。

Read And Write

对于Write操作，Kudu提供了Insert，Update和Delete的write API。不支持多行事务API。

而对于Read操作，只提供了Scan read API让用户去读取数据。目前提供了两种谓词来过滤结果：

一个常量跟一个column的值比较


一段primary key的范围

Consistency Model

Kudu提供两种一致性模型：snapshot consistency和external consistency。

默认采用Snapshot consistency，它具有更好的读性能，缺点是会有write skew 问题。而External consistency则能够完全保证整个系统的linearizability，也就是当写入一条数据之后，后面的任何读取都一定能读到最新的数据。

Timestamps

对于写操作，kudu不允许用于手动设置timestamp，因为根据kudu团队的经验，该timestamp常常会带来困惑，并容易引来问题。而对于读操作，则允许指定timestamp。这使得用户可以使用point-in-time queries，确保可以使得多个distributed tasks来共同完成一个单一的query。

Architecture

类似与GFS和Bigtable，Kudu提供了一个单独的Master服务，用来管理整个集群的元信息，同时有多个Tablet server，用来存储实际的数据。

Partitioning

同其他数据库系统一样，kudu中的表支持水平分，这些partition成为tablet。任何一个行数据都会依据primary key的值而映射到一个tablet上，这样一个update或者insert操作只会影响一个tablet。

不像Bigtable（Range）和Cassandra（Hash）仅支持一个partition方式，Kudu支持指定一系列partition schemes。当用户创建一个表的时候，同时也可以指定特定的partition schema，partition schema会将primary key映射成对应的partition key。每个Tablet上面会覆盖一段或者多段partition keys的range。当client需要操作数据的时候，它可以很方便的就知道这个数据在哪一个Tablet上面。

partition schema: primary key --> partition key

另外，一个partition schema可以包括0或者多个hash-partitioning规则以及最多一个range-partitioning规则:

hash-partitioning规则包含primary key的子集以及bucket的数量，例如：

DISTRIBUTED BY HASH(hostname, ts) INTO 16 BUCKETS

表示将hostname和ts拼接后进行hash，并将hash后获取的值对bucket数量取模，即：

paritition key = hash(hostname + ts) mod 16

取模后的连续range会存储到一个Tablet上。

range-partitioning规则使用primary key的有序子集，对该primary key的子集拼接后，形成partition key

例如，当前有一个时序应用，表schema是 (host, metric, time, value)，其中time是单调递增的，如果我们将time按照hash的方式分区，虽然能保证数据分散到不同的Tablets上面，但如果我们想查询某一段时间区间的数据，就得需要全部扫描所有的Tablets了，这严重影响了查询的并发能力。

所以通常对于time，我们都是采用range的分区方式。但range的方式会有hot range的问题，也就是同一个时间会有大量的数据写到一个range上面，而这个hot range是没法通过scale out来缓解的。所以我们可以额外将(host, metric)按照hash分区，这样就在写入并行性和查询并发能力之间提供了一个平衡