Kudu：一个融合低延迟写入和高性能分析的存储系统

Kudu 是一个基于 Raft 的分布式存储系统，它致力于融合低延迟写入和高性能分析这两种场景，而且能很好的嵌入到 Hadoop 生态系统里面，跟其余系统譬如 Cloudera Impala，Apache Spark 等对接。

Kudu 很相似 TiDB。最开始，TiDB 是为了 OLTP 系统设计的，但后来发现咱们 OLAP 的功能也愈来愈强大，因此就有了融合 OLTP 和 OLAP 的想法，固然这条路并非那么容易，咱们还有不少工做要作。由于 Kudu 的理念跟咱们相似，因此我也颇有兴趣去研究一下它，这里主要是依据 Kudu 在 2015 发布的 paper，由于 Kudu 是开源的，而且在不断的更新，因此如今代码里面一些实现可能还跟 paper 不同了，但这里仅仅先说一下我对 paper 的理解，实际的代码我后续研究了在详细说明。

为何须要 Kudu？

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

• 静态数据，数据一般都是使用二进制格式存放到 HDFS 上面，譬如 Apache Avro，Apache Parquet。但不管是 HDFS 仍是相关的系统，都是为高吞吐连续访问数据这些场景设计的，都没有很好的支持单独 record 的更新，或者是提供好的随机访问的能力。
• 动态数据，数据一般都是使用半结构化的方式存储，譬如 Apache HBase，Apache Cassandra。这些系统都能低延迟的读写单独的 record，可是对于一些像 SQL 分析这样须要连续大量读取数据的场景，显得有点捉紧见拙。

上面的两种系统，各有本身的侧重点，一类是低延迟的随机访问特定数据，而另外一类就是高吞吐的分析大量数据。以前，咱们并无这样的系统能够融合上面两种状况，因此一般的作法就是使用 pipeline，譬如咱们很是熟悉的 Kafka，一般咱们会将数据快速写到 HBase 等系统里面，而后经过 pipeline，在导出给其它分析系统。虽然咱们在必定层面上面，咱们其实经过 pipeline 来对整个系统进行了解耦，但总归要维护多套系统。并且数据更新以后，并不能直接实时的进行分析处理，有延迟的开销。因此在某些层面上面，并非一个很好的解决方案。

Kudu 致力于解决上面的问题，它提供了简单的来处理数据的插入，更新和删除，同时提供了 table scan 来处理数据分析。一般若是一个系统要融合两个特性，颇有可能就会陷入两边都作，两边都没作好的窘境，但 Kudu 很好的在融合上面取得了平衡，那么它是如何作到的呢？

Keyword

Tables 和 schemas

Kudu 提供了 table 的概念。用户能够创建多个 table，每一个 table 都有一个预先定义好的 schema。Schema 里面定义了这个 table 多个 column，每一个 column 都有名字，类型，是否容许 null 等。一些 columns 组成了 primary key。

能够看到，Kudu 的数据模型很是相似关系数据库，在使用以前，用户必须首先创建一个 table，访问不存在的 table 或者 column 都会报错。用户可使用 DDL 语句添加或者删除 column，但不能删除包含 primary key 的 column。

但在 Paper 里面说到 Kudu 不支持二级索引以及除了 primary key 以外的惟一索引，这个后续能够经过更新的代码来肯定下。

其实我这里很是关注的是 Kudu 的 Online DDL 是如何作的，只是 Paper 里面貌似没有说起，后面只能看代码了。

API

Kudu 提供了 Insert，Update 和 Delete 的 write API。不支持多行事务 API，这个不知道最新的能支持了没有，由于仅仅能对单行数据操做，还远远不够。

Kudu 提供了 Scan read API 让用户去读取数据。用户能够指定一些特定的条件来过滤结果，譬如用一个常量跟一个 column 里面的值比较，或者一段 primary key 的范围等条件。

提供 API 的好处在于实现简单，但对于用户来讲，其实更好的使用方式仍然是 SQL，一些复杂的查询最好能经过 SQL 搞定，而不是让用户本身去 scan 数据，而后本身组装。

一致性模型

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

默认 Kudu 提供 Snapshot consistency， 它具备更好的读性能，但可能会有 write skew 问题。而 External consistency 则可以彻底保证整个系统的 linearizability，也就是当写入一条数据以后，后面的任何读取都必定能读到最新的数据。

