孫 莉，何 剛，李繼云

?

基于Hadoop平臺的事實并行處理算法

孫 莉，何 剛，李繼云

(東華大學(xué)計算機科學(xué)與技術(shù)學(xué)院，上海 201620)

針對傳統(tǒng)的抽取、轉(zhuǎn)換和加載工具在面臨數(shù)據(jù)倉庫中海量事實數(shù)據(jù)時效率較低的問題，從事實表查找代理鍵和多粒度事實預(yù)聚合2個角度出發(fā)，提出在漸變維度表上的多路并行查找算法和在不同粒度上對事實數(shù)據(jù)進行聚合的算法。第1種算法綜合考慮了漸變維度和大維度的情況，運用分布式緩存方法將小維度表復(fù)制到各個數(shù)據(jù)節(jié)點的內(nèi)存中，同時對事實數(shù)據(jù)和大維度數(shù)據(jù)采用相同的分區(qū)函數(shù)進行分區(qū)，從而解決內(nèi)存不足的問題，在Map階段實現(xiàn)多路查找代理鍵，避免由于數(shù)據(jù)傳輸產(chǎn)生的網(wǎng)絡(luò)延遲。第2種算法在Reduce階段之后增加Merge階段，可有效解決事實數(shù)據(jù)按照不同粒度進行聚合的問題。實驗結(jié)果表明，與Hive數(shù)據(jù)倉庫相比，2種算法在并行處理數(shù)據(jù)倉庫的事實數(shù)據(jù)的問題上具有更高的處理效率。

MapReduce模型；維度；事實；代理鍵；并行查找；聚合

1 概述

在數(shù)據(jù)倉庫[1]領(lǐng)域，數(shù)據(jù)抽取、轉(zhuǎn)換和加載(Extract, Transform and Load, ETL)過程負責(zé)從異構(gòu)的數(shù)據(jù)源收集數(shù)據(jù)，按照用戶定義的業(yè)務(wù)規(guī)則和需求，對收集的數(shù)據(jù)集進行各種轉(zhuǎn)換和清洗，最后加載到數(shù)據(jù)倉庫。然而，隨著數(shù)據(jù)量的不斷增長，越來越多的用戶希望用最短的時間將大量的業(yè)務(wù)數(shù)據(jù)通過ETL過程加載到數(shù)據(jù)倉庫中，以便盡可能快地為高層管理者提供決策服務(wù)。

文獻[2]提出的星型模型廣泛應(yīng)用于數(shù)據(jù)倉庫，星型模型是指中間一張事實表，周圍是與事實表通過主外鍵相連的維度表。由于事實的數(shù)據(jù)量較大，因此ETL過程是一個相當(dāng)耗時的工作。目前，對事實并行處理的研究主要集中于利用多線程的思想在單個CPU上運行ETL任務(wù)，對ETL過程的數(shù)據(jù)流采用分割、并行轉(zhuǎn)換和管道并行處理3個方面進行優(yōu)化，從而解決爭奪CPU資源的沖突[3]。然而，當(dāng)數(shù)據(jù)量較大、中間轉(zhuǎn)換邏輯復(fù)雜和數(shù)據(jù)源多樣時，這種方法往往很難保證負載均衡和進程之間不產(chǎn)生死鎖。文獻[4]提出的基于MAS的分布式ETL，利用AGENT的協(xié)作性、主動性、反應(yīng)性和交互性來構(gòu)建分布式ETL，從而改進了分布式的負載均衡問題。以上方法雖然在一定程度上提高了處理數(shù)據(jù)的效率，但是當(dāng)分布式處理上的節(jié)點之間通信和ETL任務(wù)調(diào)度出現(xiàn)故障時，恢復(fù)起來是相當(dāng)困難的，而且負載均衡也很難控制，甚至當(dāng)節(jié)點越來越多時，其網(wǎng)絡(luò)開銷也會越大，而且多個節(jié)點對同一個表的處理產(chǎn)生并發(fā)沖突的概率也會增加。

