CC-MRSJ：Hadoop平臺下緩存敏感的星型聯(lián)接算法*

周國亮，朱永利，王桂蘭

(1．華北電力大學控制與計算機工程學院 保定 071003；2．國網(wǎng)冀北電力有限公司技能培訓中心 保定 071051)

1 引言

隨著大數(shù)據(jù)在各個行業(yè)的迅速發(fā)展，傳統(tǒng)的并行數(shù)據(jù)倉庫系統(tǒng)在處理大數(shù)據(jù)時由于成本高、可擴展性差及容錯能力低等原因而力不從心。因此，基于Map Reduce[1,2]技術的數(shù)據(jù)倉庫系統(tǒng)獲得了廣泛關注和應用，如Facebook的Hive[3]、Yahoo的 Pig[4]等。然而，困擾 Map Reduce數(shù)據(jù)倉庫系統(tǒng)進一步深入應用的主要問題是性能低下。因此，圍繞如何提高Map Reduce系統(tǒng)的性能是目前各方普遍關注的熱點難點。在Map Reduce環(huán)境下，數(shù)據(jù)一般以分布式文件系統(tǒng)的形式存儲在集群的磁盤上，其主要代價是磁盤和網(wǎng)絡I/O，而不同的數(shù)據(jù)存儲格式對系統(tǒng)的I/O有較大的影響，從而影響系統(tǒng)性能。在傳統(tǒng)數(shù)據(jù)倉庫系統(tǒng)中普遍采用列存儲提高性能[5~7]。通過列存儲可以有效地較少I/O，并利用數(shù)據(jù)壓縮技術，進一步提高性能。因此，隨著Map Reduce技術的流行，列存儲在Map Reduce系統(tǒng)中獲得廣泛關注，提出了 RCFile[8]、COF[9]和 CFile[10]等列存儲文件格式?；诹写鎯Φ臄?shù)據(jù)結構，Map Reduce性能獲得了一定提高。

在數(shù)據(jù)倉庫中，通常采用星型模式作為數(shù)據(jù)組織和存儲模型。一般情況下，星型模型由一個很大的事實表和多個較小的維表組成。因此在數(shù)據(jù)倉庫中，星型聯(lián)接(star join)是一項核心和基本的操作，即事實表和多個維表之間的聯(lián)接。Map Reduce中已有的聯(lián)接算法[10~12]較少考慮星型聯(lián)接的特殊性，從而造成事實表數(shù)據(jù)移動和龐大的中間結果，因此性能很難提高?？紤]星型模型的結構，提高性能的關鍵是盡量減少事實表的磁盤I/O和數(shù)據(jù)移動，事實表進行本地處理。

另一方面，Map Reduce起初部署在配置較低的機器上，內存與磁盤的存取速度差距不大，較少考慮現(xiàn)代處理器的多層存儲體系結構[13]，造成CPU緩存利用率低。而目前的Map Reduce普遍部署在配置很高的機器上，內存與磁盤的讀寫速度差距越來越大，因此充分利用現(xiàn)代處理器的緩存特性[14]，有利于進一步提高 Map Reduce算法的性能[15,16]。

本文提出了一種具有緩存敏感特性的CC-MRSJ（cache consc iousMap Reduces tarjoin）算法，首先對事實表進行垂直劃分，每列單獨存儲；維表根據(jù)具有的維層次結構劃分為不同的列簇；外鍵列和對應維表采用相關性存儲，盡量部署在相同的節(jié)點上。通過上述存儲結構，可以提高緩存利用率，減少數(shù)據(jù)在節(jié)點間的移動。算法通過兩個階段完成，首先外鍵列和對應維表進行聯(lián)接，可以采用map-reduce聯(lián)接或map聯(lián)接；然后對中間結果根據(jù)位置索引進行reduce聯(lián)接并輸出，利用位置索引隨機讀取查詢中涉及的測度列。通過對SSB[17]數(shù)據(jù)集的Q3.1和Q4.1測試表明，CC-MRSJ算法具有較高的執(zhí)行效率。

2 相關工作

2.1 MapReduce及其數(shù)據(jù)存儲格式

