Yige

數倉系列 - Kylin 概覽#

Kylin 的特點#

Kylin 的主要特點包括支持 SQL 接口、支持超大規模數據集、亞秒級響應、可伸縮性、高吞吐率、BI 工具集成等。

1. 標準SQL接口：Kylin 是以標準的 SQL 作為對外服務的接口。
2. 支持超大數據集：Kylin 對於大數據的支撐能力可能是目前所有技術中最為領先的。早在 2015 年 eBay 的生產環境中就能支持百億記錄的秒級查詢，之後在移動的應用場景中又有了千億記錄秒級查詢的案例。
3. 亞秒級響應：Kylin 擁有優異的查詢相應速度，這點得益於預計算，很多複雜的計算，比如連接、聚合，在離線的預計算過程中就已經完成，這大大降低了查詢時刻所需的計算量，提高了響應速度。
4. 可伸縮性和高吞吐率：單節點 Kylin 可實現每秒 70 個查詢，還可以搭建 Kylin 的集群。
5. BI工具集成： Kylin 可以與現有的 BI 工具集成，具體包括如下內容。
   1. ODBC：與 Tableau、Excel、PowerBI 等工具集成
   2. JDBC：與 Saiku、BIRT 等 Java 工具集成
   3. RestAPI：與 JavaScript、Web 網頁集成

Kylin 的基本架構#

REST Server
REST Server 是一套面向應用程序開發的入口點，旨在實現針對 Kylin 平台的應用開發工作。 此類應用程序可以提供查詢、獲取結果、觸發 cube 構建任務、獲取元數據以及獲取用戶權限等等。另外可以通過 Restful 接口實現 SQL 查詢。

查詢引擎（Query Engine）
當 cube 準備就緒後，查詢引擎就能夠獲取並解析用戶查詢。它隨後會與系統中的其它組件進行交互，從而向用戶返回對應的結果。

Routing
負責將解析的 SQL 生成的執行計劃轉換成 cube 緩存的查詢，cube 是通過預計算緩存在 HBase 中，這部分查詢可以在秒級設置毫秒級完成

元數據管理工具（Metadata）
Kylin 是一款元數據驅動型應用程序。元數據管理工具是一大關鍵性組件，用於對保存在 Kylin 當中的所有元數據進行管理，其中包括最為重要的 cube 元數據。其它全部組件的正常運作都需以元數據管理工具為基礎。 Kylin 的元數據存儲在 HBase 中。

任務引擎（Cube Build Engine）
這套引擎的設計目的在於處理所有離線任務，其中包括 shell 腳本、Java API 以及 Map Reduce 任務等等。任務引擎對 Kylin 當中的全部任務加以管理與協調，從而確保每一項任務都能得到切實執行並解決其間出現的故障。

Kylin 工作原理#

Apache Kylin 的工作原理本質上是MOLAP（Multidimension On-Line Analysis Processing）Cube，也就是多維立方體分析，詳細概念參考: 數倉系列 - 基本概念整理

核心算法#

對數據模型做 Cube 預計算，並利用計算的結果加速查詢：

• 指定數據模型，定義維度和度量；

• 預計算 Cube，計算所有 Cuboid 並保存為物化視圖；
  預計算過程是 Kylin 從 Hive 中讀取原始數據，按照我們選定的維度進行計算，並將結果集保存到 HBase 中，默認的計算引擎為 MapReduce，可以選擇 Spark 作為計算引擎。一次 build 的結果，我們稱為一個 Segment。構建過程中會涉及多個 Cuboid 的創建，具體創建過程由 kylin.cube.algorithm 參數決定，參數值可選 auto，layer 和 inmem， 默認值為 auto，即 Kylin 會通過採集數據動態地選擇一個算法 (layer or inmem)，如果用戶很了解 Kylin 和自身的數據、集群，可以直接設置喜歡的算法。

• 執行查詢，讀取 Cuboid，運行，產生查詢結果

layer (逐層構建) 算法#