目前，一個可編寫和運行分布式應(yīng)用的處理大規(guī)模數(shù)據(jù)的開源框架——Hadoop[5-7]的興起，吸引了眾多用戶的關(guān)注，他們將部署在昂貴的服務(wù)器上的應(yīng)用程序遷移到價格低廉的商品機集群中進行各種各樣的分布式應(yīng)用，這不僅為企業(yè)解決大數(shù)據(jù)問題提供了便利性，而且還為企業(yè)節(jié)省了大量成本。文獻[8-10]給出了在Hadoop平臺下進行表之間連接的相關(guān)研究，可以有效地解決在Hadoop平臺下表之間的連接操作。因此，為了能夠提高數(shù)據(jù)倉庫中事實處理的效率，本文采用Hadoop作為ETL的執(zhí)行平臺，基于Map- Reduce-Merge框架[11]，考慮漸變維度表和大維度表的情況，針對事實表查找代理鍵和事實數(shù)據(jù)按照不同粒度聚合的問題分別提出相應(yīng)的算法。

2 相關(guān)概念

Google提出的MapReduce[5]是一個用于處理和生成大數(shù)據(jù)集編程模型。它是基于集群計算的體系結(jié)構(gòu)，用于處理密集型數(shù)據(jù)的并行計算范式，是基于Hadoop框架的一種通用編程模型。該編程模型主要是基于2個可編程的函數(shù)：

用戶編寫的函數(shù)有2個輸入變量：鍵1和值1，函數(shù)輸出的是中間結(jié)果的鍵值對[(2,2)]列表，這些鍵值對列表將由MapReduce類庫中的分區(qū)函數(shù)按照鍵2進行分區(qū)，同一個鍵2的值列表將屬于同一個分組。另外，函數(shù)同樣需要由用戶編寫，該函數(shù)有2個輸入變量：中間鍵2和中間值列表[2]，輸出為值列表[3]。

Map-Reduce-Merge框架的和函數(shù)和MapReduce框架是類似的，兩者主要的區(qū)別在于函數(shù)是輸出鍵值對列表，而不是值列表。最后加入的階段是為了合并輸出的鍵值對列表。

3 漸變維度表上的多路并行查找算法

為了跟蹤維度的變化情況，文獻[2]在維度建模中提出的漸變維度主要有2種類型，即類型1和類型2。類型1的漸變維度采用直接更新的方法，不需要記錄維度的歷史變化；類型2的漸變維度采用更新-插入的方法，此類型的漸變維度需要添加2個時間戳字段和一個標(biāo)識字段，其中， 2個時間戳字段分別表示維度的開始生效時間和失效時間，標(biāo)識字段表示維度是否為當(dāng)前正在使用。圖1為含有類型1和2的漸變維度的星型模型。

圖1 產(chǎn)品銷售的星型模型

從圖1可以看出，商店store是類型1的漸變維度，產(chǎn)品product是類型2的漸變維度(因為可以通過validatefrom和validateto 2個時間戳字段記錄產(chǎn)品維度的歷史變化情況)，事實表與維度表之間按照主外鍵的關(guān)系進行關(guān)聯(lián)，事實表的主鍵是由各個維度表的代理鍵復(fù)合而成的。

本文提出的在漸變維度表上的多路并行查找算法(MPLK-SCD)主要考慮了2種情況：

(1)當(dāng)漸變維度表的數(shù)據(jù)量比較小時，可以完全裝入到內(nèi)存中，首先在MapReduce作業(yè)啟動之前將類型1的漸變維度表的自然鍵和代理鍵或者類型2的漸變維度表的自然鍵、代理鍵和2個時間戳字段存儲到每一個Tasktracker的內(nèi)存中，然后調(diào)用setup函數(shù)(見算法1中第1步~第5步)，從本地緩存中將維度表讀取出來，只需要在Map階段(見算法1中第6步~第11步)處理每一行事實數(shù)據(jù)時，將事實記錄與剛才從內(nèi)存中讀取的維度表關(guān)聯(lián)查找出相應(yīng)的代理鍵。如圖2所示，如果為類型1的漸變維度表，則只需要將維度表的自然鍵store_no和事實表的自然鍵store_no關(guān)聯(lián)；如果為類型2的漸變維度表，則不僅需要將維度表的自然鍵product_no與事實表的自然鍵product_no關(guān)聯(lián)，而且還需要將事實記錄的事務(wù)日期order_ date與維度的2個時間戳字段validatefrom和validateto進行匹配，從而查找出正確的代理鍵。

圖2 2類漸變維度中代理鍵的查找

