幾種并行編程框架在數(shù)據(jù)挖掘領(lǐng)域的比較

何淵淘， 齊兵輝

(鄭州航空工業(yè)管理學(xué)院， 鄭州 450000)

摘要：將機(jī)器學(xué)習(xí)并行化是進(jìn)行海量數(shù)據(jù)挖掘的重要方式，但由于并行計(jì)算框架、機(jī)器學(xué)習(xí)算法的多樣性，導(dǎo)致計(jì)算框架的選取及算法并行化存在著困難。本文對(duì)幾種常見的并行計(jì)算框架的模型結(jié)構(gòu)和工作機(jī)理進(jìn)行了分析，根據(jù)算法中變量的依存關(guān)系將其分類，并將這幾類算法進(jìn)行了實(shí)驗(yàn)對(duì)比。實(shí)驗(yàn)結(jié)果表明，算法中變量的依存關(guān)系對(duì)其在并行化后的性能有巨大的影響。

關(guān)鍵詞：MapReduce; Pregel; Hama; GraphLab; MPI; 數(shù)據(jù)挖掘

中圖分類號(hào)：TP3

文獻(xiàn)標(biāo)志碼：A

DOI:10.3969/j.issn.1671-6906.2015.03.021

Abstract:To slove the date mining on large dataset, the parallelizaion of algorithm is the most important solution. Due to the diversity of the parallel frameworks and the machine learning algorithms, it is difficult to choose a framework and algorithm parallelizaion. In this paper, the models and mechanism of the parallel framework are analyzed, and it is classified based on the parameter relations. In the end, the experiments are conducted and the results show that the relation of the algorithm parameters have great impact on the performance

由于傳感技術(shù)和通信網(wǎng)絡(luò)的發(fā)展，數(shù)據(jù)收集和存儲(chǔ)的規(guī)模在飛速增大，如何從海量數(shù)據(jù)中挖掘出有效的信息是當(dāng)前的研究熱點(diǎn)。目前較為普遍的方案是采用機(jī)群系統(tǒng)和分布式框架來提高數(shù)據(jù)處理的效率[1-2]。由于數(shù)據(jù)挖掘算法本身的差異，導(dǎo)致其在不同的并行框架下有著顯著的性能差異。本文在常見的并行框架下比較了幾種數(shù)據(jù)挖掘算法，并根據(jù)實(shí)驗(yàn)結(jié)果分析了不同算法在并行計(jì)算框架下的適用性，為機(jī)器學(xué)習(xí)算法的并行化提供了實(shí)驗(yàn)依據(jù)。

1常見的并行計(jì)算框架

海量數(shù)據(jù)處理的關(guān)鍵在于把問題分解為“映射”和“規(guī)約”兩種操作，“映射”將數(shù)據(jù)集進(jìn)行分割和處理，而“規(guī)約”則將“映射”后的結(jié)果進(jìn)行整理和歸納。并行計(jì)算框架將并行節(jié)點(diǎn)中復(fù)雜的“通訊”“同步”工作進(jìn)行了封裝，極大降低了數(shù)據(jù)處理的難度。并行框架的模型分為三類：基于消息傳遞的模型、基于數(shù)據(jù)流的模型、基于圖的模型[2]。常見的并行計(jì)算框架和其特征見表1，其中MPI屬于基于消息傳遞的模型，MapReduce、Twister、Haloop屬于基于數(shù)據(jù)流的模型，Pregel、Hama、GraphLab屬于基于圖的模型。

表1　幾種分布式計(jì)算框架的對(duì)比

1.1MPI

MPI(Message Passing Interface)是消息傳遞接口的標(biāo)準(zhǔn)規(guī)范，最初由MPI論壇發(fā)布，支持C和Fortran語言。嚴(yán)格來說，MPI是一個(gè)函數(shù)庫，僅提供了對(duì)并行計(jì)算最基礎(chǔ)的支持。MPI的一個(gè)極大優(yōu)勢是使用MPI編寫的程序具有較好的移植性，同時(shí)也有著最高的分布式計(jì)算效率。當(dāng)前基于MPI實(shí)現(xiàn)的函數(shù)庫主要有OpenMPI和MPICH。