MapReduce是由Google公司近年來提出的一種大數(shù)據(jù)處理框架，MapReduce程序一般分為map和reduce兩個階段。map函數(shù)接受鍵值對(key-valuepair)輸入，產生一組中間結果鍵值對，MapReduce框架會將map函數(shù)生成結果中鍵相同的值傳遞給一個reduce函數(shù)；reduce函數(shù)接受一個鍵以及相關的一組值，將這組值進行合并產生一組規(guī)模更小的值。

MapReduce的思想是將計算推向數(shù)據(jù)，盡量減少數(shù)據(jù)移動。目前，由Apache基金會組織開發(fā)的MapReduce開源實現(xiàn)Hadoop[18]在工業(yè)界和學術界獲得了廣泛應用。

MapReduce中的數(shù)據(jù)主要存儲在磁盤，程序運行中的主要代價是磁盤和網(wǎng)絡I/O。因而不同的數(shù)據(jù)存儲格式對性能具有較大的影響，通過列存儲可以有效減少I/O，進而提高性能，因此列存儲技術在MapReduce中獲得了廣泛關注，多種列存儲模型被提出。列存儲中一般通過位置索引（隱式RowId）組織不同列的數(shù)據(jù)。目前，Hive系統(tǒng)提供RCFile（record columnarfile）[8]格式數(shù)據(jù)存儲，RCFile 首先將數(shù)據(jù)水平劃分為數(shù)據(jù)塊，然后在塊內垂直劃分為列簇。而CFile[10]不考慮水平劃分，而是將表垂直劃分為多個列簇。CIF[9]采用每列單獨存儲，不考慮列簇，但不同列的數(shù)據(jù)采用相關性存儲，從而減少列組合時的網(wǎng)絡數(shù)據(jù)傳輸。然而這些存儲模型并沒有考慮星型模型的特殊結構。

2.2 聯(lián)接算法

聯(lián)接操作是數(shù)據(jù)庫中的一項核心操作，研究人員對此進行了深入系統(tǒng)的研究。而隨著MapReduce對多數(shù)據(jù)源分析需求的增加，研究人員提出了多種Map Reduce環(huán)境下的聯(lián)接算法，目前Hive系統(tǒng)中主要支持的聯(lián)接算法如下。

（1）map-reduce聯(lián)接

在map階段，兩個表根據(jù)聯(lián)接關鍵字進行劃分和排序；在reduce階段，對每一個分區(qū)進行聯(lián)接操作。這種方式類似于關系數(shù)據(jù)庫的散列聯(lián)接（Hashjoin）。

（2）map 聯(lián)接

假如參加聯(lián)接操作的兩個表中有一個足夠小，小表被分發(fā)到各個節(jié)點，于是在map階段大表的每個片段與小表進行聯(lián)接。這種方式類似于數(shù)據(jù)庫中的嵌套循環(huán)聯(lián)接（nested loopjoin）。

（3）分塊的 map 聯(lián)接

將參加聯(lián)接的表劃分為對應的數(shù)據(jù)塊，于是在map階段一個表中的塊與另一表中對應的塊進行聯(lián)接，不需要reduce操作。這種方式類似于Sort-Merge聯(lián)接中的Merge操作。

星型聯(lián)接是數(shù)據(jù)倉庫中非常普遍的一種操作，通常情況下星型聯(lián)接是一個事實表F和多個維表D1,D2,…,Dn之間的聯(lián)接。事實表和維表具有如下特征：

·維表Di的主鍵PK和事實表F的外鍵列FKi之間存在對應關系；

·事實表一般由外鍵列和測度列組成：F（fk1,fk2,…,fkn,m1,m2,…,mk）。

為了在Map Reduce環(huán)境下高效處理星型聯(lián)接，相關研究人員提出了并發(fā)聯(lián)接（concurrent join）[10]和Scatter-Gather-Merge（SGM）[12]算法。這兩個算法的核心思想都是把星型聯(lián)接劃分為多個小的聯(lián)接，然后再合并中間結果。并發(fā)聯(lián)接算法采用垂直劃分，SGM采用水平劃分。但兩者未考慮事實表和維表相關性存儲及緩存特性。

3 C C-MRSJ算法