为了实现 External consistency，Kudu 提供了几种方法：

• 在 clients 之间显示的传递时间戳。当写入一条数据以后，用户用要求 client 去拿一个时间戳做为 token，而后经过一个 external channel 的方式传递给另外一个 client。而后另外一个 client 就能够经过这个 token 去读取数据，这样就必定能保证读取到最新的数据了。不过这个方法实在是有点复杂。
• 提供相似 Spanner 的 commit-wait 机制。当写入一条数据以后，client 须要等待一段时间来肯定写入成功。Kudu 并无采用 Spanner TrueTime 的方案，而是使用了 HybridTime 的方案。HybridTime 依赖 NTP，这个可能致使 wait 的时间很长，但 Kudu 认为将来随着 read-time clock 的完善，这应该不是问题了。

Kudu 是我已知的第二个采用 HybridTime 来解决 External consistency 的产品，第一个固然就是 CockroachDB 了。TiDB 跟他们不同，咱们采用的是全局授时的方案，这个会简单不少，但其实也有跟 PD 交互的网络开销。后续TiDB 可能使用相似 Spanner 的 GPS + 原子钟，现阶段相关硬件的制造方式 Google 并无说明，但其实难度不大。由于已经有不少硬件厂商主动找咱们但愿一块儿合做提供，只是比较贵，而现阶段咱们大多数客户并无跨全球事务这种场景。

Kudu 的一致性模型依赖时间戳，这应该是如今全部分布式系统通用的作法。Kudu 并无给用户保留时间戳的概念，主要是以为用户极可能会困惑，毕竟不是全部的用户都能很好的理解 MVCC 这些概念。固然，对于 read API，仍是容许用户指定特定的一个时间戳，这样就能读取到历史数据。这个 TiDB 也是相似的作法，用户不知道时间戳，只是咱们额外提供了一个设置 snapshot 的操做，让用户指定生成某个时间点的快照，读取那个时间点的数据。这个功能已经帮不少公司恢复了由于错误操做写坏的数据了。

架构

上面说了一些 Kudu 的 keyword， 如今来讲说 Kudu 的总体架构。Kudu 相似 GFS，提供了一个单独的 Master 服务，用来管理整个集群的元信息，同时有多个 Tablet 服务，用来存储实际的数据。

分区

Kudu 支持对数据按照 Range 以及 Hash 的方式进行分区。 每一个大的 table 均可以经过这种方式将数据分不到不一样的 Tablet 上面。当用户建立一个表的时候，同时也能够指定特定的 partition schema，partition schema 会将 primary key 映射成对应的 partition key。每一个 Tablet 上面会覆盖一段或者多段 partition keys 的range。当 client 须要操做数据的时候，它能够很方便的就知道这个数据在哪个 Tablet 上面。

一个 partition schema 能够包括 0 或者多个 hash-partitioning 规则和最多一个 range-partitioning 规则。用户能够根据本身实际的场景来设置不一样的 partition 规则。

譬若有一行数据是 (host, metric, time, value)，time 是单调递增的，若是咱们将 time 按照 hash 的方式分区，虽然能保证数据分散到不一样的 Tablets 上面，但若是咱们想查询某一段时间区间的数据，就得须要所有扫描全部的 Tablets 了。因此一般对于 time，咱们都是采用 range 的分区方式。但 range 的方式会有 hot range 的问题，也就是同一个时间会有大量的数据写到一个 range 上面，而这个 hot range 是无法经过 scale out 来缓解的，因此咱们能够将 (host, metric) 按照 hash 分区，这样就在 write 和 read 之间提供了一个平衡。

经过多个 partition 规则组合，能很好的应对一些场景，但同时这个这对用户的要求比较高，他们必须更加了解 Kudu，了解本身的整个系统数据会如何的写入以及查询。如今 TiDB 还只是单纯的支持 range 的分区方式，但将来不排除也引入 hash。

Raft

Kudu 使用 Raft 算法来保证分布式环境下面数据一致性，这里就再也不详细的说明 Raft 算法了，由于有太多的资料了。