(2)當(dāng)漸變維度表的數(shù)據(jù)量比較大時，導(dǎo)致其無法完全被存儲在主存中，那么為了解決該問題，本文引入分區(qū)的方法。該方法的主要思想是將漸變維度表與事實表按照相同的分區(qū)函數(shù)進行分區(qū)，經(jīng)過分區(qū)后，自然鍵相同的維度和事實數(shù)據(jù)將會出現(xiàn)在同一個分區(qū)中。如圖3所示，假設(shè)MapReduce作業(yè)含有個map任務(wù)，事實數(shù)據(jù)按照分區(qū)函數(shù)分成1,2, …,，漸變維度數(shù)據(jù)按照相同的分區(qū)函數(shù)分成1,2 ,…,，緩存中的維度統(tǒng)稱為，其中每一個map任務(wù)分別讀取映射在相同分區(qū)的事實數(shù)據(jù)和漸變維度數(shù)據(jù)，而且在每一個tasktracker的主存中均含有小維度數(shù)據(jù)的復(fù)本，接下來就可以在map階段完成多路查找過程。

圖3 基于分區(qū)方法在漸變維度上的多路并行查找過程

因此，針對以上2種情況均可以在Map階段完成代理鍵的并行查找，從而消除在reduce階段之前由于數(shù)據(jù)遷移產(chǎn)生的網(wǎng)絡(luò)延遲。在漸變維度表上的多路并行查找算法見算法1。

算法1在漸變維度表上的多路并行查找算法

輸入

輸出

//setup階段，用于獲取本地緩存漸變維度

第1步初始化維度數(shù)據(jù)集Dims=F，同時從本地緩存中獲取漸變維度數(shù)據(jù)集CacheDims，跳至第2步。

第2步如果CacheDims是類型2的漸變維度，則跳至第3步，否則跳至第4步。

第3步如果CacheDims未遍歷結(jié)束，則從中讀取一行記錄，記為Dim，從Dim獲取自然鍵NK、代理鍵SK、維度開始生效時間ST和維度失效時間ET，并存入Dims數(shù)據(jù)集中，繼續(xù)第3步，否則跳至第5步。

第4步如果CacheDims未遍歷結(jié)束，則從中讀取一行記錄，記為Dim，從Dim獲取自然鍵NK和代理鍵SK，并存入Dims數(shù)據(jù)集中，繼續(xù)第4步，否則跳至第5步。

第5步輸出Dims。

//map階段，用于查找漸變維度的代理鍵

第6步如果value不為空，則跳至第7步，否則跳至第11步。

第7步如果Dims為類型2的漸變維度，跳至第8步，否則跳至第9步。

第8步遍歷Dims，按照圖2，將value中相應(yīng)的字段和NK、ST和ET進行匹配，查找出正確的SK，將SK作為key'，value中的度量值作為value'，跳至第10步。

第9步遍歷Dims，按照圖2，將value中相應(yīng)的字段和NK進行匹配，查找出正確的SK，將SK作為key'，value中的度量值作為value'，跳至第10步。

第10步輸出。

第11步算法結(jié)束。

4 不同粒度上聚合事實的算法

在數(shù)據(jù)倉庫的OLAP應(yīng)用中，在不同粒度上對事實表的聚合是一種十分常見的操作。在Hadoop平臺下，可以直接使用Hive的類SQL語句實現(xiàn)不同粒度的聚合，然而Hive的類SQL語句最終會轉(zhuǎn)化為map和reduce任務(wù)去執(zhí)行，因為每一次在某個粒度上聚合事實數(shù)據(jù)時都會造成重大的開銷，而且Hive無法一次性實現(xiàn)多粒度的聚合，所以為了提高在不同粒度上的查詢響應(yīng)時間，將不同粒度上的事實數(shù)據(jù)一次性聚合后存儲到Hive中。利用Map-Reduce-Merge框架的思想，在Reduce階段之后加入了Merge階段(見算法2)。

算法2在不同的粒度上聚合事實的算法(Agg-MPM)

輸入和

輸出

//merge階段，用于合并2個Reduce端產(chǎn)生的相同粒度//的事實數(shù)據(jù)

第1步開始遍歷value1和value2 2個數(shù)據(jù)集，初始化度量值sum為0。

第2步如果value1和value2 2個數(shù)據(jù)集均為空，則跳至第7步；否則跳至第3步。

