Apache Kylin on Apache HBase 方案經過長時間的發展已經比較成熟，但是存在著一定的局限性。因此，Kyligence 推出了 Kylin on Parquet 方案。本文中，Kyligence 的大數據研發工程師王汝鵬講解了該解決方案的架構、原理以及如何開發調試代碼。 ...
Apache Kylin on Apache HBase 方案經過長時間的發展已經比較成熟，但是存在著一定的局限性。Kylin 查詢節點當前主要的計算是在單機節點完成的，存在單點問題。而且由於 HBase 非真正列存的問題，Cuboids 信息需要壓縮編碼，讀取 HBase 數據的時候再反序列化、分割，額外增加了計算壓力。另外，HBase 運維難度比較大，不便於上雲。面對以上問題，Kyligence 推出了 Kylin on Parquet 方案。下文中，Kyligence 的大數據研發工程師王汝鵬講解了 Kylin on Parquet 解決方案的架構、原理以及如何開發調試代碼。
本文主要包括以下幾方面的內容：首先會給大家介紹架構設計，然後說明一下我們為什麼會去做 Kylin on Parquet，接下來會介紹一下全新的構建和查詢引擎以及相比較於 Kylin 3.0 的性能表現，最後有一個現場演示 Demo，給大家介紹一下產品的使用和代碼調試方法。
01
架構
Apache Kylin 很早就被設計成了可插拔的架構，基於這種架構我們就可以很方便的去替換某個模塊而不會影響其他模塊。
Kylin on Parquet 也是在 Kylin 原來架構的基礎上實現了新的查詢、構建引擎和存儲模塊。通過 Spark 實現的查詢引擎，能夠提交計算任務到 Yarn 上，實現分散式的處理。
Cube 構建這邊也是完全通過 Spark 進行處理，不再支持 MapReduce 構建。
數據源現在支持 Hive 和本地 CSV 數據源，目前可以擺脫沙箱的限制，通過本地的 CSV 數據源搭建一個調試環境。
存儲層去掉了 HBase，最終構建完成的 Cube 數據都是通過 Parquet 的形式直接存儲在文件系統中。
02
為什麼是 Kylin on Parquet?
首先，原來 Kylin 依賴 HBase 的架構在查詢的時候會存在單點問題，因為一次查詢任務在通過 Coprocessor 獲取到數據之後的處理是在查詢結點單機上完成的。
HBase 不是一個真正的列式存儲，它通過 RowKey 來保留每一行的數據，之所以稱之為“列式”，是因為它通過列族的結構管理列數據，何為真正列式存儲，可以通過下麵文章瞭解更多：https://en.wikipedia.org/wiki/Column-oriented_DBMS。
我們可以看到下麵Cube邏輯視圖中，Kylin 3.0 及以前對於 Cube 是通過將所有的維度和度量分別壓縮成一列進行存儲的，這樣在查詢的時候還需要對這一列進行反序列化、分割等操作，額外增加了計算壓力。
最後，HBase 比較難於維護，運維難度比較高。
查詢過程主要就是 Calcite 會將 SQL 解析成一棵物理執行計劃樹，其中的計算邏輯的代碼都是通過 Calcite 生成的，這些代碼會比較難於調試和定位問題。
Kylin on Parquet 目前能夠通過 Spark 進行分散式的查詢，我們對 Calcite 生成的執行計劃做了一層轉換，轉換成了 Spark 的執行計劃，其中每一層的處理的數據我們都是能夠通過添加斷點查看的。
現在查詢相關的邏輯代碼也是比較方便調試的，比如我們懷疑在聚合(Agg)這一層出了問題，我們就可以在 Agg 這一步添加斷點，查看一下數據是不是符合我們的期望。
存儲這邊我們替換成了 Parquet，所有的維度和度量會按照每一列進行存儲，後面對於存儲的結構也會有更加詳細的介紹。
03
Cube 構建與查詢
1. 構建引擎
接下來給大家介紹一下全新的構建引擎以及其中的功能是怎麼實現的。
1）關鍵特性
以下是關鍵的特性：
- 構建引擎完全的通過 Spark 進行處理，中間的所有流程都能夠在 SparkUI 上監控到。如果構建過程出現了問題，也能夠在 SparkUI 上查看任務的執行情況。
- 構建引擎加入了自動調參的功能，這個主要是針對用戶沒有手動去配置 Spark 參數的情況下，根據構建任務量的情況去調整 Spark 相關的參數，這樣能更高效地去執行任務。
- 構建引擎實現了全局字典的分散式構建。
- 加入了自動恢復失敗任務的功能，當任務失敗之後，構建引擎會分析當前任務失敗的原因， 然後根據不同失敗的情況執行不同處理的策略。
2）介面設計
分享的開頭裡，我提到了 Kylin 可插拔式的架構設計，所以上層實現的介面從 AbstractExecutable 到 CubingJob 都是 Kylin 原有的介面，通過調用 SparkCubingJob 的 create 方法可以提交一個構建 Segment 的任務，然後接下來我們抽象出來了兩個步驟，一是資源探測，二是構建 Cube。這兩步後面也會進行更加詳細的介紹。最後，這兩步會串聯起來通過 Spark 任務的方式提交到集群或者本地去執行。
3）步驟
構建步驟包括資源探測和 Cube 構建。資源探測主要做了三件事，首先它會去估算一下當前數據源表的大小，這裡也是為了接下來第二步自動調參準備的，第三點是構建全局字典。
Cube 構建這一步其實和原來的構建引擎整體步驟是差不多的，首先會通過 Spark 創建平表，然後逐層地構建 Cube，接下來通過 Parquet 的形式進行存儲，最後再更新一下 Metadata。為什麼我們會把這麼多處理集合成一個步驟，主要是因為數據主要是通過 Spark 在記憶體中進行處理，如果再拆分成多步，還需要對中間數據進行持久化等操作，這樣處理效率就會打折扣。右圖是構建任務在前端的執行情況。
4）自動調參
自動調參功能預設是打開的，並且只在集群模式下生效，而且手動配置的優先順序要高於自動調整。它會根據數據源的大小等情況，估算一下當前構建任務需要的計算資源，最終調整 Spark 任務中 executor 相關的參數。
5）全局字典
全局字典功能相對於 Kylin 3.0 主要有兩點提升：能夠分散式地處理；不再局限於整數類型最大值的限制。其實當前 Kylin 3.0 是新加入了分散式構建字典的功能的，不過預設還是單機構建的方式。
具體步驟如下：
- 通過 Spark 創建平表和獲取對應列的 distinct 值
- 將數據分配到多個桶中
- 對每一個桶內的數據進行編碼
- 保存字典文件和 metadata 數據(桶數量和桶的 offset 值)
第一次構建字典的時候會對每個桶內的值從 1 開始編碼，在編碼完成後再根據每個桶的 offset 值進行一次整體字典值的分配。
第二次提交 Segment 構建任務的時候，會對每個桶的值進行一次再分配，相對於桶內已有值進行編碼，然後根據新的 offset 去更新每個桶內相對於全局的一個字典值。
磁碟上保存的目錄結構如圖所示。
6）自動重試
自動重試功能會分析導致構建任務失敗的異常或錯誤，並分別採取不同的處理策略。
- 當遇到 OutOfMemoryError 的時候，引擎會檢查當前 Spark 任務是否開啟了 AUTO_BROADCASTJOIN_THRESHOLD 這個參數，這個功能比較容易導致Spark任務出現記憶體不足的報錯，嘗試禁用這個功能，然後重新提交構建任務。
- 如果遇到的是 ClassNotFoundException，構建引擎會直接終止當前任務並拋出異常。
- 對於其他異常，構建引擎會嘗試調整 executor core 的數量和分配記憶體大小，然後重新提交任務。
此功能的預設重試次數為三次，而且是預設打開的，如果想禁用此功能，可以將 kylin.engine.max-retry-time 設置為 0 或者如任意負數。
7）度量
構建過程對所有的度量都是會做處理的，具體處理邏輯可以在 CuboidAggregator.scala 文件中查看。由於現在查詢引擎還存在一些相容性的問題，TopN, CountDistinct, Percentile 現在還查不了，但是已經有 issue 在做了。
8）存儲
假設我們最終生成的 cuboid 內容如上圖所示，存在三個維度和兩個度量，對應的 parquet 文件的 schema 就是中間這張圖的樣子。我們會將所維度名稱映射成一個唯一的數字，這樣也是為了進一步優化存儲。我們可以將 parquet 文件下載到本地，通過 spark 看到當前 parquet 文件，也就是我們保存的 cuboid 文件的 schema 內容。
磁碟上存儲的目錄結構如上圖所示，所有文件是通過項目來歸類的，包括字典，構建產生的臨時文件以及構建完成的所有 cuboids。Segment 目錄會有一個獨立的簽名，防止出現寫入衝突等問題。
9）性能對比
我們將新的構建引擎和 Kylin 3.0 的構建引擎（MapReduce）做了一下對比，運行環境是擁有四個計算節點，Yarn 擁有 400G 記憶體和 128 內核的集群。Spark使用的內部版本，由於我們對 Spark 源碼做了一些優化，所以目前並不支持社區版 Spark。測試的數據集是標準的 SSB 數據集。
左邊是最終占用存儲空間的大小，新構建引擎存儲空間占用能夠減少一半。右邊是構建時間的對比，也能夠看到新構建引擎也比 Kylin 3.0 快了許多。
2. 查詢引擎
1）步驟
一次查詢的請求發出後，Calcite 會分析 SQL 並解析成抽象語法樹(AST)，然後對 AST 進行校驗、優化等操作後，再轉換成執行計劃樹(RelNodes)。新查詢引擎會將所有的 RelNodes 轉換成 Spark 執行計劃。最後再通過 Spark 去執行所有的查詢任務。
查詢引擎會把每一個計算邏輯轉換成對應的 Spark 邏輯。轉換的這一步其實也做了不少工作，因為 Calcite 有自己的類型，Spark 也有自己的類型，我們需要對其進行處理。Calcite 的一些函數操作也需要做一些對應的實現。
開始的時候也說過了，我們可以在每一個 DataFrame 中添加斷點去進行調試，查詢中間處理的值，這樣能夠更加方便的排查問題。查詢引擎會在第一次收到查詢請求的時候在 Yarn 上創建一個常駐進程，專門用來處理查詢任務。
針對查詢引擎還做了依賴隔離的處理，主要防止外部依賴類衝突的問題。
2）性能對比
查詢引擎的性能表現也是和 Kylin 3.0 做了一下對比，測試環境和構建性能測試環境是一樣的，這裡就不贅述了。我們對 SSB 數據集和 TPCH 數據集都做了對比。
SSB 數據集規模大概有六千萬行，不過 SSB 的標準 SQL 大都比較簡單，所有我們看到查詢基本上都是一秒內完成的。
TPCH 數據集規模大概有一千兩百萬行，TPCH 的標準 SQL 要求更高一些，我們可以看到 Kylin3.0 耗時非常長的查詢任務，新的構建引擎的查詢能夠快很多，因為我們對複雜的查詢做了一些優化。
04
Demo
請點擊播放下方現場回顧視頻，拖動進度條至 26:35 的位置，即可開始觀看。
05
規劃
06
如何體驗與貢獻
最後也歡迎大家加入我們，目前 Kylin on Parquet 也已經開源出來，對應的文檔在 Github 倉庫的 wiki 頁面也都能看到。大家有問題也可以去 JIRA 上提出來，我們後期會進行修複。最後為了方便大家討論也可以加一下上圖的微信群。