基于MPI的應(yīng)用通常由多個(gè)并發(fā)進(jìn)程組成，每個(gè)進(jìn)程位于機(jī)群系統(tǒng)的某臺(tái)主機(jī)上，并由唯一的標(biāo)示符來標(biāo)記。每個(gè)MPI進(jìn)程完成整個(gè)分布式計(jì)算任務(wù)的一部分，并通過消息機(jī)制完成并行進(jìn)程間數(shù)據(jù)的交換。使用MPI框架可以實(shí)現(xiàn)最高效的算法，但是在并行計(jì)算的過程中會(huì)遇到數(shù)據(jù)存儲(chǔ)、切分等問題，同時(shí)也要解決并發(fā)進(jìn)程間的同步、競爭等一系列問題，屬于細(xì)粒度的并行開發(fā)。這種特征使得基于MPI的數(shù)據(jù)挖掘?qū)θ说囊筝^高，算法實(shí)現(xiàn)周期長。同時(shí)MPI的應(yīng)用與機(jī)群規(guī)模關(guān)系緊密，當(dāng)機(jī)群規(guī)模發(fā)生變化后，還需要重新調(diào)整原有的程序，這不適合彈性變化的云環(huán)境。

1.2MapReduce和迭代式MapReduce

MapReduce是谷歌公司為了解決海量數(shù)據(jù)挖掘問題而設(shè)計(jì)的并行計(jì)算框架， 該框架通過Map和Reduce兩個(gè)步驟[2]完成分布式計(jì)算的“映射”和“規(guī)約”操作。海量數(shù)據(jù)通常存儲(chǔ)在分布式文件系統(tǒng)上，該框架將其分割后交給若干個(gè)Map來處理，每個(gè)并發(fā)的Map進(jìn)行本地計(jì)算后將結(jié)果輸出為<鍵/值>[1-2]的形式。Map處理完成后，系統(tǒng)會(huì)將這些二元組序列進(jìn)行排序。具有相同“鍵”的元組被匯總后交由Reduce匯總，最終數(shù)據(jù)被輸出并保存在分布式文件系統(tǒng)上。

MapReduce框架的計(jì)算流程如圖1所示。從圖可以看出，該框架適合數(shù)據(jù)集內(nèi)數(shù)據(jù)關(guān)聯(lián)性弱，數(shù)據(jù)挖掘算法較為簡單，但數(shù)據(jù)量較為龐大的一類問題[3]。這些問題在自然語言處理、生物信息學(xué)等領(lǐng)域較為普遍。數(shù)據(jù)分析人員僅需要將注意力集中在Map和Reduce的設(shè)計(jì)上，而數(shù)據(jù)存儲(chǔ)、分割和計(jì)算的同步則由框架本身的實(shí)現(xiàn)去完成。當(dāng)前谷歌和雅虎的MapReduce及微軟的Dryad都是針對(duì)該框架的，其中谷歌使用該框架重新實(shí)現(xiàn)了搜索引擎業(yè)務(wù)，使得程序的結(jié)構(gòu)更為簡潔，性能更為穩(wěn)定[2]。

圖1　MapReduce處理流程

然而在統(tǒng)計(jì)類數(shù)據(jù)挖掘領(lǐng)域中，MapReduce對(duì)數(shù)據(jù)間獨(dú)立性假設(shè)的條件難以保證[3]，這類問題的求解需要進(jìn)行多次的Map和Reduce操作，即迭代式MapReduce。而Mahout[3]就是針對(duì)迭代式MapReduce設(shè)計(jì)的框架。

1.3改進(jìn)的迭代式MapReduce