第3步如果value2數(shù)據(jù)集為空且value1數(shù)據(jù)集不為空，則從value1中獲取一個值設(shè)為left，同時計算sum=sum+left，繼續(xù)第3步；否則跳至第4步。

第4步如果value2數(shù)據(jù)集不為空且value1數(shù)據(jù)集為空，則從value2中獲取一個值設(shè)為right，同時計算sum = sum + right，繼續(xù)第4步；否則跳至第5步。

第5步如果value1和value2 2個數(shù)據(jù)集均不為空，則分別從其中獲取一個值，設(shè)為left和right，如果left和right的代理鍵相等，則計算sum=sum+left+ right，跳至第2步；否則跳至第6步。

第6步將當(dāng)前粒度加一后作為key，代理鍵和sum作為value，輸出；跳至第第7步。

第7步算法結(jié)束。

比如時間維度含有年-半年-季度-月-日的層次結(jié)構(gòu)，現(xiàn)需要按該層次結(jié)構(gòu)聚合成不同粒度的事實數(shù)據(jù)，首先通過map和reduce階段的處理之后將輸出以日作為鍵，其他字段作為值的鍵值列表，然后通過Merge階段合并這些鍵值對，可以將粒度為日的事實合并成粒度為月的事實，同樣，將粒度為月的事實合并成粒度為季度的事實，依此類推，最終可以合并成粒度為年的事實。在不同的粒度上聚合事實的算法見算法2。

5 實驗與結(jié)果分析

本文實驗采用TPC-H[12]生成的數(shù)據(jù)集在Hadoop分布式集群平臺上進行仿真測試，從算法的效率進行了驗證分析。首先搭建了Hadoop集群，該集群由7臺PC機組成。其中，1臺是NameNode節(jié)點；6臺作為DataNode節(jié)點。另外，在7臺PC機中均安裝了Ubuntu12.04、Hadoop1.0.3和JDK6，實驗平臺采用Eclipse3.7.2作為集群開發(fā)工具。

本文實驗采用表1中的4類數(shù)據(jù)集進行測試，其中，漸變維度表為Supplier；事實表為LineItem和時間維度表為Dimdate。4類測試數(shù)據(jù)集的行數(shù)統(tǒng)計見表1。

表1 測試數(shù)據(jù)集行數(shù)統(tǒng)計

從圖4可以看出，在漸變維度上處理事實的過程中，雖然2種算法隨著數(shù)據(jù)量的增加，運行時間也在不斷的增加，但是本文提出的算法MPLK-SCD要比Hive處理效率要高，因為只在map端查找維度鍵可以避免在reduce端之前由于數(shù)據(jù)遷移產(chǎn)生的網(wǎng)絡(luò)延遲，而且當(dāng)維度數(shù)據(jù)量越來越大時，Hive無法將維度緩存在各個tasktracker節(jié)點上，因此Hive必然要通過map-reduce 2個階段進行維度鍵的查找。然而，該過程可以進一步優(yōu)化，即一開始就對維度表采用垂直分區(qū)的方法，過濾掉一些不必要的數(shù)據(jù)，可為其他操作提供更多的內(nèi)存空間，如果是類型1的維度，則只需要考慮維度表的代理鍵和自然鍵兩列；如果是類型2的維度，除了代理鍵和自然鍵以外，還需要考慮維度的有效時間段。

圖4 漸變維度處理事實所用時間

從圖5可以看出，在不同粒度上聚合事實的過程中，當(dāng)數(shù)據(jù)量越來越大時，本文提出的算法Agg-MPM和Hive的處理時間都在不斷的增加，但是由于Hive通過map- reduce2個階段聚合時無法一次性實現(xiàn)多粒度的聚合，導(dǎo)致磁盤I/O增加，而Agg-MPM利用低粒度的中間結(jié)果向高粒度聚合時，可以減少數(shù)據(jù)量的遷移，因此Agg-MPM算法的處理效率要優(yōu)于Hive。

圖5 不同粒度聚合事實所用時間

6 結(jié)束語

