IndexR：實時、基于Hadoop的數(shù)據(jù)倉庫
摘要
IndexR實現(xiàn)了一種可部署于分布式環(huán)境，可并行化處理，帶索引的，列式的結(jié)構(gòu)化數(shù)據(jù)格式?；谶@種數(shù)據(jù)格式，IndexR構(gòu)建了一個數(shù)據(jù)倉庫系統(tǒng)(Data Warehouse)，它基于Hadoop生態(tài)，可以對海量數(shù)據(jù)集做快速統(tǒng)計分析（OLAP），數(shù)據(jù)可實時導(dǎo)入并且對于查詢零延遲。IndexR 為解決大數(shù)據(jù)場景下分析緩慢、數(shù)據(jù)延遲、系統(tǒng)復(fù)雜等問題而設(shè)計。本文描述了IndexR的設(shè)計思想，系統(tǒng)架構(gòu)，以及核心的技術(shù)細節(jié)。
目前IndexR項目已經(jīng)開源，項目地址：https://github.com/shunfei/indexr。
簡介
舜飛科技的核心業(yè)務(wù)之一程序化廣告業(yè)務(wù)，對接全網(wǎng)的各大媒體，每秒產(chǎn)生上百萬的分析數(shù)據(jù)。這些數(shù)據(jù)對廣告投放活動的過程進行了精細的追蹤和描述，比如創(chuàng)意的展示量、點擊量，活動產(chǎn)生的注冊數(shù)、回訪數(shù)等。我們需要對這些數(shù)據(jù)進行實時分析處理，用于包括客戶報告，投放優(yōu)化，欺詐分析，收費結(jié)算等。數(shù)據(jù)使用者的查詢模式是非固定的，無法預(yù)測的，并且隨著業(yè)務(wù)量的激增，數(shù)據(jù)量也急劇增長。我們需要一種新的技術(shù)來解決這些需求：
超大數(shù)據(jù)集，低查詢延時。
查詢模式無法預(yù)測，無法預(yù)計算；
表數(shù)據(jù)量普遍超過1億，甚至上百億千億，過濾條件有可能會命中大量數(shù)據(jù)；
數(shù)據(jù)在查詢的同時還會有大量的更新，每秒入庫幾萬的數(shù)據(jù)。
要保證較低的查詢延時，一般情況下查詢延時要求在5s以內(nèi)，常用高頻查詢要求1s以內(nèi)。
準實時。
數(shù)據(jù)從產(chǎn)生到體現(xiàn)在分析結(jié)果延時幾秒以內(nèi)。
時效性對于某些業(yè)務(wù)至關(guān)重要，并且越實時的數(shù)據(jù)，價值越大。
可靠性，一致性，高可用。
這些數(shù)據(jù)是公司最重要的數(shù)據(jù)之一，任何錯誤和不一致可能會直接體現(xiàn)在客戶報表中，對公司的業(yè)務(wù)和品牌形象產(chǎn)生影響，至關(guān)重要。
可擴展，低成本，易維護。
業(yè)務(wù)會快速發(fā)展，會產(chǎn)生新的數(shù)據(jù)源，加入新的表，舊的數(shù)據(jù)不能刪除，這帶來巨大的成本壓力，和運維壓力。
典型的更新如加列、列值更新等操作不能影響線上服務(wù)，不能帶來入庫或者查詢延遲。
SQL支持。
全面支持SQL，要像Mysql一樣好用，功能強大。
不僅僅支持常見的多維分析，還需要支持復(fù)雜的分析查詢，如JOIN，子查詢等，支持自定義函數(shù)(UDF，UDFA)。
與Hadoop生態(tài)整合。
Hadoop生態(tài)的蓬勃發(fā)展給大數(shù)據(jù)處理帶來越來越強的處理能力，如果能與它的工具鏈深度結(jié)合，會極大擴展系統(tǒng)的價值。
IndexR是舜飛科技大數(shù)據(jù)平臺組為了應(yīng)對這些挑戰(zhàn)的答案。我們無法在當(dāng)前的開源產(chǎn)品中找到可以滿足所有以上需要的工具。
目前提供相似功能的產(chǎn)品，有些通過使用傳統(tǒng)的關(guān)系型數(shù)據(jù)技術(shù)，或者通過預(yù)先建Cube加速查詢。這些方式可能會帶來一些問題，比如運維困難，數(shù)據(jù)量瓶頸，或者模式不夠靈活，無法支持業(yè)務(wù)變化。有些方案使用內(nèi)存存儲技術(shù)，使用上成本比較高，而且在大數(shù)據(jù)分析場景并無特別大的速度優(yōu)勢。近年出現(xiàn)的一些時序數(shù)據(jù)庫，解決了一些入庫延遲方面的問題，但是在查詢性能，可用性，可擴展性等方面存在一些問題。
IndexR數(shù)據(jù)倉庫系統(tǒng)基于許多優(yōu)秀的開源產(chǎn)品，并且參考了一些已經(jīng)存在的工具，精心設(shè)計和實現(xiàn)而成。它把數(shù)據(jù)存放于HDFS，使用Zookeeper在集群中通訊和交涉，使用Hive方便的管理分區(qū)數(shù)據(jù)，可以通過Kafka高速實時導(dǎo)入數(shù)據(jù)，查詢層使用優(yōu)秀的分布式查詢引擎Apache Drill。它的存儲和索引設(shè)計參考了Infobright社區(qū)版和Google Mesa論文，壓縮算法借鑒了Infobright，實時入庫從HBase和Druid獲得啟發(fā)。
本文從以下幾個方面對IndexR進行闡述：
存儲格式與索引，IndexR的核心模塊。
實時入庫模塊，實現(xiàn)快速入庫并且查詢零延遲。
層次結(jié)構(gòu)與部署架構(gòu)，如何與Hadoop生態(tài)系統(tǒng)深度結(jié)合。
工程實現(xiàn)的問題以及解決方案。
典型項目選型。
數(shù)據(jù)倉庫在新環(huán)境下的挑戰(zhàn)，IndexR的意義。
目前在舜飛已經(jīng)穩(wěn)定運行，支撐了DSP、網(wǎng)站檢測分析等核心業(yè)務(wù)的實時分析任務(wù)，集群每天入庫消息300億+，目前總數(shù)據(jù)量為千億級別。
存儲格式與索引設(shè)計
數(shù)據(jù)文件
IndexR存儲結(jié)構(gòu)化數(shù)據(jù)，比如以下是一個虛構(gòu)的廣告投放用戶表 Table A:
數(shù)據(jù)文件稱為Segment，一個Segment保存一個表的部分行，包含所有的列，如下圖。
Segment文件是自解釋的，它包含版本信息，完整的表定義，各個部分的元數(shù)據(jù)(offset)，以及索引。IndexR默認對所有的列進行索引。行順序可以是入庫的自然順序，也可以是按照用戶定義的字段排序。這樣的設(shè)計可以簡化系統(tǒng)架構(gòu)，不需要額外的元數(shù)據(jù)存儲，非常適合于分布式環(huán)境下的并行處理，也方便外部系統(tǒng)如Hive直接使用。Segment的行數(shù)據(jù)在內(nèi)部會進一步細分為pack，每個pack都有獨立的索引。pack內(nèi)部的行數(shù)據(jù)是以列存儲的，即某一列的數(shù)據(jù)會集中存放在一起。這種方式對于列數(shù)據(jù)的快速遍歷，和壓縮帶來極大的優(yōu)勢。對于現(xiàn)代通用計算機架構(gòu)，cache友好，方便vector process，充分發(fā)揮現(xiàn)代多核CPU的性能。Segment的列數(shù)據(jù)使用特別優(yōu)化的壓縮算法，根據(jù)數(shù)據(jù)類型選擇不同的算法和參數(shù)，通常壓縮率10：1以上。
在實際業(yè)務(wù)數(shù)據(jù)測試中，IndexR每個節(jié)點每秒可以處理1億個字段。測試機器配置：[Intel(R) Xeon(R) CPU E5-2620 v2 @ 2.10GHz] x 2, 60G RAM, SATA 7200 RPM DISK。這個配置在目前服務(wù)器配置中算低端的，更強大的CPU會對IndexR有非常大的性能提升。
索引
IndexR采用粗糙集索引(Rough Set Index)，它能以極低的成本，很高的精確度定位到相關(guān)文件和位置。
比如我們的某一個數(shù)據(jù)塊（pack）有以下數(shù)據(jù)，有date（int類型）和use_name（string）類型。
對于number類型，會記錄該列的最大值（max）,最小值（min），然后把它們的區(qū)間(max-min)進行分割成多個區(qū)間，每一個區(qū)間使用一個bit表示。然后把各個具體的值映射到這個區(qū)間之中。IndexR對于number和string類型有不同的索引方式，這里描述基本的思路。
比如查詢?nèi)鐖D，value值為1表示這個區(qū)間存在一行或者多行的數(shù)據(jù)，為0表示不存在。我們只需要存儲max，min，和value序列（5個bit）就完成了對這一列的索引。
SELECT user_name FROM A WHERE date = '20170106'
因為'20170106'屬于區(qū)間2，value是0，即可以知道'20170106'不存在于這個pack，可以直接跳過。這是一種類似于bloomfilter的過濾方式，索引不命中的pack一定不包含需要的數(shù)據(jù)
string類型的索引和number類似，不過更加復(fù)雜一點。
目前常見的索引有B+樹索引，倒排索引，這些索引可以精確定位到具體行，在相對小數(shù)據(jù)量情況下很有效。這種方式通常沒有特別有效的壓縮，數(shù)據(jù)文件大小一般在原始數(shù)據(jù)的1~3倍之間，當(dāng)數(shù)據(jù)量膨脹到一定程度，這類索引的代價就會被放大，甚至無法服務(wù)。
IndexR的粗糙集索引的優(yōu)勢是非?？焖伲饕募銐蛐?，可以低成本的方式load到內(nèi)存，在極大數(shù)據(jù)量場景下仍然能有效的工作。由于數(shù)據(jù)通常是排序的內(nèi)聚的，通過實際數(shù)據(jù)的觀察，列的值基數(shù)（cardinality）通常比較小，這種方式是可以有效的過濾掉無關(guān)的pack。它會對所有的列進行索引，非常適合于業(yè)務(wù)不固定，或者數(shù)據(jù)分析場景的探索型分析。
實時入庫
IndexR支持實時數(shù)據(jù)追加，但不支持數(shù)據(jù)在線更新，可以通過離線的方式使用Hive等工具更新數(shù)據(jù)，這樣的設(shè)計和Mesa類似。它的入庫速度非?？?，通常單個節(jié)點單表可以達到30k消息/s。消息到達IndexR Node之后，可以立刻被查詢。
IndexR的實時入庫模塊使用類似LSM-Tree的結(jié)構(gòu)。使用commitlog文件保存消息，最新的數(shù)據(jù)存放于內(nèi)存，在達到一定閥值之后會被寫入硬盤。
內(nèi)存中的數(shù)據(jù)周期性的存儲到硬盤，時間一久會產(chǎn)生較多碎片文件，這些文件在達到一定閥值之后，會被整理合并。
行的存儲順序可以是自然入庫順序，也可以按照指定字段排序，類似于關(guān)系型數(shù)據(jù)庫中的一級索引和HBase中的Column Family，這樣做可以讓數(shù)據(jù)更加內(nèi)聚，對于查詢非常有利。
類似于Mesa，如果需要，IndexR實時入庫可以根據(jù)多維分析（Multidimensional Analysis）的概念，把字段分成維度（Dimension）和指標（Metric/Measure），具有相同維度的行會合并到一起，指標使用聚合函數(shù)（aggregation function, e.g. SUM, COUNT），并且表之間可以設(shè)計成父子關(guān)系。
如圖，Table B 與 Table C 可以可以認為是 Table A 的子表。Table A 擁有三個維度（date, country, campaign_id），可以表達最詳細的信息。Table B 與 Table C 通過減少維度，減少了數(shù)據(jù)量，可以更加快速的獲得查詢結(jié)果。
應(yīng)用層只需要做簡單的表路由，比如
SELECT date, country, SUM(impressions) FROM B WHERE country = 'CN' GROUP BY date, country
可以路由到Table B表，快速獲得結(jié)果。如果需要下鉆（Drill Down）查詢，如
SELECT campaign_id, SUM(impressions) FROM A WHERE country = 'CN' and date = '20170101' GROUP BY campaign_id
則會路由到Table A。
這種設(shè)計類似于關(guān)系型數(shù)據(jù)庫中預(yù)聚合View。在OLAP領(lǐng)域，特別是多維分析場景，這種設(shè)計非常有效。
架構(gòu)設(shè)計
IndexR的架構(gòu)設(shè)計遵循簡單可靠、易擴展的原則。它可以大規(guī)模集群部署，支持上千個節(jié)點。事實上IndexR的硬件成本相對來說很低，并且可以通過加節(jié)點線性擴展處理能力。
Apache Drill作為IndexR的查詢層。Drill是一個全新的查詢引擎，專注于SQL計算，使用了代碼生成技術(shù)，vector process，列式計算，堆外內(nèi)存（消除GC）等技術(shù)，有專門針對對于大數(shù)據(jù)集的優(yōu)化。速度極快，并且支持標準SQL，沒有遷移負擔(dān)。從我們的使用經(jīng)驗來看，它非常穩(wěn)定，工程質(zhì)量很高。
IndexR主要負責(zé)存儲層，并且對具體的查詢過程進行優(yōu)化，比如常見的條件下推（predicate pushdown），limit下推等，未來還將支持聚合下推（aggregation pushdown）。IndexR通過任務(wù)分配算法，結(jié)合數(shù)據(jù)距離、節(jié)點繁忙程度等，把計算任務(wù)分配到最合適的節(jié)點。
HDFS存儲具體的數(shù)據(jù)文件，分布式文件系統(tǒng)幫助構(gòu)建節(jié)點無狀態(tài)的服務(wù)。數(shù)據(jù)存放于HDFS中，可以方便的使用各種Hadoop工具進行其他復(fù)雜分析。我們對接了Hive，方便對數(shù)據(jù)進行離線處理。由于HDFS上的數(shù)據(jù)只有一份，可以同時被多個工具處理，省去了繁瑣的同步步驟，在10：1的高壓縮比上又節(jié)省一倍空間。
數(shù)據(jù)經(jīng)過Kafka等隊列高速導(dǎo)入IndexR。IndexR的實時導(dǎo)入非常靈活，可以隨時增加或者刪除導(dǎo)入節(jié)點。它擁有極高的導(dǎo)入性能（30k/s），入庫延遲的壓力成為歷史。
在IndexR集群中只有一種節(jié)點（IndexR Node），有利于部署和維護，不需要對節(jié)點進行劃分。目前IndexR作為Drill插件嵌入了Drillbit進程。
IndexR提供了indexr-tool工具，提供了完整的運維工具。比如可以在線更新表結(jié)構(gòu)，在線添加、修改實時入庫配置。
工程實現(xiàn)的挑戰(zhàn)
算法和數(shù)據(jù)結(jié)構(gòu)要真正落地，必須通過具體的工程來實現(xiàn)，而工程實現(xiàn)的質(zhì)量決定了項目的最終效果。如果空有高超的設(shè)計圖紙，而沒有高質(zhì)量的施工和合適的材料，高樓大廈是建不起來的。IndexR在工程上最求極致的性能，但又不失靈活的擴展性。
使用直接內(nèi)存（Direct Memory））。
IndexR主要使用Java8編寫，而Java的堆內(nèi)存（Heap）與垃圾回收（GC）的模式在大數(shù)據(jù)運算場景下面臨比較大的挑戰(zhàn)。
在需要使用較大內(nèi)存（超過32G）以及數(shù)據(jù)更新頻繁時，JVM的GC問題比較明顯，容易造成性能不穩(wěn)定，并且對象實例的內(nèi)存模型通常很浪費內(nèi)存。
我們在IndexR項目中把所有的存儲數(shù)據(jù)和運算臨時數(shù)據(jù)存放于堆外，手動管理內(nèi)存申請釋放。
這樣提高了代碼復(fù)雜度，但相比于傳統(tǒng)的堆內(nèi)存模式，節(jié)省了超過1/2內(nèi)存，并且沒有了GC代價，涉及大量數(shù)據(jù)的賦值操作通?？梢允褂脙?nèi)存拷貝，節(jié)省大量CPU循環(huán)。
充分利用現(xiàn)代CPU能力。
IndexR的堆外內(nèi)存模型對于充分發(fā)掘硬件潛能非常有益，它們通常是連續(xù)的內(nèi)存塊，沒有類指針跳轉(zhuǎn)，沒有虛函數(shù)損耗，CPU寄存器和多級緩存都可以充分利用，而且對于使用vector processor非常便利，沒有結(jié)構(gòu)轉(zhuǎn)換開銷。
避免隨機讀取。
通常磁盤的特點是連續(xù)讀取非常快，因而Kafka可以使用磁盤做消息隊列；
而隨機讀取相對很慢，故傳統(tǒng)數(shù)據(jù)庫的瓶頸一般在IO。
IndexR的索引方式對磁盤連續(xù)讀取友好，并且它會對數(shù)據(jù)進行整理從而更加內(nèi)聚。
我們還特別對文件讀取方式進行了細致的優(yōu)化。
優(yōu)化線程、IO調(diào)度。
在任務(wù)非常繁忙的時候，CPU爭搶帶來的線程切換的開銷變的不可忽視。
并且由于數(shù)據(jù)庫環(huán)境的特殊性，在做繁忙CPU任務(wù)的同時，還會進行網(wǎng)絡(luò)、IO操作。
如何做任務(wù)調(diào)度，合理安排線程數(shù)量和任務(wù)，對整體性能影響比較大。
有時候單線程比多線程效率更高，并且更省資源。
關(guān)鍵性能點使用C++實現(xiàn)。
它在同時涉及內(nèi)存操作和復(fù)雜CPU運算場景時，運行效率優(yōu)勢明顯。
我們把關(guān)鍵的性能點，比如壓縮算法，使用C++實現(xiàn)。
工具選型
IndexR是一個新的工具，如果你的項目有以下需求，或者之前已經(jīng)有一些選型但是無法滿足需求，可以考慮使用IndexR。
IndexR適合的經(jīng)典場景：
需要在海量數(shù)據(jù)之上做快速的統(tǒng)計分析查詢。
要求入庫速度非?？欤⑶倚枰獙崟r分析。
存放超大量歷史明細數(shù)據(jù)庫。
比如網(wǎng)站瀏覽信息，交易信息，安保數(shù)據(jù)，電力行業(yè)數(shù)據(jù)，物聯(lián)網(wǎng)設(shè)備采集數(shù)據(jù)等。
這類數(shù)據(jù)通常量非常大，數(shù)據(jù)內(nèi)容復(fù)雜，存放時間比較久，且希望在需要時可以比較快速的根據(jù)各種條件做明細查詢，或者在一定范圍內(nèi)做復(fù)雜的分析。
這種情況下可以充分發(fā)揮IndexR的低成本，可擴展，適合超大數(shù)據(jù)集的優(yōu)勢。
目前業(yè)界典型選型：
使用Mysql，PostgreSQL等關(guān)系型數(shù)據(jù)庫，不僅用于業(yè)務(wù)查詢（OLTP），也做統(tǒng)計分析，一般是在現(xiàn)有業(yè)務(wù)數(shù)據(jù)庫上直接做一些分析需求。
這種方式在數(shù)據(jù)量增長之后就會遇到性能問題，特別是分析查詢會對業(yè)務(wù)查詢產(chǎn)生極大影響。
可以考慮把數(shù)據(jù)導(dǎo)入IndexR做分析，即把業(yè)務(wù)數(shù)據(jù)庫和分析數(shù)據(jù)庫分開。
ES，Solr等全文搜索數(shù)據(jù)庫用于統(tǒng)計分析場景。
這類數(shù)據(jù)庫最大的特點是使用了倒排索引解決索引問題。
對于統(tǒng)計分析場景通常沒有特別優(yōu)化，在大數(shù)據(jù)量場景下內(nèi)存和磁盤壓力比較大。
如果遇到性能問題，或者數(shù)據(jù)量撐不住了，可以考慮使用IndexR。
Druid，Pinot等所謂時序數(shù)據(jù)庫。
在查詢條件命中大量數(shù)據(jù)情況下可能會有性能問題，而且排序、聚合等能力普遍不太好，從我們的使用經(jīng)驗來看運維比較困難，靈活性和擴展性不夠，比如缺乏Join、子查詢等。
在保存大量歷史數(shù)據(jù)情況下需要的硬件資源相對昂貴。
這種場景下可以考慮使用IndexR直接替換，不用擔(dān)心業(yè)務(wù)實現(xiàn)問題。
Infobright，ClickHose等列式數(shù)據(jù)庫。
列式數(shù)據(jù)庫本身非常適合于OLAP場景，IndexR也屬于列式數(shù)據(jù)庫。
最大的區(qū)別在于IndexR是基于Hadoop生態(tài)的。
離線預(yù)聚合，建Cube，結(jié)果數(shù)據(jù)存放于HBase等KV數(shù)據(jù)庫，如Kylin等。
這種方式在只有多維分析場景且查詢比較簡單的情況下非常有效。
問題就在于靈活性不足（flexibility），無法探索式分析，以及更復(fù)雜的分析需求。
IndexR可以通過表配置達到預(yù)聚合的效果，并且聚合是實時，沒有延遲的；
可以保留原始數(shù)據(jù)或者高維度數(shù)據(jù)，通過表路由決定具體的查詢表。
為了解決大數(shù)據(jù)量的即時分析問題，上層使用Impala，Presto，SparkSQL，Drill等計算引擎來做查詢，存儲層使用開源數(shù)據(jù)格式比如Parquet，基于Hadoop生態(tài)。
這類架構(gòu)和IndexR很相似。
IndexR的優(yōu)勢在于更有效的索引設(shè)計，更好的性能，并且支持實時入庫，秒級延遲。
我們在相同環(huán)境下與Parquet格式做過查詢性能對比，IndexR的查詢速度提升在3~8倍以上。
之后IndexR經(jīng)歷了很大的性能優(yōu)化，估計會有更好的表現(xiàn)。
Kudu，Phoenix等既支持OLTP場景，又為OLAP場景優(yōu)化等開源產(chǎn)品。
通常很難兩者兼顧，建議分成實時庫和歷史庫，針對不同數(shù)據(jù)特點采用不用的存儲方案。
內(nèi)存數(shù)據(jù)庫。
貴。
舜飛科技大數(shù)據(jù)平臺組對于以上提到的大部分技術(shù)選型有著豐富的經(jīng)驗，即這些工具我們或者在生成環(huán)境中使用過，或者有過深入的調(diào)研和測試，這也促使了IndexR的誕生。
思考和總結(jié)
大數(shù)據(jù)經(jīng)過近些年的快速發(fā)展，完整的生態(tài)漸漸成熟，已經(jīng)早已不是只有Hadoop跑MR任務(wù)的時代。人們在在滿足了能夠分析大量數(shù)據(jù)集的需求之后，漸漸的對時效性、易用性等方面提出了更高的要求，因而誕生了如Storm，Spark等新的工具。新的問題催生新的挑戰(zhàn)，提供新的機遇。而傳統(tǒng)的數(shù)據(jù)倉庫產(chǎn)品，在面對大數(shù)據(jù)的沖擊顯得非常無力。IndexR為解決這種現(xiàn)狀，提供了新的思路和方向。
IndexR是新一代數(shù)據(jù)倉庫系統(tǒng)，為OLAP場景而設(shè)計，可以對超大量的結(jié)構(gòu)化數(shù)據(jù)進行快速的分析，支持快速的實時入庫。它功能強大，并簡單可靠，支持大規(guī)模集群部署。它與Hadoop生態(tài)系統(tǒng)深度整合，可以充分發(fā)揮大數(shù)據(jù)工具的能力。
不用再為分析能力的瓶頸擔(dān)憂，不用放棄經(jīng)典的OLAP理論，不用降級你的服務(wù)，不用擔(dān)心業(yè)務(wù)人員對大數(shù)據(jù)工具不熟悉，IndexR像Mysql一樣好用，會SQL就好了。
IndexR在開源之后，我們已經(jīng)看到有不少使用案例，包括國內(nèi)外的不同團隊。有意思的是，有些團隊的使用方式比較特別，比如用于存放超大量（單表千億級別）的復(fù)雜明細數(shù)據(jù)，做歷史數(shù)據(jù)的明細查詢。IndexR不僅可以用于多維分析，商業(yè)智能等OLAP經(jīng)典領(lǐng)域，還可以用于物聯(lián)網(wǎng)，輿情監(jiān)控，人群行為分析等新興方向。