Kylin 2.0 Spark Cubing 優化改進

Kylin 2.0 引入了Spark Cubing beta版本，本文主要介紹我是如何讓 Spark Cubing 支持 啓用Kerberos的HBase集羣，再介紹下Spark Cubing的性能測試結果和適用場景。

Spark Cubing 簡介

在簡介Spark Cubing以前，我簡介下MapReduce Batch Cubing。所謂的MapReduce Batch Cubing就是利用MapReduce 計算引擎 批量計算Cube，其輸入是Hive表，輸出是HBase的KeyValue，整個構建過程主要包含如下6步：

1. 創建Hive的大寬表； （MapReduce計算）
2. 對須要字典編碼的列計算列基數； （MapReduce計算）
3. 構建字典； （JobServer計算 or MapReduce計算）
4. 分層構建Cuboid； （MapReduce計算）
5. 將Cuboid轉爲HBase的KeyValue結構（HFile）； （MapReduce計算）
6. 元數據更新和垃圾回收。

詳細的Cube生成過程能夠參考 Apache Kylin Cube 構建原理。

而Kylin 2.0的Spark Cubing就是在Cube構建的第4步替換掉MapReduce。

以下圖，就是將5個MR job轉換爲1個Spark job：

（注：如下兩個圖片引自 Apache Kylin 官網的blog： By-layer Spark Cubing， 更詳細的介紹也能夠參考這篇blog。）

MapReduce 計算5層的Cuboid會用5個MR Application計算： 

Spark 計算Cuboid只會用1個 Application計算： 

Spark Cubing的核心實現類是SparkCubingByLayer。

爲何目前只有計算Cuboid這一步用Spark計算？

我的認爲主要有如下兩點：

1. 實現並不複雜。
2. 收益會比較明顯。Cuboid的分層計算方法自然能夠利用RDD的Cache特性來加速Cuboid 計算，最理想的狀況，若是executor的內存可以徹底cache住每層Cuboid的RDD，那就能夠徹底避免讀寫磁盤，必然會比MapReduce快不少。

我認爲第2點是主要緣由。

Cube構建的其餘步驟不能夠用Spark計算嗎？

固然能夠！ 其中第1步 創建Hive的大寬表 和 第5步 生成HFile 替換爲Spark是十分簡單的，可是性能提高可能不會十分明顯。 至於2步計算列基數，其代碼邏輯應該是整個Cube構建中最複雜的一步，複雜的主要緣由就是這一步肩負的使命略多。 還有第3步MR構建字典，由於MR構建自己尚不成熟，天然不急着遷移到Spark。

Spark Cubing beta版本目前最大的問題就是不支持啓用Kerberos認證的HBase集羣，而事實上很多企業級的HBase服務都啓用了Kerberos認證。不支持的緣由主要是Spark Cubing須要直接從HBase中訪問cube，dict等元數據信息。

Spark Cubing 訪問 Kerberos認證的 HBase 解1

第一種簡單的作法是將訪問HBase的token從Kylin的JobServer傳遞到executor中，這種作法的限制是只能運行在Yarn-client模式中，即必須讓driver運行在Kylin的JobServer中。 關於yarn-cluster mode和yarn-client mode兩種模式的區別能夠參考： Apache Spark Resource Management and YARN App Models。

這種作法的實現方式很簡單，只需在SparkCubingByLayer的new SparkConf()以前加入如下3行代碼：

Configuration configuration = HBaseConnection.getCurrentHBaseConfiguration();
        HConnection connection = HConnectionManager.createConnection(configuration);
        TokenUtil.obtainAndCacheToken(connection, UserProvider.instantiate(configuration).create(UserGroupInformation.getCurrentUser()));

可是若是隻能在yarn-client模式下運行，必然沒法運行在生產環境，由於Kylin JobServer機器的內存確定不夠用。

Spark Cubing 訪問 Kerberos認證的 HBase 解2

既然Spark Cubing在 啓用Kerberos認證的 HBase集羣下沒法運行的根本緣由是 Spark Cubing須要從HBase 直接訪問Job相關的Kylin元數據， 那咱們把元數據換個地方存不就能夠了， 因此咱們將每一個Spark Job相關的Kylin元數據上傳到HDFS，並用Kylin的HDFSResourceStore來管理元數據。

在介紹實現思路前，我先簡介下Kylin元數據的存儲結構和Kylin的ResourceStore。

首先，Kylin每一個具體的元數據都是一個JSON文件，整個元數據的組織結構是個樹狀的文件目錄。 如圖是Kylin元數據的根目錄： 下圖是project目錄下的具體內容，其中learn_kylin和kylin_test都是project名稱： 

咱們知道Kylin元數據的組織結構後，再簡介下Kylin元數據的存儲方式。 元數據存儲的抽象類是ResourceStore，具體的實現類共有3個：

• FileResourceStore 本地文件系統
• HBaseResourceStore HBase
• HDFSResourceStore HDFS

其中只有HBase能夠用於生產環境，本地文件系統主要用來測試，HDFS不能用於生產的緣由是併發處理方面還有些問題。具體用哪一個ResourceStore是經過配置文件的kylin.metadata.url來決定的。

因此下面的問題就是咱們如何將HBase中的元數據轉移到HDFS 和如何將HBaseResourceStore 轉爲 HDFSResourceStore？

1. 肯定Spark Job須要讀取哪些Kylin元數據
2. 將須要的Kylin元數據dump到本地
3. 改寫kylin.metadata.url並將全部配置寫到本地的元數據目錄
4. 利用ResourceTool將本地的元數據上傳到指定的HDFS目錄
5. 在Spark executor中根據指定HDFS元數據目錄的Kylin配置文件構造出HDFSResourceStore。