Kudu 的 heartbeat 是 500 毫秒，election timeout 是 1500 毫秒，这个时间其实很频繁，若是 Raft group 到了必定量级，网络开销会比较大。另外，Kudu 稍微作了一些 Raft 的改动：

• 使用了 exponential back-off 算法来处理 leader re-election 问题。
• 当一个新的 leader 跟 follower 进行交互的时候，Raft 会尝试先找到这两个节点的 log 分叉点，而后 leader 再从这个点去发送 log。Kudu 直接是经过 committedIndex 这个点来发送。

对于 membership change，Kudu 采用的是 one-by-one 算法，也就是每次只对一个节点进行变动。这个算法的好处是不像 joint consensus 那样复杂，容易实现，但其实仍是会有一些在极端状况下面的 corner case 问题。

当添加一个新的节点以后，Kudu 首先要走一个 remote bootstrap 流程。

1. 将新的节点加入到 Raft 的 configuration 里面
2. Leader 发送 StartEmoteBootstrap RPC，新的 follower 开始拉去 snapshot 和以后的 log
3. Follower 接受完全部数据并 apply 成功以后，开始响应 Raft RPC

能够看到，这个流程跟 TiKV 的作法相似，这个其实有一个缺陷的。假设咱们有三个节点，加入第四个以后，若是新的节点还没 apply 完 snapshot，这时候挂掉了一个节点，那么整个集群实际上是无法工做的。

为了解决这个问题，Kudu 引入了 PRR_VOTER 概念。当新的节点加入的时候，它是 PRE_VOTE 状态，这个节点不会参与到 Raft Vote 里面，只有当这个节点接受成功 snapshot 以后，才会变成 VOTER。

当删除一个节点的时候，Leader 直接提交一个新的 configuration，删除这个节点，当这个 log 被 committed 以后，这个节点就把删除了。被删除的节点有可能不知道本身已经被删除了，若是它长时间没有收到其余的节点发过来的消息，就会问下 Master 本身还在不在，若是不在了，就本身干掉本身。这个作法跟 TiKV 也是相似的。

Master

Kudu 的 Master 是整个集群最核心的东西，相似于 TiKV 里面的 PD。在分布式系统里面，一些系统采用了无中心化的架构设计方案，但我我的以为，有一个中心化的单点，能更好的用全局视角来控制和调度整个系统，并且实现起来很简单。

在 Kudu 里面，Master 本身也是一个单一的 Tablet table，只是对用户不可见。它保存了整个集群的元信息，而且为了性能，会将其所有缓存到内存上面。由于对于集群来讲，元信息的量其实并不大，因此在很长一段时间，Master 都不会有 scale 的风险。同时 Master 也是采用 Raft 机制复制，来保证单点问题。

这个设计其实跟 PD 是同样的，PD 也将全部的元信息放到内存。同时，PD 内部集成 etcd，来保证整个系统的可用性。跟 Kudu Master 不同的地方在于，PD 是一个独立的组件，而 Kudu 的 Master 其实仍是集成在 Kudu 集群里面的。

Kudu 的 Master 主要负责如下几个事情：

Catalog manager

Master 的 catalog table 会管理全部 table 的一些元信息，譬如当前 table schema 的版本，table 的 state（creating，running，deleting 等），以及这个 table 在哪些 Tables 上面。

当用户要建立一个 table 的时候，首先 Master 在 catalog table 上面写入须要建立 table 的记录，table 的 state 为 CREATING。而后异步的去选择 Tablet servers 去建立相关的元信息。若是中间 Master 挂掉了，table 记录里面的 CREATING state 会代表这个 table 还在建立中，新的 Master leader 会继续这个流程。

Cluster coordinator

当 Tablet server 启动以后，会给 Master 注册，而且持续的给 Master 进行心跳汇报消后续的状态变化。

虽然 Master 是整个系统的中心，但它实际上是一个观察者，它的不少信息都须要依赖 Tablet server 的上报，由于只有 Tablet server 本身知道当前本身有哪一些 tablet 在进行 Raft 复制，Raft 的操做是否执行成功，当前 tablet 的版本等。由于 Tablet 的状态变动依赖 Raft，每一次变动其实就在 Raft log 上面有一个对应的 index，因此上报给 Master 的消息必定是幂等的，由于 Master 本身会比较 tablet 上报的 log index 跟当前本身保存的 index，若是上报的 log index 是旧的，那么会直接丢弃。

