Apache Kudu： Hadoop生態系統的新成員實現對快速資料的快速分析
A new addition to the open source Apache Hadoop ecosystem, Apache Kudu completes Hadoop’s storage layer to enable fast analytics on fast data.
開源Apache Hadoop生態系統的新成員，Apache Kudu完善了Hadoop的儲存層，以實現對快速資料的快速分析。
Kudu 是 Cloudera 開源的結構化資料的開源儲存引擎，是 Apache Hadoop 生態圈的新成員之一，支援低延遲隨機訪問以及高效的分析訪問模式。 Kudu使用水平分割槽分發資料，並使用 Raft 複製每個分割槽，提供較低的平均恢復時間和較低的尾部延遲。 Kudu 是在 Hadoop生態系統的環境中設計的，並通過諸如Cloudera Impala ，Apache Spark 和 MapReduce 等工具支援多種訪問模式。
Kudu是一種新的儲存系統，從頭開始設計和實施，以填補HDFS等高吞吐量順序存取儲存系統與HBase或Cassandra等低延遲隨機存取系統之間的這種差距。
雖然這些現有系統在某些情況下仍然具有優勢，但Kudu提供了一種替代方案，可以大大簡化許多常見工作負載的架構。 特別是，Kudu為行級插入，更新和刪除提供了一個簡單的API，同時提供類似於Parquet的表掃描。
本文主要對 Kudu 的背景以及架構進行簡單介紹。
1、背景
1.1、功能上的空白
Hadoop 生態系統有很多元件， 每一個元件有不同的功能。 在現實場景中， 使用者往往需要同時部署很多 Hadoop 工具來解決同一個問題， 這種架構稱為 混合架構 (hybrid architecture)
很多公司都部署了 HDFS/Parquet + HBase 混合架構，然而這種架構較為複雜，維護上也十分困難。首先，使用者需要使用Flume 或 Kafka 等資料Ingest工具將資料匯入HBase，使用者可能會在HBase上對資料做一些修改，然後每隔一段時間將資料從HBase 匯入到 Parquet 檔案，作為一個新的partition 放在 HDFS 上，最後使用Impala 等計算引擎進行查詢，最終生成報表。
這樣一條工具鏈繁瑣而複雜，而且還存在很多問題，比如：
1、如何處理某一過程失敗？
2、從HBase將資料匯出到檔案，頻率多少比較合適？
3、當生產報表時，最近的資料如何體現在查詢結果上？
4、維護叢集時，如何保證關鍵任務不失敗？
5、Parquet 是 immutable 的，因此當HBase 中資料變更時，往往需要人工干預同步
這個時候，使用者就希望能夠有一種優雅的儲存解決方案，來應付不同型別的工作流程，並保持高效能的計算能力。Cloudera 很早就意識到這個問題，在2012年開始計劃開發Kudu這個儲存系統，在2015年釋出並開源。Kudu 是對HDFS 和 HBase 功能上的補充，能提供快速的分析和實時計算能力，並且充分利用CPU 和 I/O 資源，支援資料來源的修改，支援簡單的、可擴充套件的資料模型。
1.2、新硬體裝置的發展
RAM的技術發展非常快，它變得越來越便宜，容量也越來越大。Cloudera的客戶資料顯示，他們的客戶所部署的伺服器，2012年每個節點僅有32GB RAM，現如今增長到每個節點有128GB或256GB RAM。儲存裝置上更新也非常快，在很多普通伺服器中部署SSD也是屢見不鮮。HBase、HDFS、以及其他的Hadoop工具都在不斷自我完善，從而適應硬體上的升級換代。然而，從根本上，HDFS基於03年GFS，HBase基於05年BigTable，在當時系統瓶頸主要取決於底層磁碟速度。當磁碟速度較慢時，CPU利用率不足的根本原因是磁碟速度導致的瓶頸，當磁碟速度提高了之後，CPU利用率提高，這時候CPU往往成為系統的瓶頸。HBase、HDFS由於年代久遠，已經很難從基本架構上進行修改，而Kudu是基於全新的設計，因此可以更充分地利用RAM、I/O資源，並優化CPU利用率。我們可以理解為，Kudu相比與以往的系統，CPU使用降低了，I/O的使用提高了，RAM的利用更充分了。
2、簡介
Kudu設計之初，是為了解決一下問題：
1、對資料掃描(scan)和隨機訪問(random access)同時具有高效能，簡化使用者複雜的混合架構
2、高CPU效率，使使用者購買的先進處理器的的花費得到最大回報
3、高IO效能，充分利用先進儲存介質
4、支援資料的原地更新，避免額外的資料處理、資料移動
5、支援跨資料中心replication
Kudu的很多特性跟HBase很像，它支援索引鍵的查詢和修改。Cloudera曾經想過基於Hbase進行修改，然而結論是對HBase的改動非常大，Kudu的資料模型和磁碟儲存都與Hbase不同。HBase本身成功的適用於大量的其它場景，因此修改HBase很可能吃力不討好。最後Cloudera決定開發一個全新的儲存系統。
Kudu的定位是提供”fast analytics on fast data”，也就是在快速更新的資料上進行快速的查詢。它定位OLAP和少量的OLTP工作流，如果有大量的random accesses，官方建議還是使用HBase最為合適。
3、架構與設計
3.1、基本框架
Kudu是用於儲存結構化(structured)的表(Table)。表有預定義的帶型別的列(Columns)，每張表有一個主鍵(primary key)。主鍵帶有唯一性(uniqueness)限制，可作為索引用來支援快速的random access。
類似於BigTable，Kudu的表是由很多資料子集構成的，表被水平拆分成多個Tablets. Kudu用以每個tablet為一個單元來實現資料的durability。Tablet有多個副本，同時在多個節點上進行持久化。
Kudu有兩種型別的元件，Master Server和Tablet Server。Master負責管理元資料。這些元資料包括talbet的基本資訊，位置資訊。Master還作為負載均衡伺服器，監聽Tablet Server的健康狀態。對於副本數過低的Tablet，Master會在起replication任務來提高其副本數。Master的所有資訊都在記憶體中cache，因此速度非常快。每次查詢都在百毫秒級別。Kudu支援多個Master，不過只有一個active Master，其餘只是作為災備，不提供服務。
Tablet Server上存了10~100個Tablets，每個Tablet有3(或5)個副本存放在不同的Tablet Server上，每個Tablet同時只有一個leader副本，這個副本對使用者提供修改操作，然後將修改結果同步給follower。Follower只提供讀服務，不提供修改服務。副本之間使用raft協議來實現High Availability，當leader所在的節點發生故障時，followers會重新選舉leader。根據官方的資料，其MTTR約為5秒，對client端幾乎沒有影響。Raft協議的另一個作用是實現Consistency。Client對leader的修改操作，需要同步到N/2+1個節點上，該操作才算成功。
Kudu 採用了類似 log-structured 儲存系統的方式，增刪改操作都放在記憶體中的 buffer ，然後才 merge 到持久化的列式儲存中。 Kudu 還是用了 WALs 來對記憶體中的 buffer 進行災備。
3.2、列式儲存
持久化的列式儲存儲存，與 HBase 完全不同，而是使用了類似 Parquet 的方式，同一個列在磁碟上是作為一個連續的塊進行存放的。例如，圖中左邊是 twitter 儲存推文的一張表，而圖中的右邊表示了表在磁碟中的的儲存方式，也就是將同一個列放在一起存放。這樣做的第一個好處是，對於一些聚合和 join 語句，我們可以儘可能地減少磁碟的訪問。例如，我們要使用者名稱為 newsycbot
的推文數量，使用查詢語句：
SELECT COUNT(*) FROM tweets WHERE user_name = ‘newsycbot’
我們只需要查詢 User_name 這個 block 即可。同一個列的資料是集中的，而且是相同格式的， Kudu 可以對資料進行編碼，例如字典編碼，行長編碼， bitshuffle 等。通過這種方式可以很大的減少資料在磁碟上的大小，提高吞吐率。除此之外，使用者可以選擇使用通用的壓縮格式對資料進行壓縮，如 LZ4, gzip, 或 bzip2 。這是可選的，使用者可以根據業務場景，在資料大小和 CPU 效率上進行權衡。這一部分的實現上， Kudu 很大部分借鑑了 Parquet 的程式碼。
HBase 支援 snappy 儲存，然而因為它的 LSM 的資料儲存方式，使得它很難對資料進行特殊編碼，這也是 Kudu 聲稱具有很快的 scan 速度的一個很重要的原因。不過，因為列式編碼後的資料很難再進行修改，因此當這寫資料寫入磁碟後，是不可變的，這部分資料稱之為 base 資料。 Kudu 用 MVCC （多版本併發控制）來實現資料的刪改功能。更新、刪除操作需要記錄到特殊的資料結構裡，儲存在記憶體中的 DeltaMemStore 或磁碟上的 DeltaFIle 裡面。 DeltaMemStore 是 B-Tree 實現的，因此速度快，而且可修改。磁碟上的 DeltaFIle 是二進位制的列式的塊，和 base 資料一樣都是不可修改的。因此當資料頻繁刪改的時候，磁碟上會有大量的 DeltaFiles 檔案， Kudu 借鑑了 Hbase 的方式，會定期對這些檔案進行合併。
3.3、對外介面
Kudu 提供 C++ 、 JAVA API 、Python API，可以進行單條或批量的資料讀寫， schema 的建立修改。除此之外， Kudu 還將與 hadoop 生態圈的其它工具進行整合。目前， kudu 對 Impala 支援較為完善，支援用 Impala 進行建立表、刪改資料等大部分操作。 Kudu 還實現了 KuduTableInputFormat 和 KuduTableOutputFormat ，從而支援 Mapreduce 的讀寫操作。同時支援資料的 locality 和 spark 的整合。
Kudu提供了與如下系統的整合：
MapReduce: 提供針對Kudu使用者表的Input以及Output任務對接。
Spark: 提供與Spark SQL以及DataFrames的對接。
Impala: Kudu自身未提供Shell以及SQL Parser，它的SQL能力源自與Impala的整合。在這些整合中，能夠很好的感知Kudu表資料的本地性資訊，能夠充分利用Kudu所提供的過濾器對查詢進行優化，同時，Impala本身的DDL/DML語法針對Kudu也做了一些擴充套件
3.4、Kudu的底層資料模型
Kudu的底層資料檔案的儲存，沒有采用HDFS這樣的較高抽象層次的分散式檔案系統，而是自行開發了一套可基於Table/Tablet/Replica檢視級別的底層儲存系統。這套實現基於如下的幾個設計目標：
1、可提供快速的列式查詢。
2、可支援快速的隨機更新
3、可提供更為穩定的查詢效能保障。
為了實現如上目標，Kudu參考了一種類似於Fractured Mirrors的混合列儲存架構。Tablet在底層被進一步細分成了一個稱之為RowSets的單元
MemRowSets 可以理解成HBase中的MemStore,DiskRowSets可理解成HBase中的HFile。
MemRowSets中的資料按照行試圖進行儲存，資料結構為B-Tree。MemRowSets中的資料被Flush到磁碟之後，形成DiskRowSets。 DisRowSets中的資料，按照32MB大小為單位，按序劃分為一個個的DiskRowSet。
DiskRowSet中的資料按照Column進行組織，與Parquet類似。這是Kudu可支援一些分析性查詢的基礎。每一個Column的資料被儲存在一個相鄰的資料區域，而這個資料區域進一步被細分成一個個的小的Page單元，與HBase File中的Block類似，對每一個Column Page可採用一些Encoding演算法，以及一些通用的Compression演算法。
既然可對Column Page可採用Encoding以及Compression演算法，那麼，對單條記錄的更改就會比較困難了。
前面提到了Kudu可支援單條記錄級別的更新/刪除，是如何做到的呢？
與HBase類似，是通過增加一條新的記錄來描述這次更新/刪除操作的。一個DiskRowSet包含兩部分資料：基礎資料(Base Data)，以及變更資料(Delta Stores)。更新/刪除操作所生成的資料記錄，被儲存在變更資料部分。
圖源自Kudu的源工程檔案
從上圖來看，Delta資料部分應該包含REDO與UNDO兩部分，這裡的REDO與UNDO與關係型資料庫中的REDO與UNDO日誌類似（在關係型資料庫中，REDO日誌記錄了更新後的資料，可以用來恢復尚未寫入Data File的已成功事務更新的資料。 而UNDO日誌用來記錄事務更新之前的資料，可以用來在事務失敗時進行回滾），但也存在一些細節上的差異：
REDO Delta Files包含了Base Data自上一次被Flush/Compaction之後的變更值。
REDO Delta Files按照Timestamp順序排列，UNDO Delta Files包含了Base Data自上一次Flush/Compaction之前的變更值。這樣才可以保障基於一箇舊Timestamp的查詢能夠看到一個一致性檢視。
UNDO按照Timestamp倒序排列。
3.5、Kudu 資料讀寫流程
3.5.1、Kudu 資料寫流程
Kudu不允許使用者資料的Primary Key重複，因此，在Tablet內部寫入資料之前，需要先從已有的資料中檢查當前新寫入的資料的Primary Key是否已經存在，儘管在DiskRowSets中增加了BloomFilter來提升這種判斷的效率，但可以預見，Kudu的這種設計將會明顯增大寫入的時延。
資料一開始先存放於MemRowSets中，待大小超出一定的閾值之後，再Flush成DiskRowSets。隨著Flush次數的不斷增加，生成的DiskRowSets也會不斷的增多，在Kudu內部也存在一個Compaction流程，這樣可以將已經存在的多個 Primary Key 交集的DiskRowSets重新排序而生成一個新的DiskRowSets。如下圖所示：
3.5.2、Kudu 資料讀流程
讀資料的流程，既要考慮存在於記憶體中的MemRowSets,又要讀取位於磁碟中的一個或多個DiskRowSets，在Scanner的高層抽象中，應該與HBase類似。
重點說一些細節的優化點：
1、通過Scan的範圍，與每一個DiskRowSets中的Primary Key Range進行對比，可以首先過濾掉一些不必要參與此次Scan的DiskRowSets。
2、Delta Store部分，針對記錄級別的更改，記錄了Base Data中對應原始資料的Offset。這樣，在判斷一條記錄是否存在更改的記錄時，將會更加的快速。
3、由於DiskRowSets的底層檔案是按照列組織的，基於一些列的條件進行過濾查詢時，可以優先過濾掉一些不必要的Primary Keys。Kudu並不會在一開始讀取的時候就將一行資料的所有列讀取出來，而是先讀取與過濾條件相關的列，通過將這些列與查詢條件匹配之後，再來決定是否去讀取符合條件的行中的其它的列資訊。這樣可以節省一些磁碟IO。這就是Kudu所提供的Lazy Materialization特性。
##### 3.6、Raft模型
Kudu的多副本之間的資料共識協議採用了Raft協議，Raft是比Paxos更容易理解且更簡單的一種一致性協議。
關於Raft的更多資訊，請參考：https://raft.github.io/
4、Kudu 與 HBase 區別
Tablet與HBase中的Region大致相似，但存在如下一些明顯的區別點：
Tablet包含兩種分割槽策略，一種是基於Hash Partition方式，在這種分割槽方式下使用者資料可較均勻的分佈在各個Tablet中，但原來的資料排序特點已被打亂。
另外一種是基於Range Partition方式，資料將按照使用者資料指定的有序的Primary Key Columns的組合String的順序進行分割槽。而HBase中僅僅提供了一種按使用者資料RowKey的Range Partition方式。
一個Tablet可以被部署到了多個Tablet Server中。
在HBase最初的架構中，一個Region只能被部署在一個RegionServer中，它的資料多副本交由HDFS來保障。從1.0版本開始，HBase有了Region Replica（HBASE-10070）特性，該特性允許將一個Region部署在多個RegionServer中來提升讀取的可用性，但多Region副本之間的資料卻不是實時同步的。
Kudu的資料多副本機制
HBase的資料多副本機制
1、Kudu的資料分割槽方式相對多樣化，而HBase較單一。
2、Kudu的Tablet自身具備多副本機制，而HBase的Region依賴於底層HDFS的多副本機制。
3、Kudu底層直接採用本地檔案系統， 而HBase依賴於HDFS。
4、Kudu的底層檔案格式採用了類似於Parquet的列式儲存格式，而HBase的底層HFile檔案卻是按行來組織的。
5、Kudu關於底層的Flush任務以及Compaction任務，能夠結合忙時或者閒時進行自動的調整。HBase還尚不具備這種排程能力。
6、Kudu的Compaction無Minor/Major的區分，限制每一次Compaction的IO總量在128MB大小，因此，並不存在長久執行的Compaction任務。 Compaction是按需進行的，例如，如果所有的寫入都是順序寫入，則將不會觸發Compaction。
7、Kudu的設計，既兼顧了分析型的查詢能力，又兼顧了隨機讀寫能力，這樣，勢必也會付出一些代價。 例如，寫入資料時關於Primary Key唯一性的限制，就要求寫入前要檢查對應的Primary Key是否已經存在，這樣勢必會增大寫入的時延。而底層儘管採用了類似於Parquet的列式檔案設計，但與HBase類似的冗長的讀取路徑，也會對分析性的查詢帶來一些影響。另外，這種設計在整行讀取時，也會付出較高的代價。
5、Kudu的適用場景
5.1、具有近實時可用性的流輸入
資料分析中的一個共同挑戰是新資料快速且持續地到達，並且需要幾乎實時地提供相同的資料以進行讀取，掃描和更新。 Kudu提供快速插入和更新的強大組合，以及高效的柱狀掃描，以在單個儲存層上實現實時分析用例。
5.2、具有各種訪問模式的時間序列應用程式
時間序列模式是根據資料點發生的時間組織和鍵入資料點的模式。 這對於調查指標隨時間的效能或嘗試根據過去的資料預測未來行為非常有用。 例如，時間序列客戶資料既可用於儲存購買點選流歷史記錄，也可用於預測未來購買或供客戶支援代表使用。 雖然這些不同型別的分析正在發生，但插入和突變也可能單獨和大量發生，並且可立即用於讀取工作負載。 Kudu可以以可擴充套件且高效的方式同時處理所有這些訪問模式。
由於多種原因，Kudu非常適合時間序列工作負載。 由於Kudu支援基於雜湊的分割槽，結合其對複合行金鑰的原生支援，可以很容易地設定跨多個伺服器的表，而不存在使用範圍分割槽時常見的“熱點”風險。 Kudu的柱狀儲存引擎在這種情況下也很有用，因為許多時間序列工作負載只讀取幾列，而不是整行。
過去，您可能需要使用多個數據儲存來處理不同的資料訪問模式。 這種做法增加了應用程式和操作的複雜性，並使資料重複，使所需的儲存量翻倍（或更差）。 Kudu可以本地高效地處理所有這些訪問模式，而無需將工作解除安裝到其他資料儲存。
5.3、預測建模
資料科學家經常從大量資料開發預測性學習模型。 當學習發生或建模的情況發生變化時，可能需要經常更新或修改模型和資料。 此外，科學家可能想要改變模型中的一個或多個因素，看看隨著時間的推移會發生什麼。 更新儲存在HDFS中的檔案中的大量資料是資源密集型的，因為每個檔案都需要完全重寫。 在Kudu，更新幾乎是實時發生的。 科學家可以調整值，重新執行查詢，並在幾秒或幾分鐘內重新整理圖表，而不是幾小時或幾天。 此外，批處理或增量演算法可以隨時跨資料執行，並具有接近實時的結果。
5.4、將Kudu中的資料與遺留系統相結合
公司從多個來源生成資料並將其儲存在各種系統和格式中。 例如，您的一些資料可能儲存在Kudu中，一些儲存在傳統的RDBMS中，一些儲存在HDFS中的檔案中。 您可以使用Impala訪問和查詢所有這些源和格式，而無需更改舊系統。
6、Spark 與 Kudu 整合已知問題和限制
- 儘管Kudu Spark 2.x整合與Java 7相容，但Spark 2.2+ 以後在執行時需要Java 8。Spark 2.2 是Kudu 1.5.0 的預設依賴版本。
- 當註冊為臨時表時，名稱必須為包含大寫或非ascii字元
- 包含大寫或非ascii字元的列名的Kudu表不能與SparkSQL一起使用。 可以在Kudu中重新命名列以解決此問題。
- <>和OR謂詞不會被推送到Kudu，而是由Spark任務評估。
- 只有具有後綴萬用字元的LIKE謂詞被推送到Kudu，這意味著LIKE“FOO％”會被推，但LIKE“FOO％BAR”則不會。
- Kudu不支援Spark SQL支援的每種型別。 例如，不支援日期和複雜型別。
- Kudu表只能在SparkSQL中註冊為臨時表。 使用HiveContext可能無法查詢Kudu表。
7、效能測試
7.1、Kudu 與 Parquet 對比
圖是官方給出的用 Impala 跑 TPC-H 的測試，對比 Parquet 和 Kudu 的計算速度。從圖中我們可以發現， Kudu 的速度和 parquet 的速度差距不大，甚至有些 Query 比 parquet 還快。然而，由於這些資料都是在記憶體快取過的，因此該測試結果不具備參考價值。
7.2、Kudu 與 Hbase 對比
圖是官方給出的另一組測試結果，從圖中我們可以看出，在 scan 和 range 查詢上， kudu 和 parquet 比 HBase 快很多，而 random access 則比 HBase 稍慢。然而資料集只有 60 億行資料，所以很可能這些資料也是可以全部快取在記憶體的。對於從記憶體查詢，除了 random access 比 HBase 慢之外， kudu 的速度基本要優於 HBase
7.3. 超大資料集的查詢效能
Kudu 的定位不是 in-memory database 。因為它希望 HDFS/Parquet 這種儲存，因此大量的資料都是儲存在磁碟上。如果我們想要拿它代替 HDFS/Parquet + HBase ，那麼超大資料集的查詢效能就至關重要，這也是 Kudu 的最初目的。然而，官方沒有給出這方面的相關資料。
Kudu 參考資料
Documentation：官方文件永遠是學習開源專案的最好去處。
Paper：Kudu的論文可以幫助您深入瞭解Kudu的設計思想。
Raft協議：Kudu的一致性協議使用了Raft協議，瞭解Raft協議可以幫助您更好地瞭解Kudu及其他分散式開源系統。