一個 N 維的 Cube，是由 1 個 N 維子立方體、N 個 (N-1) 維子立方體、N*(N-1)/2 個 (N-2) 維子立方體、......、N 個 1 維子立方體和 1 個 0 維子立方體構成，總共有 2^N 個子立方體組成，在逐層算法中，按維度數逐層減少來計算，每個層級的計算（除了第一層，它是從原始數據聚合而來），是基於它上一層級的結果來計算的。比如，[Group by A, B] 的結果，可以基於 [Group by A, B, C] 的結果，通過去掉 C 後聚合得來；這樣可以減少重複計算；當 0 維度 Cuboid 計算出來的時候，整個 Cube 的計算也就完成了。

注意:
每一輪的計算都是一個 MapReduce 任務，且串行執行；一個 N 維的 Cube，至少需要 N 次 MapReduce Job

算法優點：

1. 此算法充分利用了 MapReduce 的能力，處理了中間複雜的排序和洗牌工作，故而算法代碼清晰簡單，易於維護；
2. 受益於 Hadoop 的日趨成熟，此算法對集群要求低，運行穩定；在內部維護 Kylin 的過程中，很少遇到在這幾步出錯的情況；即便是在 Hadoop 集群比較繁忙的時候，任務也能完成。

算法缺點：

1. 當 Cube 有比較多維度的時候，所需要的 MapReduce 任務也相應增加；由於 Hadoop 的任務調度需要耗費額外資源，特別是集群較龐大的時候，反復遞交任務造成的額外開銷會相當可觀；
2. 由於 Mapper 不做預聚合，此算法會對 Hadoop MapReduce 輸出較多數據；雖然已經使用了 Combiner 來減少從 Mapper 端到 Reducer 端的數據傳輸，所有數據依然需要通過 Hadoop MapReduce 來排序和組合才能被聚合，無形之中增加了集群的壓力；
3. 對 HDFS 的讀寫操作較多：由於每一層計算的輸出會用做下一層計算的輸入，這些 Key-Value 需要寫到 HDFS 上；當所有計算都完成後，Kylin 還需要額外的一輪任務將這些文件轉成 HBase 的 HFile 格式，以導入到 HBase 中去；

總體而言，該算法的效率較低，尤其是當 Cube 維度數較大的時候。

inmem (快速構建) 算法#

也被稱作 “逐段”(By Segment) 或 “逐塊”(By Split) 算法，從1.5.x開始引入該算法，利用 Mapper 端計算先完成大部分聚合，再將聚合後的結果交給 Reducer，從而降低對網絡瓶頸的壓力。該算法的主要思想是：

• 對 Mapper 所分配的數據塊，將它計算成一個完整的小 Cube 段（包含所有 Cuboid）；
• 每個 Mapper 將計算完的 Cube 段輸出給 Reducer 做合併，生成大 Cube，也就是最終結果；

如圖所示解釋了此流程。

與 layer (逐層構建) 算法相比，快速算法主要有兩點不同：

• Mapper 會利用內存做預聚合，算出所有組合；Mapper 輸出的每個 Key 都是不同的，這樣會減少輸出到 Hadoop MapReduce 的數據量，Combiner 也不再需要；
• 一輪 MapReduce 便會完成所有層次的計算，減少 Hadoop 任務的調配

Kylin Cube 的構建過程#

如圖，是一個構建 Cube 的 Job 過程:

幾個重要流程分析#

從 Hive 表生成 Base Cuboid#

在實際的 cube 構建過程中，會首先根據 cube 的 Hive 事實表和維表生成一張大寬表，然後計算大寬表列的基數，建立維度字典，估算 cuboid 的大小，建立 cube 對應的 HBase 表，再計算 base cuboid。

計算 base cuboid 就是一個 MapReduce 作業，其輸入是上面提到的 Hive 大寬表，輸出的 key 是各種維度組合，value 是 Hive 大寬表中指標的值

核心源碼: org.apache.kylin.engine.mr.steps.BaseCuboidMapperBase:

// map 階段生成key-value的代碼
protected void outputKV(String[] flatRow, Context context) throws IOException, InterruptedException {
    byte[] rowKey = baseCuboidBuilder.buildKey(flatRow);
    outputKey.set(rowKey, 0, rowKey.length);

    ByteBuffer valueBuf = baseCuboidBuilder.buildValue(flatRow);
    outputValue.set(valueBuf.array(), 0, valueBuf.position());
    context.write(outputKey, outputValue);
}

