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

周國亮，朱永利，王桂蘭

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

1 引言

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

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

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

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

2 相關工作

2.1 MapReduce及其數據存儲格式

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

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

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

2.2 聯接算法

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

（1）map-reduce聯接

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

（2）map 聯接

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

（3）分塊的 map 聯接

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

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

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

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

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

3 C C-MRSJ算法

3.1 數據存儲格式

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

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

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

3.2 算法描述

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

圖1 維表的列簇存儲模式

圖2 星型模型存儲模式

圖3 星型聯接執行計劃

可以用式（1）表示：

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

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

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

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

下面以SSB數據集中的Q3.1為例，說明算法的執行過程，Q3.1涉及多個查詢條件，是一個事實表和3個維表的聯接。只考慮聯接操作，因此去掉了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;

詳細的執行過程如圖4所示。

圖4 星型聯接的例子

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

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

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

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

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

輸入：

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

3.3 算法分析

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

其中，p表示代價，k表示比例因數。

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

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

4 實驗結果與分析

4.1 實驗環境

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

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

實驗室搭建了包括6個節點的MapReduce測試環境，主節點配置的機器是DELLPowerEdgeT310，使用Intel XeonCPUX3430@2.40GHz處理器和2GBRAM，系統LinuxUbantu版本。數據節點是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個維表的聯接操作。

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

4.2 實驗結果

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

圖5 原始查詢

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

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

圖6 不包括測度列

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

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

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

5 結束語

圖7 不包括WHERE條件

本文初步探討了在MapReduce環境下利用合理的數據劃分和組織格式，提高星型聯接算法執行效率，算法設計過程中也充分考慮了現代處理器的緩存特性。通過實驗比較，算法與Hive中的聯接算法相比，獲得了近似兩倍的性能提升。另外，現代處理器普遍具有多核和大內存，而目前的MapReduce系統對此資源利用不足，因此，如何進一步利用現代處理器特征，提高MapReduce算法的執行效率，使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