这个设计的好处在于极大的简化了整个系统的设计，若是要 Master 本身去负责管理整个集群的状态变动，譬如 Master 给一个 tablet 发送增长副本的命令，而后等待这个操做完成，在继续处理后面的流程。整个系统光异常处理，都会变得特别复杂，譬如咱们须要关注网络是否是断开了，超时了究竟是成功了仍是失败了，要不要再去 tablet 上面查一下？

相反，若是 Master 只是给 tablet 发送一个添加副本的命令，而后无论了，剩下的事情就是一段时间后让 tablet 本身上报回来，若是成功了继续后面的处理，不成功则尝试在加一次。虽然依赖 tablet 的上报会有延迟（一般状况，只要有变更，tablet 会及时的上报通知，因此这个延迟其实挺小的），整个架构简单了不少。

其实看到这里的时候，我以为很是的熟悉，由于咱们也是采用的这一套架构方案。最开始设计 PD 的时候，咱们还设想的是 PD 主动去控制 TiKV，也就是我上面说的那套复杂的发命令流程。但后来发现实在是太复杂了，因而改为 TiKV 主动上报，这样 PD 其实就是一个无状态的服务了，无状态的服务好处就是若是挂了，新启动的 PD 能马上恢复（固然，实际仍是要作一些不少优化工做的）。

Tablet directory

由于 Master 知道集群全部的信息，因此当 client 须要读写数据的时候，它必定要先跟 Master 问一下对应的数据在哪个 Tablet server 的 tablet 上面，而后才能发送对应的命令。

若是每次操做都从 Master 获取信息，那么 Master 铁定会成为一个性能瓶颈，鉴于 tablet 的变动不是特别的频繁，因此不少时候，client 会缓存访问的 tablet 信息，这样下次再访问的时候就不用从 Master 再次获取。

由于 tablet 也可能会变化，譬如 leader 跑到了另外一个 server 上面，或者 tablet 已经不在当前 server 上面，client 会收到相关的错误，这时候，client 就从新再去 Master 获取一下最新的路由信息。

这个跟咱们的作法仍然是同样的，client 缓存最近的路由信息，当路由失效的时候，从新去 PD 获取一下。固然，若是只是单纯的 leader 变动，其实返回的错误里面一般就会带上新的 leader 信息，这时候 client 直接刷新缓存，在直接访问了。

Tablet storage

Tablet server 是 Kudu 用来存放实际数据的服务，为了更好的性能，Kudu 本身实现了一套 tablet storage，而没有用现有的开源解决方案。Tablet storage 目标主要包括：

• 快速的按照 Column 扫描数据
• 低延迟的随机更新
• 一致的性能

RowSets

Tablets 在 Kudu 里面被切分红更小的单元，叫作 RowSets。一些 RowSets 只存在于内存，叫作 MemRowSets，而另外一些则是使用 disk 和 memory 共享存放，叫作 DiskRowSets。任何一行数据只存在一个 RowSets 里面。

在任什么时候候，一个 tablet 仅有一个单独的 MemRowSet 用来保存最近插入的数据。后台有一个线程会按期的将 这些 MemRowSets 刷到 disk 上面。

当一个 MemRowSet 被刷到 disk 以后，一个新的空的 MemRowSet 被建立出来。以前的 MemRowSet 在刷到 disk 以后，就变成了 DiskRowSet。当刷的同时，若是有新的写入，仍然会写到这个正在刷的 MemRowSet 上面，Kudu 有一套机制可以保证新写入的数据也能一块儿被刷到 disk 上面。

MemRowSet

MemRowSet 是一个支持并发，提供锁优化的 B-tree，主要基于 MassTree，也有一些不一样：

1. 由于 Kudu 使用的是 MVCC，因此任何的删除其实也是插入，因此这个 tree 没有删除操做。
2. 不支持任意的 in-place 数据变动操做，除非此次操做不会改变 value 的大小。
3. 将 Leaf link 起来，相似 B+-tree，这样对于 scan 会有明显的性能提高。
4. 并无彻底实现 trie of trees，是只是使用了一个单一 tree，由于 Kudu 并无太多高频随机访问的场景。