迭代式MapReduce在海量數(shù)據(jù)挖掘領(lǐng)域中有較多的應(yīng)用，例如商品推薦系統(tǒng)[3]。然而在迭代過程中，Map和Reduce會(huì)頻繁進(jìn)行序列化和反序列化操作，這些操作導(dǎo)致了較高的輸入、輸出開銷?；贛ahout框架的數(shù)據(jù)挖掘應(yīng)用就面臨了類似的問題[3]。Ekanayake J等指出，當(dāng)前很多數(shù)據(jù)集規(guī)模小于機(jī)群環(huán)境中的內(nèi)存總量[3]，因此可以將全部數(shù)據(jù)存放在內(nèi)存中以避免序列化和反序列化操作。由此產(chǎn)生了對(duì)迭代式MapReduce的改進(jìn)，其代表為Twister和Haloop[4-7]。

Twister和Haloop將迭代過程中的數(shù)據(jù)分為靜態(tài)和動(dòng)態(tài)兩種類型。靜態(tài)數(shù)據(jù)持久存放在機(jī)群的內(nèi)存中，而少量動(dòng)態(tài)數(shù)據(jù)采用NarradaBrokering[3-4]消息總線進(jìn)行傳輸。這兩種策略極大減少了序列化和反序列化的開銷，顯著提升了算法的運(yùn)行效率。圖2和圖3為Twister與其他幾種并行框架在商品推薦算法和K-means算法上的性能對(duì)比。從圖中可以看出，兩種算法在Twister下的開銷比在Hadoop和Dryad下低幾個(gè)數(shù)量級(jí)，與在MPI下的開銷較為接近。

圖2　商品推薦算法在3種并行框架下的性能對(duì)比

圖3　K-means算法在4種并行框架下的性能對(duì)比

然而Twister和Haloop對(duì)輸入數(shù)據(jù)有特殊要求，即用戶需要提前進(jìn)行數(shù)據(jù)的切分。同時(shí)Twister和Haloop沒有提供任何的容錯(cuò)機(jī)制，一旦某個(gè)分布式計(jì)算進(jìn)程出現(xiàn)錯(cuò)誤，整個(gè)計(jì)算任務(wù)就必須重新開始。這種問題在使用廉價(jià)計(jì)算機(jī)搭建的云環(huán)境下更為致命。

1.4基于BSP的Pregel和Hama

Pregel和Hama是基于BSP的圖計(jì)算框架，其中Pregel是谷歌針對(duì)大數(shù)據(jù)下的圖遍歷、最小生成樹、最短路徑等而設(shè)計(jì)的[8-9]。這類框架的典型應(yīng)用是網(wǎng)頁排名和社交網(wǎng)絡(luò)中的人際關(guān)系數(shù)據(jù)挖掘。

當(dāng)前大部分?jǐn)?shù)據(jù)挖掘算法可以轉(zhuǎn)換為圖的結(jié)構(gòu)。以圖4中多元素相加為例，算法中的變量可以轉(zhuǎn)變?yōu)閳D中的點(diǎn)，而變量之間的運(yùn)算關(guān)系可以轉(zhuǎn)變?yōu)閳D中的邊。用上述方式可以將大部分統(tǒng)計(jì)機(jī)器學(xué)習(xí)算法用圖來描述，進(jìn)而在Pregel和Hama[9]下進(jìn)行實(shí)現(xiàn)。圖5為K-means算法在Mahout和Hama兩種并行框架下的性能對(duì)比。從圖5可以看出，由于K-means需要進(jìn)行高頻率的數(shù)據(jù)傳遞，基于Mahout的并行框架時(shí)間開銷較高。

圖4　算法內(nèi)變量依賴關(guān)系的圖形化表示

圖5　iris數(shù)據(jù)集上K-means算法的性能對(duì)比

從結(jié)構(gòu)來看，Pregel和Hama的基礎(chǔ)為BSP(Bulk Synchronous Parallel)[9]。BSP由超步(Superstep)組成，超步的結(jié)構(gòu)如圖6所示。每個(gè)超步包含了本地計(jì)算、節(jié)點(diǎn)間通信、同步3個(gè)過程。一個(gè)超步通常由多個(gè)并發(fā)的本地計(jì)算組成，每個(gè)本地計(jì)算位于機(jī)群中的一個(gè)計(jì)算機(jī)節(jié)點(diǎn)上，這些本地計(jì)算進(jìn)程使用機(jī)群間的網(wǎng)絡(luò)完成通信和同步工作。