3.1 數(shù)據(jù)存儲格式

列存儲一般將所有列劃分為不同的列簇，經(jīng)常一起訪問的列組成一個簇，從而減少了列組合實體化（materialization）的代價，也避免了訪問不相關的列。但是，預測哪些列經(jīng)常一起訪問是非常困難的，因此需要構建多種列組合模式的列簇，某些列會多次出現(xiàn)在不同的列簇中，從而造成存儲空間浪費?？紤]到星型模型的結構，事實表很大而且列數(shù)較多，而預測哪些列經(jīng)常一起訪問存在一定困難，所以事實表采取每列單獨存儲方式。而用戶對維表的訪問，一般是和某個維層次（dimens ionhi erarchy）相關的，因此維表采用列簇存儲，根據(jù)維表具有的維層次劃分為多個列簇，每個列簇包含主鍵和對應的維層次屬性。比如SSB[16]中的customer維表可以劃分為如下幾個列簇：{[CUSTKEY、NAME]、[CUSTKEY、CITY]、[CUSTKEY、NATION]、[CUSTKEY、REGION]、[CUSTKEY、其他列]}。劃分后的 customer表如圖1所示。

通過這種方式有效地減少了聯(lián)接操作中的磁盤I/O；同時無論是事實表還是維表只訪問需要操作的列，提高了CPU緩存利用率，使算法具有緩存敏感特性。

事實表的外鍵列占用存儲空間遠遠大于維表，比如SSB數(shù)據(jù)集在SF=10時，事實表外鍵列l(wèi)o_custkey占用存儲空間約為1GB（含位置索引），如果Hadoop的塊大小為64MB，采用3份復制策略，則lo_custkey需要48塊存儲，可以分布到小規(guī)模集群中的大部分節(jié)點。而維表customer的大小約為28MB，Hadoop會采用一個數(shù)據(jù)塊存儲，最多只能分布在3個節(jié)點上。因此對事實表和維表采取不同的復制策略，要求維表的復制數(shù)至少為nd=size(fki)/64。另外，星型模型中事實表的每個外鍵列一般情況下只需要與對應的維表進行聯(lián)接操作，事實表和維表采取相關性存儲，也就是事實表外鍵列和對應維表的數(shù)據(jù)塊存儲在同一節(jié)點，從而減少聯(lián)接操作過程中從其他節(jié)點拉取數(shù)據(jù)的情況，減少網(wǎng)絡I/O。存儲格式可以用圖2表述，其中D2-hn表示維表列簇塊，fki表示外鍵列的分塊。

3.2 算法描述

基于上述存儲結構，本文提出了CC-MRSJ算法。將星型聯(lián)接劃分為兩個步驟完成：首先每個維表和對應的事實表外鍵列進行聯(lián)接；然后對產生的中間結果進行聯(lián)接并讀取測度列的值，從而得到最終結果。執(zhí)行過程如圖3所示。

圖1 維表的列簇存儲模式

圖2 星型模型存儲模式

圖3 星型聯(lián)接執(zhí)行計劃

可以用式（1）表示：

在算法的第一個階段中，維表和外鍵列的聯(lián)接操作根據(jù)集群中的計算資源可以并行或串行。所謂并行，就是多個外鍵列和對應的維表同時進行聯(lián)接；串行是一次一個維表和一個外鍵列聯(lián)接。通過并行可以進一步提高算法的效率。

根據(jù)維表的存儲和分布特點，聯(lián)接方式可以是map-reduce聯(lián)接，也可以是map聯(lián)接。如果維表的全部數(shù)據(jù)可以在一個節(jié)點獲得，則外鍵列的一部分與維表聯(lián)接，使聯(lián)接可以在map階段完成，否則需要map-reduce聯(lián)接才能完成。

這一階段的輸出結果是以位置索引為key的鍵值對，值為聯(lián)接需要從維表中讀取的某列的值。這一階段是CC-MRSJ算法的主要代價。

第二個階段對位置索引進行排序，然后計算每個位置索引的個數(shù)(count)，如果個數(shù)與參加星型聯(lián)接的維表個數(shù)相等，則將此位置索引對應的值組合輸出。對于測度列，根據(jù)位置索引以隨機訪問的方式讀取對應位置的值。

