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 後續其實能夠考慮的是行列混存，這樣就能適配不一樣的場景了，只是這個目標比較遠大，但願感興趣的同窗一塊兒加入來實現。