圖6　BSP的邏輯結(jié)構(gòu)

由于本地計(jì)算進(jìn)程的數(shù)量遠(yuǎn)大于機(jī)群中主機(jī)的個(gè)數(shù)，因此多個(gè)本地計(jì)算進(jìn)程共享同一臺(tái)主機(jī)。而超步中每個(gè)節(jié)點(diǎn)上的本地運(yùn)算所耗費(fèi)的時(shí)間不等，因此同步機(jī)制導(dǎo)致大量節(jié)點(diǎn)處于等待狀態(tài)，這使得基于BSP框架的算法有著較高的時(shí)間開銷。除此之外，超步內(nèi)的本地進(jìn)程需要讀入初始數(shù)據(jù)，而這些初始數(shù)據(jù)通常存儲(chǔ)在主機(jī)節(jié)點(diǎn)上；如果初始數(shù)據(jù)在這些計(jì)算節(jié)點(diǎn)間分配不合理，僅在數(shù)據(jù)讀取這個(gè)階段就會(huì)有很長的等待時(shí)間，這會(huì)造成更為嚴(yán)重的同步等待現(xiàn)象。

1.5GraphLab的異步圖計(jì)算

GraphLab是CMU[10]針對(duì)大數(shù)據(jù)環(huán)境下的圖數(shù)據(jù)挖掘提出的框架，該框架擴(kuò)展了BSP對(duì)異步的支持，同時(shí)也更適合統(tǒng)計(jì)類數(shù)據(jù)挖掘算法[10-13]。由于GraphLab使用共享內(nèi)存的方式在節(jié)點(diǎn)間進(jìn)行被動(dòng)的信息傳遞，這種方式避免了在Hama等框架下無效的數(shù)據(jù)傳輸[10]，為高效的異步計(jì)算和通訊提供了支持。

GraphLab使用了基于圖的模型，因此可以將統(tǒng)計(jì)類算法轉(zhuǎn)變?yōu)橛邢驘o環(huán)圖來求解[11]。該框架用節(jié)點(diǎn)代表算法中的變量和數(shù)據(jù)，用邊來表示數(shù)據(jù)的依賴關(guān)系,將并行算法執(zhí)行過程中的數(shù)據(jù)傳遞動(dòng)作抽象成Gather、Apply、Scatter[10]3個(gè)操作。為了避免本地計(jì)算進(jìn)程在數(shù)據(jù)讀寫上的“競爭條件”和“同步問題”，GraphLab使用了3種一致性模型:“節(jié)點(diǎn)一致”模型、“邊一致”模型和“完全一致”模型[10]。3種一致性模型按數(shù)據(jù)的讀寫順序?qū)⒄麄€(gè)圖分割成若干子圖，并在不同的子圖上執(zhí)行并行計(jì)算，這種策略避免了數(shù)據(jù)一致性問題。3種一致性模型的差別在于所劃分子圖中邊和節(jié)點(diǎn)的數(shù)量，以及對(duì)并行計(jì)算的支持度[10]。

圖7和圖8為GraphLab和其他幾種并行框架下Netflix電影推薦算法和名稱實(shí)體算法的性能對(duì)比。從圖7和圖8可以看出，GraphLab對(duì)異步計(jì)算的支持和共享內(nèi)存的消息傳遞方式使得其對(duì)節(jié)點(diǎn)數(shù)量的依賴性較低，其與其他兩種框架相比具有極低的時(shí)間復(fù)雜度。

圖7　Netflix電影推薦算法在3種并行框架下的性能對(duì)比

圖8　名稱實(shí)體識(shí)別算法在3種并行框架下的性能對(duì)比

2幾種框架的比較和性能分析

2.1MPI與其他幾種編程框架的對(duì)比

MPI使用消息傳遞函數(shù)實(shí)現(xiàn)不同計(jì)算節(jié)點(diǎn)間的數(shù)據(jù)傳遞。與其他幾類分布式框架相比，其抽象程度最低，因而數(shù)據(jù)分析人員面臨的開發(fā)難度最大。使用MPI可以完成在其他并行框架下實(shí)現(xiàn)的任何算法，理論上來說基于MPI的算法有著最高的性能。然而由于人的因素，MPI絕非任何場景下的最優(yōu)選擇。從圖2、圖3、圖7、圖8可得出，盡管MapReduce、BSP、GraphLab抽象程度較高，但算法的性能依然接近MPI下的性能表現(xiàn)。

