Apache Kylin在 1.5.0 推出了從流數(shù)據(jù)進(jìn)行準(zhǔn)實(shí)時(shí)(Near Real Time)處理功能，可以直接從Apache Kafka的主題(Topic)中消費(fèi)數(shù)據(jù)來構(gòu)建Cube。Apache Kylin 1.5.0的流數(shù)據(jù)是一次實(shí)驗(yàn)性的探索，它打破了以往只能從Apache Hive表構(gòu)建Cube的局限，將數(shù)據(jù)從產(chǎn)生到可查詢的延遲從小時(shí)級降低到了分鐘級，滿足了一些對實(shí)時(shí)性要求比較高的場景;但它在實(shí)現(xiàn)上存在一些局限︰

不可擴(kuò)展︰ 由于是利用單個(gè) Java 進(jìn)程(而不是利用某種計(jì)算框架)對數(shù)據(jù)做處理，當(dāng)遇到流數(shù)據(jù)高峰時(shí)，可能由于資源不足而導(dǎo)致構(gòu)建失敗;

可能會(huì)丟失數(shù)據(jù)︰ 由于使用一個(gè)起始時(shí)間+結(jié)束時(shí)間在Kafka隊(duì)列中使用二分查找近似地尋找消息的偏移量(offset)，過早或過晚到達(dá)的消息將會(huì)被遺漏，從而使得查詢結(jié)果有誤差 ;

難以監(jiān)控︰ 用于構(gòu)建的任務(wù)是單獨(dú)通過shell腳本執(zhí)行的，而不是像其它Cube那樣由任務(wù)引擎統(tǒng)一調(diào)度和執(zhí)行，所以這些任務(wù)是在Web界面和REST API上都無法查詢到的，使得用戶無法方便地使用工具進(jìn)行監(jiān)控和管理;

其它︰ 必須持續(xù)執(zhí)行，如果有系統(tǒng)宕機(jī)將會(huì)造成某些時(shí)間窗口的任務(wù)沒有被執(zhí)行，從而必須依靠管理員手動(dòng)恢復(fù);如果宕機(jī)時(shí)間較長，管理員不得不將長時(shí)間窗口切成多個(gè)小時(shí)間窗口依次來恢復(fù)，非常繁瑣 。

為了克服這些限制，Apache Kylin團(tuán)隊(duì)基于Kafka 0.10的API，開發(fā)了新版的準(zhǔn)實(shí)時(shí)流式處理，它已經(jīng)在內(nèi)部測試了一些時(shí)間，目前正在公開測試中。

新版流式構(gòu)建是在Kylin v1.5的"可插拔 "架構(gòu)下的一個(gè)***實(shí)現(xiàn)︰ 將Kafka主題視為一種數(shù)據(jù)源，實(shí)現(xiàn)相應(yīng)的適配器，將數(shù)據(jù)先抽取、轉(zhuǎn)換和保存到 HDFS，接下來使用各種Kylin的構(gòu)建引擎(MR/Spark等)對數(shù)據(jù)進(jìn)行并行計(jì)算 。圖 1 是高層次的架構(gòu)圖。

Kylin的Kafka適配器參考了kafka-hadoop-loader的思路和部分代碼， 將Kafka主題抽象成Hadoop輸入文件格式(InputFileFormat)，為主題的每個(gè)分區(qū)(partition)分配一個(gè)Mapper消費(fèi)數(shù)據(jù); 之后Kylin將利用現(xiàn)有框架進(jìn)行并行處理，從而使得方案變得可擴(kuò)展且具有容錯(cuò)性。

要解決"數(shù)據(jù)丟失"問題，Kylin將開始/結(jié)束消息的偏移量(offset)計(jì)入了每個(gè)Cube segment，并使用偏移量作為分區(qū)值 ，offset是順序遞增的且不能有重疊和遺漏(如果主題有多個(gè)分區(qū)，使用各分區(qū)偏移量之和作分區(qū)值);這將確保沒有數(shù)據(jù)丟失，一個(gè)消息只會(huì)被消費(fèi)一次。晚到達(dá)的消息會(huì)被稍后的segment統(tǒng)計(jì)進(jìn)來;每個(gè)Segment 有"最早時(shí)間”和"最晚時(shí)間"; 當(dāng)用戶按時(shí)間條件查詢時(shí)，Kylin將掃描與查詢時(shí)間范圍相匹配的所有段。圖 2解釋了這個(gè)設(shè)計(jì)。