在目前大數(shù)據(jù)時代，并行處理數(shù)據(jù)倉庫中的海量事實數(shù)據(jù)是當(dāng)前熱門的研究方向。本文提出的基于Hadoop平臺的2個并行處理事實的算法，即在漸變維度表上的多路并行查找算法和在不同粒度上聚合事實的算法，可以有效地解決數(shù)據(jù)倉庫中海量事實數(shù)據(jù)的處理問題。通過實驗結(jié)果表明，與Hive相比，MPLK-SCD算法和Agg-MPM算法，具有更高的處理效率，可以及時地為用戶提供決策支持。下一步將研究事實的增量聚合和Merge階段合并時選擇并行性更高的工作流等問題。

[1] Inmon W H. Building the Data Warehouse[M]. Indianapolis, USA: Wiley, 2005: 33-43.

[2] 譚明金. 數(shù)據(jù)倉庫工具箱: 維度建模的完全指南[M]. 北京:電子工業(yè)出版社, 2003.

[3] 王 欣. 基于分布式ETL的電子政務(wù)決策系統(tǒng)設(shè)計和實現(xiàn)[D]. 上海: 復(fù)旦大學(xué), 2012.

[4] 徐艷華, 郭朝珍. 基于MAS的分布式ETL模型[J]. 鄭州大學(xué)學(xué)報, 2007, 39(4): 118-121.

[5] Dean J, Ghemawat S. MapReduce: Simplified Data Processing on Large Clusters[C]//Proc. of the 6th Symposium on Operating System Design and Implementation. Berkeley, USA: [s. n.], 2004: 137-150.

[6] Ghemawat S, Gobioff H. The Google File System[C]//Proc. of the 19th ACM Symposium on Operating Systems Principles. New York, USA: ACM Press, 2003: 29- 43.

[7] White T. Cluster Specification Hadoop: The Definitive Guide[M]. [S. l.]: O’Reilly Media, 2009: 255-259.

[8] Wang Yuxiang, Song Aibo, Luo Junzhou. A MapReduce Merge-based Data Cube Construction Method[C]//Proc. of the 9th International Conference on Grid and Cloud Computing. Washington D. C., USA: IEEE Computer Society, 2010: 1-6.

[9] Liu Xiufeng, Thomsen C, Pedersen T B. CloudETL Scalable Dimensional ETL for Hadoop and Hive[D]. Aalborg, Denmark: Aalborg University, 2012.

[10] Miner D, Shook A. MapReduce Design Patterns[M]. [S. l.]:O’Reilly Media, 2013: 103-123.

[11] Yang H C, Dasdan A, Hsiao R L, et al. Map-Reduce-Merge Simplified Relational Data Processing on Large Clusters[C]// Proc. of ACM SIGMOD International Conference on Management of Data. New York, USA: ACM Press, 2007: 1029-1040.

[12] TPC-H. Homepage[EBOL]. [2013-05-07]. http://www.tpc.org/ tpch/default.asp.

編輯 索書志

Parallel Processing Algorithms for Facts Based on Hadoop Platform

SUN Li, HE Gang, LI Ji-yun

(School of Computer Science and Technology, Donghua University, Shanghai 201620, China)

In view of that traditional Extract, Transform, Load(ETL) tools face the efficient problem of the massive fact data in data warehouse, two algorithms about parallel processing facts are designed and implemented based on Hadoop platform. From the two perspectives of surrogate key lookup of fact table and aggregation for fact data on the different granularity, a multi-way parallel lookup algorithm on slowly changing dimensions and an algorithm of aggregation for fact data on the different granularity are presented. The first algorithm considers slowly changing dimensions and big dimensions synthetically. In order to solve the problem of out of memory, the algorithm adopts an approach to the distributed cache to copy small dimensions to every date nodes’ memory. And implementing multi-way lookup of dimension keys in the stage of map is to avoid network delay result from data transmission. The second algorithm adds merge stage after reducing stage, so it is beneficial to solve the aggregation problem of the fact data according to different granularity effectively. Experimental results show that the two algorithms have better efficient than Hive data warehouse with respect to the problem of parallel processing facts data in data warehouse.

MapReduce model; dimension; fact; surrogate key; parallel lookup; aggregation

1000-3428(2014)03-0059-04

A

TP311

孫 莉(1964－)，女，副教授、博士，主研方向：數(shù)據(jù)庫技術(shù)，面向?qū)ο蠓治雠c設(shè)計；何 剛，碩士研究生；李繼云，副教授、博士。

2013-09-02

2013-10-25 E-mail：sli@dhu.edu.cn

10.3969/j.issn.1000-3428.2014.03.012