DiskRowSet

当 MemRowSets 被刷到 disk 以后，就变成了 DiskRowSets。当 MemRowSets 被刷到 disk 的时候，Kudu 发现超过 32 MB 了就滚动一个新的 DiskRowSet。由于 MemRowSet 是顺序的，因此 DiskRowSets 也是顺序的，各滚动的 DiskRowSet 里面的 primary keys 都是不相交的。

一个 DiskRowSet 包含 base data 和 delta data。Base data 按照 column 组织，也就是一般咱们说的列存。各个 column 会被独立的写到 disk 里面一段连续的 block 上面，数据会被切分红多个 page，使用一个 B-tree 进行高效索引。

除了刷用户自定义的 column，Kudu 还默认将 primary key index 写到一个 column，同时使用 Bloom filter 来保证能快速经过找到 primary key。

为了简单，当 column 的数据刷到 disk，它就是默认 immutable 的了，但在刷的过程当中，有可能有更新的数据，Kudu 将这些数据放到一个 delta stores 上面。Delta stores 可能在内存 DeltaMemStores，或者 disk DeltaFiles。

Delta store 维护的一个 map，key 是 (row_offset, timestamp)，value 就是 RowChangeList 记录。Row offset 就是 row 在 RowSet 里面的索引，譬如，有最小 primary key 的 row 在 RowSet 里面是排在最前面的，它的 offset 就是 0。Timestamp 就是一般的 MVCC timestamp。

当须要给 DiskRowSet 更新数据的时候，Kudu 首先经过 primary key 找到对应的 row。经过 B-tree 索引，能知道哪个 page 包含了这个 row，在 page 里面，能够计算 row 在整个 DiskRowSet 的 offset，而后就把这个 offset 插入到 DeltaMemStore 里面。

当 DeltaMemStore 超过了一个阀值，一个新的 DeltaMemStore 就会生成，原先的就会被刷到 disk，变成 immutable DeltaFile。

每一个 DiskRowSet 都有一个 Bloom filter，便于快速的定位一个 key 是否存在于该DiskRowSet 里面。DIskRowSet 还保存了最小和最大的 primary key，这样外面就能经过 key 落在哪个 key range 里面，快速的定位到这个 key 属于哪个 DiskRowSet。

Compaction

当作查询操做的时候，Kudu 也会从 DeltaStore 上面读取数据，因此若是 DeltaStore 太多，整个读性能会急剧降低。为了解决这个问题，Kudu 在后台会按期的将 delta data 作 compaction，merge 到 base data 里面。

同时，Kudu 还会按期的将一些 DIskRowSets 作 compaction，生成新的 DiskRowSets，对 RowSet 作 compaction 能直接去掉 deleted rows，同时也能减小重叠的 DiskRowSets，加速读操做。

总结

上面对 Kudu 大概进行了介绍，主要仍是参考 Kudu 本身的论文。Kudu 在设计上面跟 TiKV 很是相似，因此对于不少设计，我是特别能理解为啥要这么作的，譬如 Master 的信息是经过 tablet 上报这种的。Kudu 对 Raft 在实现上面作了一些优化，以及在数据 partition 上面也有不错的作法，这些都是后面能借鉴的。

对于 Tablet Storage，虽然 Kudu 是本身实现的，但我发现，不少方面其实跟 RocksDB 差不了多少，相似 LSM 架构，只是可能这套系统专门为 Kudu 作了定制优化，而不像 RocksDB 那样具备普适性。对于 storage 来讲，如今咱们仍是考虑使用 RocksDB。

另外，Kudu 采用的是列存，也就是每一个列的数据单独聚合存放到一块儿，而 TiDB 这边仍是主要使用的行存，也就是存储整行数据。列存对于 OLAP 很是友好，但在刷盘的时候压力可能会比较大，若是一个 table 有不少 column，写入性能可会有点影响。行存则是对于 OLTP 比较友好，但在读取的时候会将整行数据全读出来，在一些分析场景下压力会有点大。但不管列存仍是行存，都是为知足不一样的业务场景而服务的，TiDB 后续其实能够考虑的是行列混存，这样就能适配不一样的场景了，只是这个目标比较远大，但愿感兴趣的同窗一块儿加入来实现。