上圖中有三個(gè)segment，它們的offset依次連續(xù)且無重疊(左包右閉)，Seg[100-400]中的消息時(shí)間跨度是1:04 – 1:11，Seg[400 - 2000]的時(shí)間跨度是1:08 – 1:40;當(dāng)用戶要查詢1:10的統(tǒng)計(jì)信息時(shí)，Kylin發(fā)現(xiàn)這兩個(gè)Segment都可能有這個(gè)時(shí)間的消息，故而會(huì)掃描這兩個(gè)Segment然后再次做匯總計(jì)算。

新版流計(jì)算引擎也進(jìn)行其它一些更改和增強(qiáng)︰

允許同時(shí)構(gòu)建/合并多個(gè)segment，前后的構(gòu)建任務(wù)都是獨(dú)立的

自動(dòng)從前一個(gè)segment或從Kafka尋找消息的開始及結(jié)束的offset

支持嵌入格式(結(jié)構(gòu)化)的JSON消息

增加了觸發(fā)流式構(gòu)建的REST API

增加了來檢查和部分填補(bǔ)segment空洞的REST API

內(nèi)部的集成測試結(jié)果初步驗(yàn)證了當(dāng)初的目標(biāo) ︰

可伸縮︰ 它能夠在一次構(gòu)建中輕松處理上億條消息;

靈活︰ 可以在任何時(shí)候，以你期望的頻率觸發(fā)構(gòu)建，例如︰ 在白天每隔 5 分鐘觸發(fā)一次， 在夜間將頻率降低到每個(gè)小時(shí)，在需要做的維護(hù)可以隨時(shí)暫停; 由于是Kylin管理所有主題的offset，再恢復(fù)時(shí)它可以自動(dòng)從上一次的結(jié)束位置繼續(xù);

穩(wěn)定︰ 穩(wěn)定性大大提高，在上一版中經(jīng)常發(fā)生的OutOfMemory錯(cuò)誤再?zèng)]有出現(xiàn)過;

易于管理︰ 用戶可通過Kylin的"Monitor"頁面或 REST API檢查所有構(gòu)建任務(wù)的狀態(tài);

構(gòu)建性能︰相比于前一版構(gòu)建時(shí)間略長(因?yàn)橛蠬adoop任務(wù)的調(diào)度)，但延遲依然在可接受的分鐘級別。

在一個(gè)小規(guī)模的測試群集 (8臺(tái) AWS實(shí)例，消費(fèi) Twitter Sample 消息流) 中，創(chuàng)建一個(gè)有9個(gè) 維度和3個(gè)度量的Cube,每秒約一萬條消息，當(dāng)構(gòu)建間隔是 2 分鐘的時(shí)候，平均每次構(gòu)建需 3 分鐘; 當(dāng)構(gòu)建間隔是 5 分鐘的時(shí)候，平均每次構(gòu)建需要 4 分鐘; 這里是幾個(gè)測試中的截圖 ︰

總結(jié)，這是比前一版本相比更加健壯和完善的流數(shù)據(jù)OLAP 解決方案?，F(xiàn)在你可以從Apache Kylin的下載頁面下載到 1.6.0-SNAPSHOT 的二進(jìn)制包，然后按照此教程生成***個(gè)流式Cube。

作者介紹：

史少鋒，Apache Kylin PMC 成員，核心開發(fā)人員之一，eBay高級軟件工程師，2014年加入eBay Kylin 團(tuán)隊(duì)并轉(zhuǎn)向大數(shù)據(jù)分析領(lǐng)域，參與了Kylin一系列優(yōu)化和新功能的開發(fā)，并致力為Kylin社區(qū)用戶提供支持和幫助。史少峰碩士畢業(yè)于上海交通大學(xué)計(jì)算機(jī)系，在IBM從事多年軟件全球化和云計(jì)算等方面的設(shè)計(jì)和開發(fā)。