Kudu

2022-08-22

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分区,这样就在写入并行性和查询并发能力之间提供了一个平衡  

Replication

Kudu使用Raft算法来实现数据Replication,对于Raft算法,这里不再赘述了。具体可参考raft

The Kudu Master

标题使用了The Kudu Master,显而易见,整个Kudu集群只有一个Master。Kudu mater有如下几个职责:

  • Catalog Manager

Master的catalog table会管理所有table的一些元信息,例如当前table schema的版本、table的state(creating,running,deleting等)、以及这个table在哪些tablet server上。

当用户要创建一个table的时候,首先Master在catalog table上面写入需要创建table的记录,table的state为CREATING。然后异步的去选择Tablet servers去创建相关的元信息。如果中间Master挂掉了,table 记录里面的CREATING state会表明这个table还在创建中,新的Master leader会继续这个流程。

  • Cluster Coordination

当tablet server启动之后,会给Master注册,并且持续的给Master进行心跳汇报消后续的状态变化。

虽然Master是系统中catalog的信息源,但它其实是一个观察者(Observer),它的很多信息都需要依赖tablet server的上报。因为只有tablet server自己知道当前自己有哪些tablet在进行Raft复制,Raft的操作是否执行成功,当前tablet的版本等等。由于Tablet的状态变更依赖Raft,每一次变更其实就在Raft log上面有一个对应的index,Master比较tablet上报的log index跟当前自己保存的index,如果上报的log index是旧的,那么会直接丢弃。

这个设计将很多职责交给了tablet server,极大简化了整个系统的设计。

  • Tablet Directory

Master知道集群所有的信息,所以当client需要读写数据的时候,一定要先向Master获取对应的数据在哪一个tablet server的tablet上面,然后才能发送对应的命令。

这里有个问题,如果每次操作都从Master获取信息,那么Master将会成为一个性能瓶颈,鉴于tablet的变更不是特别的频繁,所以很多时候,client会缓存访问的tablet信息,这样下次再访问的时候就不用从Master再次获取。当tablet发生变化时,例如leader跑到了另一个server上面,或者tablet已经不在当前server上,client会收到相关的错误,这时client就重新再去Master获取一下最新的路由信息。

Tablet Storage

Tablet Storage是Kudu的核心部分,也是Kudu为何可以做到OLAP和OLTP平衡的关键点。其设计目标主要包括:

  • 快速的(列式)扫描数据

  • 快速的随机访问,使得随机访问查找复杂度为O(n)

  • 一致的性能,用户更愿意以峰值性能换取可预测的性能。

这些目标使得kudu既可以支持AP场景、也可以支持TP场景,当然总体还是偏向AP一些。

Rowsets

MemRowSet

DiskRowSet