2.2MapReduce和BSP的差別

MapReduce的抽象程度要高于BSP，因而在該框架下算法的實(shí)現(xiàn)難度小于BSP。Low Y等指出，任何在MapReduce下實(shí)現(xiàn)的算法都可以在BSP框架下實(shí)現(xiàn)，而且有著相近或者是更高的運(yùn)行效率[10]。對(duì)數(shù)據(jù)集內(nèi)數(shù)據(jù)依賴性強(qiáng)、數(shù)據(jù)處理需要迭代求解的問題，MapReduce性能較差。而BSP避免了序列化和反序列化操作，相比于MapReduce時(shí)間開銷較低。從圖7、圖8可以看出，基于BSP的GraphLab相比于MapReduce(hadoop)有著極低的運(yùn)算開銷。Gonzalez J E等也指出，基于BSP的框架更適合圖遍歷和最短路徑樹等算法[12]。

當(dāng)前統(tǒng)計(jì)類數(shù)據(jù)挖掘算法在大數(shù)據(jù)領(lǐng)域有較多應(yīng)用，而大部分統(tǒng)計(jì)類機(jī)器學(xué)習(xí)算法可以抽象為算法中變量的依存關(guān)系，這種關(guān)系可以轉(zhuǎn)化為有向無環(huán)圖，從而在基于BSP的框架下實(shí)現(xiàn)。

2.3GraphLab和BSP的差別

基于BSP的Pregel和Hama僅支持同步計(jì)算，然而超步中的等待機(jī)制導(dǎo)致大量節(jié)點(diǎn)處于等待狀態(tài)，從而造成計(jì)算資源的浪費(fèi)。Corbett J C等指出，大部分統(tǒng)計(jì)類的數(shù)據(jù)挖掘算法具有較強(qiáng)的數(shù)據(jù)依賴性和變量依賴性[13]，如果使用Hama等框架來進(jìn)行處理會(huì)出現(xiàn)超步過多和同步時(shí)間過長的現(xiàn)象。GraphLab不僅支持異步計(jì)算，也支持節(jié)點(diǎn)上的動(dòng)態(tài)調(diào)度。同時(shí)其采用了節(jié)點(diǎn)動(dòng)態(tài)優(yōu)先級(jí)調(diào)度和共享內(nèi)存方式傳遞數(shù)據(jù)，這些策略降低了超步的個(gè)數(shù)和同步時(shí)的等待時(shí)間。以PageRank為例，僅當(dāng)某節(jié)點(diǎn)所代表的頁面權(quán)重發(fā)生變化時(shí)才使得周圍的節(jié)點(diǎn)進(jìn)入計(jì)算狀態(tài)，這樣，大量節(jié)點(diǎn)的權(quán)重不需要重新計(jì)算[10]。被動(dòng)的信息傳遞方式，使得當(dāng)前節(jié)點(diǎn)讀取周圍節(jié)點(diǎn)數(shù)據(jù)時(shí)不需要鄰接節(jié)點(diǎn)進(jìn)入運(yùn)行狀態(tài)，避免了無效的重復(fù)計(jì)算。

3結(jié)語

從數(shù)據(jù)挖掘算法的理論效率來看，并行數(shù)據(jù)挖掘應(yīng)當(dāng)盡可能使用抽象程度較低的框架，然而實(shí)驗(yàn)數(shù)據(jù)表明，一些抽象度較高的分布式計(jì)算框架在眾多算法上有著與MPI相近的性能。除此之外，這些抽象程度較高的框架提供了數(shù)據(jù)切分、計(jì)算任務(wù)調(diào)度和容災(zāi)等能力，從而可以提升數(shù)據(jù)挖掘的效率，而這些是MPI所不能提供的。