下面以SSB數(shù)據(jù)集中的Q3.1為例，說明算法的執(zhí)行過程，Q3.1涉及多個查詢條件，是一個事實表和3個維表的聯(lián)接。只考慮聯(lián)接操作，因此去掉了groupby操作和聚集計算，該查詢語句具有如下形式：

SELECT c_nation,s_nation,d_year,lo_revenue

FROM customer

JOIN lineorder ON lo_custkey=c_custkey

JOIN supplier ON lo_suppkey=s_suppkey

JOIN ddate ON lo_orderdate=d_datekey

WHERE c_region= ’ASIA’

AND s_region= ’ASIA’

AND d_year>=1992 and d_year<=1997;

詳細的執(zhí)行過程如圖4所示。

圖4 星型聯(lián)接的例子

首先，外鍵列和對應維表的列簇進行聯(lián)接，生成結果的鍵值對是位置索引和維表中維層次的值；如圖4所示，3個維表分別與對應的外鍵列聯(lián)接。

然后，每個中間結果作為輸入進行reduce計算，根據(jù)位置索引進行排序，如果同一位置索引的值達到3則輸出；如圖4中位置索引為5的紀錄滿足3個查詢條件，是最終結果集中的一條記錄。

最后，根據(jù)上一步結果中的位置索引，到測度列中以隨機訪問的方式讀取對應位置的測度值，并與結果集合并，形成聯(lián)接操作的最終結果。

CC-MRSJ算法的形式化描述如下，這里省略了算法的解釋。

map(k,v)//mapsidejoin:維表存儲在分布式緩存中DistributedCache

輸入：

kisakey//position:位置索引，隱式或顯式RowId

3.3 算法分析

影響CC-MRSJ算法的主要因素是算法第一階段的磁盤I/O和網(wǎng)絡數(shù)據(jù)傳輸。其中磁盤I/O主要包括原始數(shù)據(jù)的讀取和中間結果的寫入、讀?。欢W(wǎng)絡數(shù)據(jù)傳輸主要是map的結果向reduce節(jié)點的傳輸。事實表外鍵列所需存儲空間比維表要大很多，而且一般情況下維表和外鍵列聯(lián)接后的結果是過濾后的數(shù)據(jù)，相對于原始數(shù)據(jù)需要更少的存儲空間。因此，算法的主要代價在第一個階段，而第一個階段是由多個相似的事實表外鍵列和維表層次列簇之間的聯(lián)接，而此操作具有相似的代價，所以算法的代價與聯(lián)接中維表的個數(shù)，也就是外鍵列的個數(shù)成正比：

其中，p表示代價，k表示比例因數(shù)。

另外，如果集群的規(guī)模足夠大，可以支持多個維表和事實表外鍵的并發(fā)聯(lián)接操作，算法可以保持近似的執(zhí)行效率，即算法具有橫向擴展特征。

目前，計算機普遍配置有很高的內存容量，因此可以利用大內存優(yōu)化算法的性能。如果維表的列簇可以存儲在節(jié)點的內存中，將維表分發(fā)到每個節(jié)點的本地內存中，則外鍵列和維表之間的聯(lián)接可以通過map聯(lián)接方式完成，從而獲得更高的效率。Hadoop的最新版本也提供了分布式緩存（Hadoopdistributedcache）功能，為應用程序充分利用大內存提供了便利。

4 實驗結果與分析

4.1 實驗環(huán)境

本實驗采用SSB數(shù)據(jù)集，SSB數(shù)據(jù)集是由TPC-H數(shù)據(jù)集演變而來，其中包括一個事實表（LINEORDER）和4個維表（CUSTOMER、SUPPLIER、PART 和 DATE）。當設置 SF=10時，事實表大約為6GB，可以劃分為約100個64MB的數(shù)據(jù)塊，每個數(shù)據(jù)塊配置2個備份，對于包含有限個節(jié)點的小集群，數(shù)據(jù)可以均勻地覆蓋整個集群。數(shù)據(jù)以文本文件的形式存儲在磁盤上。對于事實表的每列只存儲列值，在map操作中，輸入的鍵值對為位置索引和對應位置的列值。