從 HBase Cuboid 逐層計算 Cuboid#

核心源碼: org.apache.kylin.engine.mr.steps.CuboidReducer

@Override
    public void doReduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        aggs.reset();

        for (Text value : values) {
            if (vcounter++ % BatchConstants.NORMAL_RECORD_LOG_THRESHOLD == 0) {
                logger.info("Handling value with ordinal (This is not KV number!): " + vcounter);
            }
            codec.decode(ByteBuffer.wrap(value.getBytes(), 0, value.getLength()), input);
            aggs.aggregate(input, needAggrMeasures);
        }
        aggs.collectStates(result);

        ByteBuffer valueBuf = codec.encode(result);

        outputValue.set(valueBuf.array(), 0, valueBuf.position());
        context.write(key, outputValue);
    }

Cuboid 轉化為 HBase 的 HFile#

參考： Hive 數據 bulkload 導入 HBase

Kylin 的部署以及基本使用#

參考官方鏈接: Apache Kylin

補充:
在實踐過程中，我們採用 Nginx + Kylin 集群方案 (參考《Kylin 的集群部署模式部署》)，但由於資源緊缺， Kylin 集群和 Hadoop 集群在統一的伺服器上，經常會出現某個時刻資源緊張導致 Kylin 進程崩潰，甚至拖垮 HBase 的問題，因此建議部署單獨的 Kylin 集群，同時構建好監控機制

Kylin 的優化#

Kylin 核心在於 Cube，如何設計一個好的 Cube 對 Kylin 的性能來說便是至關重要的，影響 Cube 膨脹率和構建時間的重要因素主要有以下幾個方面:

• Cube 中的維度數量較多，且沒有進行很好的 Cuboid 剪枝優化，導致 Cuboid 數量極多；

• Cube 中存在較高基數的維度，導致包含這類維度的每一個 Cuboid 佔用的空間都很大，這些 Cuboid 累積造成整體 Cube 體積變大；

• 存在比較佔用空間的度量，例如 Count Distinct，因此需要在 Cuboid 的每一行中都為其保存一個較大的寄存器，最壞的情況將會導致 Cuboid 中每一行都有數十 KB，從而造成整個 Cube 的體積變大；

優化思路#

維度優化#

維度優化手段

• 聚合組
• 衍生維度
• 強制維度
• 層次維度
• 聯合維度
• Extended Column

詳細參考: Apache Kylin 優化指南

並發粒度優化#

當 Segment 中某一個 Cuboid 的大小超出一定的閾值時，系統會將該 Cuboid 的數據分片到多個分區中，以實現 Cuboid 數據讀取的並行化，從而優化 Cube 的查詢速度。

構建引擎根據 Segment 估計的大小，以及參數“kylin.hbase.region.cut”的設置決定 Segment 在存儲引擎中總共需要幾個分區來存儲，如果存儲引擎是 HBase，那麼分區的數量就對應於 HBase 中的 Region 數量。kylin.hbase.region.cut的默認值是 5.0，單位是 GB，也就是說對於一個大小估計是 50GB 的 Segment，構建引擎會給它分配 10 個分區。用戶還可以通過設置kylin.hbase.region.count.min（默認為 1）和kylin.hbase.region.count.max（默認為 500）兩個配置來決定每個 Segment 最少或最多被劃分成多少個分區。

由於每個 Cube 的並發粒度控制不盡相同，因此建議構建 cube 的時候為每個 Cube 量身定制控制並發粒度的參數。

參考鏈接#

• Apache Kylin 深入 Cube 和查詢優化
• Apache Kylin Job 生成和調度詳解
• Apache Kylin Cube 構建原理
• Apache Kylin 優化指南
• 【案例分享】Apache Kylin 在美團點評的應用

推薦 編程小夢 | 康凱森的技術博客, 博主是 Apache Kylin 等多個開源項目的 commiter, 博客有很多關於 OLAP 離線大數據方向的優質文章