從實(shí)驗(yàn)結(jié)果可以得出，并行框架與數(shù)據(jù)集和算法之間存在著密切的關(guān)系。以Mahout為代表的迭代式MapReduce適合數(shù)據(jù)量極大，數(shù)據(jù)之間關(guān)聯(lián)度小，算法中各變量關(guān)聯(lián)度也較小的一類問題。而Twister和Haloop類型的迭代式MapReduce適合數(shù)據(jù)量適中，數(shù)據(jù)之間關(guān)聯(lián)度小，算法中各變量關(guān)聯(lián)度也小的問題。Hama、Pregel和GraphLab適合數(shù)據(jù)集內(nèi)關(guān)聯(lián)度大，算法中變量依賴性強(qiáng)，并行節(jié)點(diǎn)間通訊較為密集的一類問題，其中GraphLab對(duì)異步計(jì)算和通訊的支持使得其適合對(duì)計(jì)算序列要求不嚴(yán)格的一類算法。

參考文獻(xiàn)：

[1]Dean J, Ghemawat S. MapReduce: Simplified Data Processing on Large Clusters[J]. Communications of the ACM, 2008, 51(1): 107-113.

[2]Ranger C, Raghuraman R, Penmetsa A, et al. Evaluating Mapreduce for Multi-core and Multiprocessor Systems[C]//Proceedings of the 13th Symposiu on High Performance Computer Architecture(HPCA). Washington: IEEE Computer Society, 2007: 13-24.

[3]Ekanayake J, Li H, Zhang B J, et al. Twister: A Runtime for Iterative Mapreduce[C]//Proceedings of the 19th ACM International Symposium on High Performance Distributed Computing. New York: ACM, 2010: 810-818.

[4]Ekanayake J. Architecture and Performance of Runtime Environments for Data Intensive Scalable Computing[D]. Bloomington: Indiana University, 2010.

[5]Ekanayake J, Gunarathne T, Fox G, et al. Dryadlinq for Scientific Analyses[C]//Fifth IEEE International Conference on E-Science. Oxford: IEEE, 2009.

[6]Bu Y, Howe B, Balazinska M, et al. HaLoop: Efficient Iterative Data Processing on Large Clusters[J]. Proceedings of the VLDB Endowment, 2010, 3(1-2): 285-296.

[7]Bu Y, Howe B, Balazinska M, et al. The HaLoop Approach to Large-scale Iterative Data Analysis[J]. The VLDB Journal-The International Journal on Very Large Data Bases, 2012, 21(2): 169-190.

[8]Pace M F. BSP vs MapReduce[J]. Procedia Computer Science, 2012(9): 246-255.

[9]Seo S, Yoon E J, Kim J, et al. Hama: An Efficient Matrix Computation with the Mapreduce Framework[C]//Proceedings of The IEEE 2nd International Conference Cloud Computing Technology and Science. Singapore: IEEE, 2010: 721-726.

[10]Low Y, Bickson D, Gonzalez J, et al. Distributed GraphLab: A Framework for Machine Learning and Data Mining in the Cloud[J]. Proceedings of the VLDB Endowment, 2012, 5(8): 716-727.

[11]Low Y, Gonzalez J, Kyrola A, et al. Graphlab: A New Framework for Parallel Machine Learning[EB/OL]. [2014-05-20]. http://www.docin.com/p-661735882.html.

[12]Gonzalez J E, Low Y, Gu H, et al. PowerGraph: Distributed Graph-Parallel Computation on Natural Graphs[C]. Hollywood：The 10th USENIX Symposium on Operating Systems Designand Implementation，2012.

[13]Corbett J C, Dean J, Epstein M, et al. Spanner: Google’s Globally Distributed Database[J]. ACM Transactions on Computer Systems (TOCS), 2013, 31(3): 8.

(責(zé)任編輯：張同學(xué))

The Comparison of Several Parallel Model in the Data Dining Fields

HE Yuan-tao， QI Bing-hui

(Zhengzhou Institute of Aeronautical Industry Management, Zhengzhou 450100, China)

key words:MapReduce; Pregel; Hama; GraphLab; MPI; data mining