Kylin on Parquet 介紹和快速上手

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 上提出來，我們后期會進行修復，最后為了方便大家討論也可以加一下上圖的微信群，

了解更多大資料資訊，點擊進入Kyligence官網

標籤：大數據

上一篇：如何在 HBase Shell 命令列正常查看十六進制編碼的中文？哈哈~

下一篇：字串相似度處理函式