4 月 13 日，Apache Kylin Meetup 北京站順利落幕，吸引了衆多對大資料分析技術感興趣的夥伴們到場參與，現場座無虛席。Kyligence 大資料研發工程師俞霄翔在現場與大家分享了 Kylin Real-time OLAP 功能的設計和實作，利用該功能實作的新浪微網誌實時熱點分析 Demo 掀起了現場的小高潮。
△ 俞霄翔講解 Kylin Real-time OLAP
Kylin Real-time OLAP 功能的推出是以曆史資料分析見長的 Kylin 涉足實時資料分析領域的重要裡程碑。除了在 Meetup 現場的分享，俞霄翔還寫作了《向 Kylin 添加實時 OLAP 能力》一文，文中詳細講解了 Real-time OLAP 的原理與使用步驟，同時列出了使用中的常見問題與答案。下文節選了該篇文章中《什麼是 Real-time OLAP for Kylin》章節的内容。您可以點選文末的“戳此擷取完整版文章”下載下傳完整版的文章。
背景
Apache Kylin 在誕生之初，主要目的在于解決海量資料上進行互動式資料分析的需求，資料源主要來自于資料倉庫（Hive），資料大都是曆史的而非實時的。流式資料處理是一個大資料開發的新興領域，它要求資料一旦進入系統即刻可被查詢。直到 v2.6，Apache Kylin 的主要能力仍然發揮在曆史資料分析的領域，即便是 v1.6 引入的準實時資料攝入，依然有數分鐘的延遲，難以滿足對流式資料的實時查詢需求。
為了緊跟大資料開發的發展步伐，eBay 的 Kylin 開發團隊（GitHub ID: allenma，mingmwang，sanjulian，wangshisan 等）基于 Kylin 開發了 Real-time OLAP 的特性，實作了 Kylin 對 Kafka 流式資料的實時查詢。此功能在 eBay 内部已經用于生産，并穩定運作超過一年時間，于 2018 年下半年開源，與社群共同改進和完善。
流式資料處理和實時 OLAP
對于很多商業公司，使用者消息被分析後可用于商業決策和制定更好的市場計劃，若消息更早地進入資料分析平台，那麼決策者可以更快地做出響應，進而減少時間和資金的浪費。利用流式資料處理，意味着更快的資訊回報，決策者是以可以進行更加頻繁和靈活的計劃調整。
企業資料源類型多樣，包括伺服器、手機等移動裝置、物聯網裝置等，來自不同源頭的消息往往通過不同的主題加以區分，彙聚到消息隊列（Message Queue/Message Bus）以供資料分析使用。傳統的資料分析工具使用批處理工具如 MapReduce 來進行資料分析，其資料延遲較大，通常為數小時到數天。從下圖我們可以看出，主要的資料延遲來自于兩個過程：從消息隊列通過 ETL 流程抽取到資料倉庫，和從資料倉庫抽取資料進行預計算将結果加載進 OLAP 系統供分析系統消費。由于這兩部分都是使用批處理程式進行計算，計算耗時較長，無法滿足實時查詢需求，于是我們想到要解決問題隻能繞過這些過程，通過在資料收集和 OLAP 平台間架起一道橋梁，讓資料直接進入 OLAP 平台。
目前已經有一些成熟的實時 OLAP 方案，如 Druid，通過合并實時和曆史部分的查詢結果提供較低的資料延遲。目前 Kylin 在分析海量曆史資料的方面達到了一定的水準，為了向實時 OLAP 領域邁出一步，Kylin 開發者們開發了 Real-time OLAP。
Real-time OLAP 的簡介
在新的架構下，資料查詢請求根據時間分區列（Timestamp Partition Column）分為兩部分，曆史資料的查詢請求仍将發送給 HBase Region Server，最新時間段的查詢請求将發送到實時計算節點，Query Server 需要将兩者的結果整合後傳回給查詢用戶端。
與此同時，實時計算節點會不斷将本地資料上傳到 HDFS，在滿足一定條件時會通過 Hadoop 任務來建構完整的 segment，進而完成實時資料向曆史資料的累進，并且實作了降低實時計算節點壓力的目的。
Real-time OLAP 的概念和角色
為實作 Real-time OLAP， Kylin 引入了一些新的概念，這裡跟大家做一個初步的介紹。
1. Streaming Receiver
Streaming Receiver 的角色是 worker，每個 receiver 是一個 Java 程序，受 Coordinator 的管理，它的主要職責包含：
- 實時攝入資料；
- 在記憶體建構 cuboid，定時将儲存在記憶體的 cuboid 資料 flush 到磁盤，形成 Fragment 檔案；
- 定時 checkpoint 和合并 Fragment 檔案；
- 接受對它所負責的 Partition 的查詢請求；
- 當 segment 變為不可變後，上傳到 HDFS 或者從本地删除（依據配置）；
2. Receiver Cluster
Streaming Receiver 組成的集合稱為 Receiver 叢集。
3. Streaming Coordinator
Streaming Coordinator 作為 Receiver 叢集的 Master 節點，主要職責是管理 Receiver，包括将 Kafka topic 的 partition 配置設定/解除配置設定到指定的 Replica Set，暫停或者恢複消費，收集和展示各項統計名額(例如message per second)。當 kylin.server.mode 被設定為 all 或者 stream_coordinator，這個程序就成為一個 Streaming Coordinator。Coordinator 隻進行中繼資料和叢集排程，不攝入消息。
4. Coordinator Cluster
多個 Coordinator 可以同時存在，組成一個 Coordinator 叢集。在多個 Coordinator中，同一時刻隻存在一個 Leader，隻有 Leader 才可以響應請求，其餘程序作為 standby/backup。
5. Replica Set
Replica Set 是一組 Streaming Receiver，它們動作一緻。Replica Set 作為任務配置設定的最小機關，Replica Set 下的所有 Receiver 做相同的工作（即攝入相同的一組 partition），互為 backup。當叢集中存在一些 Receiver 程序無法通路，但能保證每一個 Replica Set 至少存在一個健康的 Receiver，那麼叢集仍能正常工作并且傳回合理的查詢結果。
在一個 Replica Set 中，将存在一個 Leader Receiver 來做額外的工作，其餘的 Receiver 作為 Follower。
6. Streaming Segment
當 Receiver 攝取一個新的消息，并且這個消息的時間分區列的時間戳不包含于現有的任意 Segment，那麼 Receiver 會建立一個新的 Segment，這個 Segment 的初始狀态為 Active，并且這個 Segment 的開始時間和結束時間的間隔等于 Segment Window，所有時間分區列的時間戳包含在這個Segment的開始時間和結束時間之間的消息，将由這個 Segment 負責。當時間達到結束時間，這個 Segment 并不會立即關閉，這是為了等待那些延遲的消息，但是等待時間并不是永久的，一旦滿足以下條件後之一後，Segment 的狀态将轉為 IMMUTABLE，一旦處于 IMMUTABLE，延遲的消息預設将被丢棄：
- 在 Segment Duration 長度的時間段一直未收到屬于本 Segment 時間段的消息
- 等待時間的總和超過了某一個固定的門檻值
7. Retention Policy
當 Segment 轉變為 IMMUTABLE 後，該 Segment 的本地資料如何處理将由這個配置項決定，這個配置項目前有兩個選項：
- FULL_BUILD 目前 Receiver 程序是所屬 Replica Set 的 Leader 時，會上傳 Segment 的本地資料檔案到 HDFS，并且當上傳成功後，将 Segment 狀态置為 REMOTE_PERSISTED；若是所屬 Replica Set 的 Follower 時，不做處理。
- PURGE 等待一定時間然後删除本地資料檔案。若使用者隻對最近一段時間的資料分析結果感興趣，可以考慮使用 PURGE 選項
8. Assignment & Assigner
Assignment 是一種 Map 型的資料結構，其 key 是 Replica Set 的 ID，value 是配置設定給該 Replica Set 的 Partition 清單（用 Java 語言來表示的話是：Map<Integer, List<Partition>>），下圖是一個簡單的例子。
Assignment 十分重要，它表示了一個 Kafka Topic 的 Partition 分别是由哪些 Replica Set 去負責的，了解它對如何使用 Rebalance 相關的 API 和學習 Streaming 中繼資料十分重要。
基于目前叢集資源和 Topic Partition 數量，如何進行合适的 partition 配置設定，有不同的政策，這些政策由Assigner負責。Assigner目前有兩種具體實作，分别是CubePartitionRoundRobinAssigner，DefaultAssigner。
9. Checkpoint
Checkpoint 能夠讓 Receiver 重新啟動後，從上次結束的安全點繼續消費。Checkpoint 使得 receiver 重新開機後，資料能夠不丢失，同時盡可能減少資料的重複消費。Checkpoint 主要分為兩種，一種是 local checkpoint，一種是 remote checkpoint。其中 remote checkpoint 發生在将本地segment 資料檔案上傳到 HDFS 時，将 offset 資訊記錄在中繼資料裡；local checkpoint 是由 receiver 定時排程或者由事件觸發，會将資料 flush 到本地，并在在本地檔案記錄 offset。
當 Receiver 啟動 Kafka Consumer API 時，會嘗試檢查 local checkpoint 和 remote checkpoint，尋找最新的 offset 開始消費。
Real-time OLAP的架構
資料流向方面，我們能看到資料的流向是從 Kafka 到 Receiver，再由 Receiver 上傳到HDFS，最後由MapReduce 程式合并和重新加工 Segment 進入 HBase。
查詢方面，查詢請求由 Query Server 發出，根據查詢條件中出現的時間分區列，分發請求到 Receiver 和HBase Region Server 兩端。
Topic Partition 的配置設定和 Rebalance、Segment 狀态管理和作業送出由 Coordinator 負責。
Real-time OLAP 的特性
- 資料一旦進入，将立刻在記憶體計算 cuboid，即刻可被查詢（毫秒級資料延遲）;
- 自動化的資料狀态管理和作業排程；
- 根據查詢條件，查詢結果可包含實時結果和曆史結果；
- 實時部分使用列式存儲和反向索引加速查詢，降低查詢延遲；
- 一個新的分布式計算和存儲叢集被引入到 Apache Kylin；
- Coordinator 和 Receiver 具有高可用性。
Real-time OLAP 的 Local Segment Cache
Receiver 端的 Segment 資料由兩部分組成，MemoryStore 和 Fragment File。在 MemoryStore 内部是用 Map<String[], MeasureAggregator[]>> 存儲聚合後的資料，其中 key 是次元值組成的字元串數組，value 是 MeasureAggregator 數組。
随着 Receiver 不斷攝入消息，當 MemoryStore 的資料行數達到門檻值會觸發 flush 操作，将整個MemoryStore flush 到磁盤形成一個 Fragment File。對 Fragment File 的讀取使用了 memory-mapped file 來加速讀取速度，可以參考源代碼：
https://github.com/apache/kylin/blob/master/stream-core/src/main/java/org/apache/kylin/stream/core/storage/columnar/FragmentData.java#L55
另外，Fragment File 為了優化掃描和過濾性能也使用了多種方式來加速掃描，包括：
- 壓縮
- 反向索引
- 列式存儲
最後，Fragment File 可以進行 Merge 來減少資料備援，提升掃描性能。
Real-time OLAP 的中繼資料結構
Real-time OLAP 增加了叢集管理和 Topic Partition 的配置設定關系、Replica Set 等新的中繼資料類型，這些中繼資料目前由儲存在 Zookeeper 中，主要包括：
- 目前的 Coordinator Leader 節點
- Receiver 節點資訊
- Replica Set 資訊和 Replica Set Leader 節點
- Cube 的 Assignment 資訊
- Cube 下 Segment 的建構狀态和上傳完整度
......
【想閱讀更多内容，請根據評論區提示檢視《向 Kylin 添加實時 OLAP 能力》原文】
--------------------------------------------------------------------------------------
2019 年 11 月 16日 Apache Kylin Meetup 北京站正在火熱報名！邀請到滴滴、微衆銀行、一點資訊以及 Kyligence的技術專家為大家呈現精彩應用案例與實踐：https://www.huodongxing.com/event/2516174942311