目前Hive算法支持星型聯(lián)接，CC-MRSJ算法與Hive進行比較，Hive的版本采用0.8.1，數(shù)據(jù)存儲格式包括TextFile和 RCFile。

實驗室搭建了包括6個節(jié)點的MapReduce測試環(huán)境，主節(jié)點配置的機器是DELLPowerEdgeT310，使用Intel XeonCPUX3430@2.40GHz處理器和2GBRAM，系統(tǒng)LinuxUbantu版本。數(shù)據(jù)節(jié)點是Lenovo計算機，配置Intel PentiumD處理器和2GBRAM。采用的Hadoop的版本為0.20.2。

SSB的Q4.1查詢語句在去掉分組和聚集運算后的格式如下：

SELECTc_nation,p_name,s_nation,d_year,

lo_revenue,lo_supplycost

FROM lineorder

JOIN customer ON lo_custkey=c_custkey

JOIN supplier ON lo_suppkey=s_suppkey

JOIN part ON lo_partkey=p_partkey

JOIN dwdate ON lo_orderdate=d_datekey

AND s_region= ‘ASIA’

AND d_year>=1992 and d_year<=1997

AND(p_mfgr= ‘MFGR#1’or p_mfgr= ‘MFGR#2’);

此查詢是一個事實表和4個維表的聯(lián)接操作。

另外，Hive需要將SQL語句轉化為MapReduce程序，目前Hive轉化的后程序還存在較大的優(yōu)化空間，于是一些研究人員設計了SQL轉MapReduce的第三方程序，比如YSmart[19，20]等，以期提高 MapReduce程序的效率，但這些轉化只是語句方面的優(yōu)化，并未考慮數(shù)據(jù)存儲格式，所以算法效率和Hive轉化后的程序并沒有較大提高，所以本文直接與Hive系統(tǒng)進行比較。

4.2 實驗結果

首先對SSB數(shù)據(jù)集上去掉聚集運算的Q3.1和Q4.1查詢進行比較測試，分別與Hive中的以TextFile和RCFile存儲格式的聯(lián)接算法進行比較，實驗結果如圖5所示。

圖5 原始查詢

對于星型聯(lián)接，目前Hive處理的方式是事實表先與其中一個維表聯(lián)接，然后中間結果和另一個維表進行聯(lián)接，直到所有維表處理完畢，即((F∞D1)∞D2)…∞Dn，這樣會產生很大的中間結果，因此性能很難提高。另外，在測試環(huán)境中Hive系統(tǒng)采用TextFile和RCFile存儲格式對算法性能的影響有限，如圖5(a)所示，所以在Q4.1的測試中省略了RCFile的執(zhí)行時間。

對于某些聯(lián)接操作，不需要讀取測度列，因此進一步測試了星型聯(lián)接中去掉測度列的情況，測試結果如圖6所示。

圖6 不包括測度列

由于CC-MRSJ算法對測度列單獨處理和存儲，聯(lián)接中如果沒有測度列，算法不需要訪問測度列，從而需要更少的磁盤I/O，而Hive系統(tǒng)無論采用TextFile還是RCFile都需要訪問測度列，從而造成額外的I/O。因此，CC-MRSJ獲得了更高的加速比。

某些情況下，聯(lián)接查詢不包括WHERE查詢條件，于是測試了CC-MRSJ將Q3.1和Q4.1中的WHERE條件省略后的效率，測試結果如圖7所示。

在這種條件下，由于沒有過濾條件，算法需要讀取的數(shù)據(jù)量比包含WHERE條件的聯(lián)接操作要多，所以算法執(zhí)行的時間更長。但是CC-MRSJ通過數(shù)據(jù)劃分及相關性存儲和時延實例化，從而有效地減少了I/O，保證了算法執(zhí)行的高效率。因此，CC-MRSJ算法同樣優(yōu)于Hive算法。

5 結束語

圖7 不包括WHERE條件

本文初步探討了在MapReduce環(huán)境下利用合理的數(shù)據(jù)劃分和組織格式，提高星型聯(lián)接算法執(zhí)行效率，算法設計過程中也充分考慮了現(xiàn)代處理器的緩存特性。通過實驗比較，算法與Hive中的聯(lián)接算法相比，獲得了近似兩倍的性能提升。另外，現(xiàn)代處理器普遍具有多核和大內存，而目前的MapReduce系統(tǒng)對此資源利用不足，因此，如何進一步利用現(xiàn)代處理器特征，提高MapReduce算法的執(zhí)行效率，使MapReduce算法在橫向擴展的同時利用處理器的縱向擴展特征將是未來的研究方向之一。

1 Dean J，Ghemawat S.MapReduce:simplified data processing on large clusters.Communications of the ACM,2008(1)

2 Chang F,Dean J,Ghemawat S,et al.Bigtable:a distributed storage system for structured data.ACM Transactions on Computer Systems,2008(2)

3 Thusoo A,Sarma J S,Jain N,et al.Hive-a warehousing solution over a MapReduce framework.Proceedings of the VLDB Endowment,2009,2(2):1626~1629

4 Gates A,Natkovich O,Chopra S,et al.Srivastava,building a high level dataflow system on top of MapReduce:the pig experience.Proceedings of the VLDB Endowment,2009,2(2):1414~1425

5 Stonebraker M,Abadi D J,Batkin A,et al.C-store:a columnoriented dbms.Proceedings of the 31st International Conference on Very Large Data Bases,Trondheim,Norway,2005:553~564

6 Abadi D J,Madden S,Hachem N.Column-stores vs row-stores:how different are they really.Proceedings of the 2008 ACM SIGMOD International Conference on Management of Data,Vancouver,2008:967~980

7 Ailamaki A,DeWitt D J,Hill M D,et al.Weaving relations for cache performance.Proceedings of the 27th International Conference on Very Large Data Bases,Roma,2001:169~180

8 Lee R,Yin H,Zheng S,et al.RCFile:a fast and space-efficient data place-ment structure in MapReduce-based warehouse systems.ICDE 2011,Hannover,Germany:2011:1199~1208

9 Floratou A,Patel J M,Shekita E J,et al.Column-oriented storage techniques for MapReduce.Proceedings of the VLDB Endowment,2011(7)

10 Lin Y T,Agrawal D,Chen C,et al.Llama:leveraging columnar storage for scalable join processing in the MapReduce framework.Proceedings of the 2011 ACM SIGMOD International Conference on Management of Data,Athens,Greece,2011

11 Blanas S,Patel J M,Ercegovac V,et al.A comparison of join algorithms for log processing in mapreduce.Proceedings of the 2010 ACM SIGMOD International Conference on Management of Data Indiana,USA,2010:975~986

12 Han H,Jung H S,Eom H S,et al.Yeom:scatter-gather-merge:an efficient star-join query processing algorithm for data-parallel frameworks.Cluster Computing,2011,14(2):183~197

13 Rao J,Ross K A.Cache conscious indexing for decision-support in main memory.VLDB,Edinbargh,Scotland,1999:78~89

14 Brewer E A.Towards robust distributed systems.Proceedings of the Nineteenth Annual ACM Symposium on Principles of Distributed Computing,Portland,Oregon,2000

15 Zhang S B,Han J Z,Liu Z Y.Accelerating MapReduce with distributed memory cache.ICPADS 2009,Shenzhen,China,2009:472~478

16 Shinnar A,Cunningham D,Saraswat V,et al.M3R:increased performance for in-memory Hadoop jobs.Proceedings of the VLDB Endowment,2012(5)

17 O’Neil P,O’Neil E,Chen X.The star schema benchmark,http://www.cs.umb.edu/~poneil/star.SchemaB.PDF,Minneapdis,2007

18 Apache Hadoop.http://hadoop.apache.org/,2012

19 Lee R,Luo T,Huai Y,et al.YSmart:Yet another SQL-to-MapReduce translator.Proceedings of the 31st International Conference on Minneapolis,MN,USA,2011:25~36

20 Huai Y,Lee R,Zhang S,et al.A matrix model for analyzing,optimizing and deploying software for big data analytics in distributed systems.Proceedings of the 2nd ACM Symposium on Cloud Computing,Cascais,2011