固然，在最後咱們須要清理掉指定HDFS目錄的元數據。 整個思路比較簡單清晰，可是實際實現中仍是有不少細節須要去考慮。

Spark Cubing 參數配置

如下是我使用的Spark配置，目的是儘量讓用戶不須要關心Spark的配置

//運行在yarn-cluster模式
kylin.engine.spark-conf.spark.master=yarn
kylin.engine.spark-conf.spark.submit.deployMode=cluster 

//啓動動態資源分配，我的認爲在Kylin生產場景中是必須的，由於咱們不可能讓每一個用戶本身去指定executor的個數
kylin.engine.spark-conf.spark.dynamicAllocation.enabled=true
kylin.engine.spark-conf.spark.dynamicAllocation.minExecutors=10
kylin.engine.spark-conf.spark.dynamicAllocation.maxExecutors=1024
kylin.engine.spark-conf.spark.dynamicAllocation.executorIdleTimeout=300
kylin.engine.spark-conf.spark.shuffle.service.enabled=true
kylin.engine.spark-conf.spark.shuffle.service.port=7337

//內存設置
kylin.engine.spark-conf.spark.driver.memory=4G
//數據規模較大或者字典較大時能夠調大executor內存
kylin.engine.spark-conf.spark.executor.memory=4G 
kylin.engine.spark-conf.spark.executor.cores=1

//心跳超時
kylin.engine.spark-conf.spark.network.timeout=600

//隊列設置
kylin.engine.spark-conf.spark.yarn.queue=root.rz.hadoop-hdp.test

//分區大小
kylin.engine.spark.rdd-partition-cut-mb=100

Spark Cubing 的構建性能

對於百萬級，千萬級，億級的源數據，且無很大字典的狀況下，個人測試結果和官方By-layer Spark Cubing 的結果基本一致，構建速度提高比較明顯，並且Cuboid的層次數越多，提速越明顯。

此外，我專門測試了數十億級源數據或者有超大字典的狀況，構建提速也十分明顯：

測試Cube1

原始數據量： 27億行 9個維度 包含1個精確去重指標 字典基數7千多萬

MR Cuboid構建耗時： 75分鐘

Spark Cuboid第一次構建耗時： 40分鐘 （spark.executor.memory = 8G，沒有加spark.memory.fraction參數）

Spark Cuboid第二次構建耗時： 24分鐘 （spark.executor.memory = 8G，spark.memory.fraction = 0.5）

爲何減少spark.memory.fraction能夠加速構建？

由於減少spark.memory.fraction，能夠增大executor中User Memory的大小，給Kylin字典更多的內存，這樣就能夠避免全局字典換入換出，減小GC。

測試Cube2

原始數據量：24億行 13個維度 38個指標（其中9個精確去重指標） 不過這個cube的精確去重指標基數比較小，只有幾百萬。

MR Cuboid構建耗時： 31分鐘

Spark Cuboid構建耗時： 8分鐘

總結來講，Spark Cubing的構建性能相比MR有1倍到3倍的提高。

Spark Cubing 的資源消耗

除了構建性能，咱們確定還會關注資源消耗。在此次測試中我沒有對因此測試結果進行資源消耗分析，只分析了幾個Cube。

個人結論是，在我採用的Spark配置狀況下，對於中小規模數據集Spark的資源消耗是小於MR的（executor的內存是4G）； 對於有大字典的狀況（executor的內存是8G），CPU資源Spark是小於MR的，可是內存資源Spark會比MR略多，在這種狀況下，咱們至關於用內存資源來換取了執行效率。

Spark Cubing 的優缺點

優勢：

• 利用RDD的Cache特性，儘量利用內存來避免重複IO
• 大部分場景下Cuboid構建速度有明顯提高
• 在集羣資源充足的狀況下，咱們能夠用更多的資源換取更好的構建性能

缺點：

• 目前版本還未歷經生產環境考驗，穩定性不肯定
• 不適合有超大字典的場景
• 引入Spark Cubing將帶來額外的運維成本和溝通成本

Spark Cubing 的適用場景

我的的結論是，除了有好幾億基數超大字典的這種狀況，其餘狀況應該都適用Spark Cubing，其中：

1. Cuboid層次越多越適用。
2. 數據規模越小越適用。
3. 字典越小越適用。

Spark Cubing 字典加載優化

Spark和MR有一點重要的區別就是Spark的Task是在線程中執行的，MR的Task是在進程中執行的。 這點區別會對Kylin的Cube 構建形成重要影響，在MR Cubing中，每一個Mapper task 只須要load一次字典，可是在Spark Cubing中，一個executor的多個task會屢次load 字典，若是字典較大，就會形成頻繁GC，致使執行變慢。

針對這個問題，我作了兩點優化：

1. 讓每一個executor裏的字典只load一次，讓該executor的全部Task共享字典。
2. 給全局字典的AppendTrieDictionary中使用的LoadingCache增長maximumSize。 我用了一個有6億基數的全局字典測試了這個優化，優化後GC時間明顯縮短。

Spark 學習資料推薦

網上公開資料若是隻推薦一份的話，我推薦： spark-internals

此外，這幾篇文章也不錯：

Spark Memory Management

Spark Architecture: Shuffle

how-to-tune-your-apache-spark-jobs-part-1

how-to-tune-your-apache-spark-jobs-part-2

Spark性能優化指南——基礎篇

Spark性能優化指南——高級篇

Spark源碼分析之-scheduler模塊

固然，看的資料再多本身不思考都沒啥卵用。 學習一個系統，咱們能夠從系統的總體架構和設計層面開始，自頂向下的學習，也能夠從一個具體的問題把整個系統涉及的全部模塊串起來，切面式學習。 我的感受兩種方式結合着效率會比較高，並且通常從具體問題入手會讓印象更深入，理解更深刻。