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

孫 莉，何 剛，李繼云

?

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

孫 莉，何 剛，李繼云

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

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

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

1 概述

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

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

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

2 相關概念

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

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

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

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

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

圖1 產品銷售的星型模型

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

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

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

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

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

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

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

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

輸入

輸出

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

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

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

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

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

第5步輸出Dims。

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

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

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

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

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

第10步輸出。

第11步算法結束。

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

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

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

輸入和

輸出

//merge階段，用于合并2個Reduce端產生的相同粒度//的事實數據

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

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

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

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

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

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

第7步算法結束。

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

5 實驗與結果分析

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

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

表1 測試數據集行數統計

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

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

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

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

6 結束語

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

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

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

[3] 王 欣. 基于分布式ETL的電子政務決策系統設計和實現[D]. 上海: 復旦大學, 2012.

[4] 徐艷華, 郭朝珍. 基于MAS的分布式ETL模型[J]. 鄭州大學學報, 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－)，女，副教授、博士，主研方向：數據庫技術，面向對象分析與設計；何 剛，碩士研究生；李繼云，副教授、博士。

2013-09-02

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

10.3969/j.issn.1000-3428.2014.03.012