鄭緯民　陳文光

摘要：認(rèn)為現(xiàn)有以MapReduce/Spark等為代表的大數(shù)據(jù)處理平臺(tái)在解決大數(shù)據(jù)問(wèn)題的挑戰(zhàn)問(wèn)題方面過(guò)多考慮了容錯(cuò)性，忽視了性能。大數(shù)據(jù)分析系統(tǒng)的一個(gè)重要的發(fā)展方向就是兼顧性能和容錯(cuò)性，而圖計(jì)算系統(tǒng)在數(shù)據(jù)模型上較好地考慮了性能和容錯(cuò)能力的平衡，是未來(lái)的重要發(fā)展方向。

關(guān)鍵詞：大數(shù)據(jù)；分布與并行處理；并行編程；容錯(cuò)；可擴(kuò)展性

Abstract：Existing big data analytic platforms， such as MapReduce and Spark， focus on scalability and fault tolerance at the expense of performance. We discuss the connections between performance and fault tolerance and show they are not mutually exclusive. Distributed graph processing systems are promising because they make a better tradeoff between performance and fault tolerance with mutable data models.

Key words：big data； distributed and parallel processing； parallel programming； fault tolerance； scalability

隨著信息化技術(shù)的發(fā)展，人類(lèi)可以產(chǎn)生、收集、存儲(chǔ)越來(lái)越多的數(shù)據(jù)，并利用這些數(shù)據(jù)進(jìn)行決策，從而出現(xiàn)了大數(shù)據(jù)的概念。大數(shù)據(jù)的定義很多，比較流行的定義是Gartner公司提出的簡(jiǎn)稱(chēng)為3V的屬性，即數(shù)據(jù)量大（Volume），到達(dá)速度快（Velocity）和數(shù)據(jù)種類(lèi)多（Variety）。大數(shù)據(jù)分析利用數(shù)據(jù)驅(qū)動(dòng)的方法，在科學(xué)發(fā)現(xiàn)、產(chǎn)品設(shè)計(jì)、生產(chǎn)與營(yíng)銷(xiāo)、社會(huì)發(fā)展等領(lǐng)域具有應(yīng)用前景。

由于大數(shù)據(jù)的3V屬性，需要在多臺(tái)機(jī)器上進(jìn)行分布與并行處理才能滿(mǎn)足性能要求，因此傳統(tǒng)的關(guān)系型數(shù)據(jù)庫(kù)和數(shù)據(jù)挖掘軟件很難直接應(yīng)用在大數(shù)據(jù)的處理分析中。傳統(tǒng)的超級(jí)計(jì)算技術(shù)，雖然具有很強(qiáng)的數(shù)據(jù)訪問(wèn)和計(jì)算能力，但其使用的MPI編程模型編程較為困難，對(duì)容錯(cuò)和自動(dòng)負(fù)載平衡的支持也有缺陷，主要運(yùn)行在高成本的高性能計(jì)算機(jī)系統(tǒng)上，對(duì)于主要在數(shù)據(jù)中心運(yùn)行的大數(shù)據(jù)分析不是非常適合。

為了解決大數(shù)據(jù)的分析處理所面臨的編程困難，負(fù)載不平衡和容錯(cuò)困難的問(wèn)題，業(yè)界發(fā)展出了一系列技術(shù)，包括分布式文件系統(tǒng)、數(shù)據(jù)并行編程語(yǔ)言和框架以及領(lǐng)域編程模式來(lái)應(yīng)對(duì)這些挑戰(zhàn)。以MapReduce[1]和Spark[2]為代表的大數(shù)據(jù)分析平臺(tái)，是目前較為流行的大數(shù)據(jù)處理生態(tài)環(huán)境，得到了產(chǎn)業(yè)界的廣泛使用。

但是在文章中，我們通過(guò)分析認(rèn)為：MapReduce和Spark系統(tǒng)將容錯(cuò)能力作為設(shè)計(jì)的優(yōu)先原則，而在系統(tǒng)的處理性能上做了過(guò)多的讓步，使得所需的處理資源過(guò)多，處理時(shí)間很長(zhǎng)，這樣反而增加了系統(tǒng)出現(xiàn)故障的幾率。通過(guò)進(jìn)一步分析性能與容錯(cuò)能力的關(guān)系，我們提出了一種性能優(yōu)先兼顧擴(kuò)展性的大數(shù)據(jù)分析系統(tǒng)構(gòu)建思路，并以一個(gè)高性能圖計(jì)算系統(tǒng)為例，介紹了如何用這種思路構(gòu)建大數(shù)據(jù)分析系統(tǒng)。

1 以MapReduce/Spark為

代表的大數(shù)據(jù)分析平臺(tái)

現(xiàn)有的大數(shù)據(jù)分析平臺(tái)主要基于開(kāi)源的Hadoop系統(tǒng)，該系統(tǒng)使用Hadoop分布式文件系統(tǒng)（HDFS），通過(guò)多個(gè)備份的方法保證大量數(shù)據(jù)的可靠存儲(chǔ)和讀取性能，其上的Hive[3]系統(tǒng)支持?jǐn)?shù)據(jù)查詢(xún)，Hadoop MapReduce則支持大數(shù)據(jù)分析程序的開(kāi)發(fā)。

與傳統(tǒng)的并行編程方法MPI[4]相比，MapReduce是近年來(lái)并行編程領(lǐng)域的重要進(jìn)展。盡管Map和Reduce在函數(shù)語(yǔ)言中早已被提出，但將其應(yīng)用于大規(guī)模分布并行處理應(yīng)歸功于Jeff Dean和Ghemewat Sanjay。在MapReduce并行編程模型中，用戶(hù)僅需要編寫(xiě)串行的Map函數(shù)體和Reduce函數(shù)體，MapReduce框架就可以完成并行的計(jì)算，并實(shí)現(xiàn)了自動(dòng)容錯(cuò)和負(fù)載均衡。這對(duì)于數(shù)據(jù)中心中采用的異構(gòu)服務(wù)器、低成本服務(wù)器集群是非常重要的。MapReduce開(kāi)始僅能在使用通用中央處理器（CPU）的分布式系統(tǒng)上運(yùn)行，但后來(lái)被移植到圖形處理器（GPU）和多種加速器上。

MapReduce需要將中間結(jié)果保存到磁盤(pán)中，從而大大影響了性能，美國(guó)加州伯克利大學(xué)提出的Spark系統(tǒng)可以看做是基于內(nèi)存的MapReduce模型，通過(guò)將中間結(jié)果保存在內(nèi)存中，大大提高了數(shù)據(jù)分析程序的性能，類(lèi)似思路的系統(tǒng)還包括HaLoop[5]和Twister[6]等。

Spark和MapReduce在大數(shù)據(jù)領(lǐng)域取得了巨大的成功， 已經(jīng)成為事實(shí)上的大數(shù)據(jù)處理標(biāo)準(zhǔn)。它們與分布式文件系統(tǒng)HDFS、查詢(xún)系統(tǒng)Hive都集成在Hadoop系統(tǒng)中，為大數(shù)據(jù)的存儲(chǔ)、查詢(xún)和處理提供了相對(duì)完整的解決方案。這一系統(tǒng)也具有完整的開(kāi)源社區(qū)支持和商業(yè)公司支持，HortonWorks和Cloudera提供Hadoop的發(fā)行版和服務(wù)，DataBricks為Spark提供發(fā)行版和服務(wù)。IBM于2016年宣布將投入10億美元開(kāi)發(fā)Spark。

2 大數(shù)據(jù)分析平臺(tái)性能的

重要性

盡管以Spark/MapReduce為代表的大數(shù)據(jù)分析平臺(tái)已經(jīng)得到了廣泛應(yīng)用，然而，其性能方面的問(wèn)題也日益暴露出來(lái)。一些研究表明：對(duì)一些大數(shù)據(jù)分析問(wèn)題來(lái)說(shuō)，使用Spark在幾十臺(tái)機(jī)器上的性能甚至不如在某些優(yōu)化過(guò)的程序在單機(jī)上的性能，例如對(duì)Twitter數(shù)據(jù)集來(lái)說(shuō)，Spark在128個(gè)處理器核上需要857 s，而優(yōu)化良好的單線程程序完成同樣的處理功能僅需要300 s的時(shí)間[7]，即在中小規(guī)模數(shù)據(jù)集上Spark的性能功耗比比單線程程序要差2個(gè)數(shù)量級(jí)，甚至在絕對(duì)處理時(shí)間上也比單線程程序要慢。

Spark/MapReduce的性能問(wèn)題，根源在于其設(shè)計(jì)理念上陷入了一個(gè)誤區(qū)：即以容錯(cuò)能力為優(yōu)先的設(shè)計(jì)目標(biāo)，忽視了處理性能。例如，MapReduce和Spark都采用只讀數(shù)據(jù)集的概念，這一方面大大方便了系統(tǒng)進(jìn)行容錯(cuò)，但也使得系統(tǒng)在處理相當(dāng)一部分應(yīng)用時(shí)，性能會(huì)受到嚴(yán)重影響。例如，對(duì)于廣泛使用的廣度優(yōu)先圖搜索問(wèn)題，需要記錄哪些結(jié)點(diǎn)被訪問(wèn)過(guò)，這個(gè)數(shù)據(jù)集如果是只讀的，就只能在每次遍歷迭代時(shí)生成新的數(shù)據(jù)集，這會(huì)大大增加所需的內(nèi)存復(fù)制操作和內(nèi)存容量需求，使得性能大大下降。

而實(shí)際上處理性能的提高，對(duì)提高系統(tǒng)的容錯(cuò)能力也是有正面意義的。一個(gè)數(shù)據(jù)分析任務(wù)的總執(zhí)行時(shí)間，可以按如式（1）估算（為描述方便，公式中略有簡(jiǎn)化）：

總執(zhí)行時(shí)間 = 無(wú)故障執(zhí)行時(shí)間①+無(wú)故障時(shí)容錯(cuò)機(jī)制開(kāi)銷(xiāo)②+故障發(fā)生概率*無(wú)故障執(zhí)行時(shí)間*單次故障恢復(fù)時(shí)間③ （1）

Spark的設(shè)計(jì)主要對(duì)②進(jìn)行優(yōu)化，即通過(guò)只讀數(shù)據(jù)集簡(jiǎn)化無(wú)故障容錯(cuò)機(jī)制的開(kāi)銷(xiāo)，卻大大增加了①的無(wú)故障執(zhí)行時(shí)間，而③實(shí)際是與①正相關(guān)的，即相同機(jī)器數(shù)，執(zhí)行時(shí)間越長(zhǎng)，出故障的概率越大，所需故障恢復(fù)時(shí)間也就越長(zhǎng)。

從上面的分析可以看出：Spark的設(shè)計(jì)理念，即使對(duì)容錯(cuò)本身來(lái)說(shuō)，也很難說(shuō)是合理的，因?yàn)槿绻阅軗p失太大，無(wú)故障執(zhí)行時(shí)間增加太多，會(huì)使得在②減少的開(kāi)銷(xiāo)被③抵消甚至超越[8]。

因此，我們認(rèn)為：大數(shù)據(jù)分析系統(tǒng)的一個(gè)重要的發(fā)展方向就是兼顧性能和容錯(cuò)性。我們需要進(jìn)一步在編程模型和框架上開(kāi)展研究，在保持自動(dòng)負(fù)載平衡和一定容錯(cuò)能力的基礎(chǔ)上，提供優(yōu)化的系統(tǒng)性能。

以Pregel[9]和GraphLab[10]等的圖計(jì)算編程框架是這一類(lèi)工作的代表，這些編程模型主要提供了基于圖結(jié)點(diǎn)（vertex）的編程抽象，并沿著圖的邊進(jìn)行通信，與Map-Reduce相比，這類(lèi)圖編程框架在處理圖數(shù)據(jù)（如社交網(wǎng)絡(luò)、航運(yùn)網(wǎng)絡(luò)和生物網(wǎng)絡(luò)等）時(shí)比Map-Reduce/Spark的表達(dá)更加自然，所獲得的性能也要好得多。這方面的工作引起了全球研究者和工業(yè)界的廣泛關(guān)注，這些工作針對(duì)圖計(jì)算中的負(fù)載不均衡、隨機(jī)訪問(wèn)多、同步和異步等問(wèn)題提出了解決方案。PowerGraph[11]和PowerLyra[12]系統(tǒng)是在GraphLab上改進(jìn)后的圖計(jì)算系統(tǒng)，其性能比GraphLab又有顯著提高。GridGraph[13]提出了利用二維混洗的數(shù)據(jù)結(jié)構(gòu)對(duì)圖計(jì)算進(jìn)行優(yōu)化，可以有效減少圖計(jì)算中的隨機(jī)內(nèi)存訪問(wèn)，提高處理性能?；贕irdGraph的分布式圖計(jì)算系統(tǒng)SAGE.D其性能比PowerLyra進(jìn)一步又提高了1倍左右。如圖1所示：SAGE.D可以在16臺(tái)機(jī)器上以30 s的時(shí)間內(nèi)完成Twitter數(shù)據(jù)集的20次PageRank迭代，性能比Spark提高了接近30倍。

我們可以看到：在某些分析任務(wù)上，基于圖計(jì)算系統(tǒng)的性能比基于Spark的分析系統(tǒng)快1～2個(gè)數(shù)量級(jí)。這意味著基于圖計(jì)算系統(tǒng)在執(zhí)行期間內(nèi)發(fā)生錯(cuò)誤的機(jī)會(huì)僅為Spark的1/10以下，從而不僅在執(zhí)行性能方面，在容錯(cuò)能力方面也優(yōu)于Spark。

3 大數(shù)據(jù)問(wèn)題展望

未來(lái)的大數(shù)據(jù)問(wèn)題會(huì)呈現(xiàn)兩種趨勢(shì)：

（1）具有較小上限的大數(shù)據(jù)問(wèn)題。以社交網(wǎng)絡(luò)的分析問(wèn)題為例，目前Facebook有約10億活躍用戶(hù)，用戶(hù)之間的關(guān)注關(guān)系大約有1 000億個(gè)，大約需要幾個(gè)TB的內(nèi)存容量。社交網(wǎng)絡(luò)的結(jié)點(diǎn)是用戶(hù)，地球上只有幾十億人口，社交網(wǎng)絡(luò)的分析問(wèn)題其上限就是將全部人口數(shù)作為網(wǎng)絡(luò)結(jié)點(diǎn)。

隨著摩爾定律的持續(xù)作用，我們今天已經(jīng)可以很容易地買(mǎi)到內(nèi)容容量為T(mén)B量級(jí)的服務(wù)器，今后可望達(dá)到幾十甚至數(shù)百TB。不斷增長(zhǎng)的硬件能力與較小上限的大數(shù)據(jù)問(wèn)題相遇的結(jié)果，就是把今天的大數(shù)據(jù)問(wèn)題變?yōu)槊魈斓男?shù)據(jù)問(wèn)題，把今天需要數(shù)十、數(shù)百服務(wù)器解決的問(wèn)題變?yōu)榻窈笾恍枰獛着_(tái)甚至單臺(tái)服務(wù)器就可以解決的問(wèn)題。

針對(duì)這類(lèi)應(yīng)用，顯然性能優(yōu)化的大數(shù)據(jù)分析處理平臺(tái)能夠獲得更好的性?xún)r(jià)比。

（2）具有較大上限的大數(shù)據(jù)問(wèn)題。高性能計(jì)算中的很多問(wèn)題規(guī)模具有非常大的上限，例如氣候模擬，需要將空間分成網(wǎng)格、時(shí)間分片，顯然空間上和時(shí)間上的進(jìn)一步細(xì)分都會(huì)導(dǎo)致計(jì)算量和存儲(chǔ)量的大幅度增加，人類(lèi)已有的計(jì)算能力還遠(yuǎn)遠(yuǎn)無(wú)法滿(mǎn)足高精度氣候模擬的要求。針對(duì)這類(lèi)應(yīng)用，性能優(yōu)化的大數(shù)據(jù)分析處理平臺(tái)能夠通過(guò)減少運(yùn)行時(shí)間，提高系統(tǒng)的處理效率和處理規(guī)模。圖2展示了不同并行編程模型在設(shè)計(jì)理念和運(yùn)行時(shí)支撐方面的差異。

綜上所述，現(xiàn)有以Spark為代表的大數(shù)據(jù)處理平臺(tái)在解決大數(shù)據(jù)問(wèn)題的挑戰(zhàn)問(wèn)題方面過(guò)多考慮了容錯(cuò)性，忽視了性能。我們認(rèn)為圖計(jì)算系統(tǒng)在數(shù)據(jù)模型上較好地考慮了性能和容錯(cuò)能力的平衡，是未來(lái)的重要發(fā)展方向。

參考文獻(xiàn)

[1] DEAN， JEFFREY， SANJAY G. MapReduce： Simplified Data Processing on Large Clusters [J]. Communications of the ACM， 2008， 51（1）： 107-113. DOI： 10.1145/1327452.1327492

[2] ZAHARIA M， CHOWDHURY M， DAS T， et al. Resilient Distributed Datasets： A Fault-Tolerant Abstraction for In-Memory Cluster Computing[C]// Proceedings of the 9th USENIX Conference on Networked Systems Design and Implementation. USA： USENIX Association， 2012：15-28

[3] THUSOO A， SARMA S J， JAIN N， et al. Hive： A Warehousing Solution over a Map-Reduce Framework [J]. Proceedings of the VLDB Endowment， 2009， 2（2）： 1626-1629. DOI： 10.14778/1687553.1687609

[4] GROPP W， LUSK E， DOSS N， et al. "A High-Performance， Portable Implementation of the MPI Message Passing Interface Standard [J]. Parallel Computing， 1996， 22（6）： 789-828. DOI： 10.1016/0167-8191（96）00024-5

[5] 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）： 285-296. DOI： 10.14778/1920841.1920881

[6] EKANAYAKE， JALIYA. Twister： A Runtime for Iterative Mapreduce [C]//Proceedings of the 19th ACM International Symposium on High Performance Distributed Computing. USA： ACM， 2010： 810-818

[7] FRANK M， MICHAEL I， MURRAY D G. Scalability！ But at what COST [C]//5th Workshop on Hot Topics in Operating Systems （HotOS XV）. USA： USENIX Association， 2015

[8] KWAK， HAEWOON. What is Twitter， A Social Network or A News Media？ [C]/Proceedings of the 19th International Conference on World Wide Web. USA： ACM， 2010： 591-600

[9] MALEWICZ， GRZEGORZ. Pregel： A System for Large-Scale Graph[C]// Proceedings of the 2010 ACM SIGMOD International Conference on Management of Data. USA： ACM， 2010： 135-146

[10] LOW， YU C. 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] GONZALEZ， Joseph E. PowerGraph： Distributed Graph-Parallel Computation on Natural Graphs [J]. OSDI， 2012， 12（1）： 23-27

[12] CHEN R. Powerlyra： Differentiated Graph Computation and Partitioning on Skewed Graphs[C]//Proceedings of the Tenth European Conference on Computer Systems. USA： ACM， 2015： 1-15

[13] ZHU X， HAN W， CHEN W. GridGraph： Large-Scale Graph Processing on a Single Machine Using 2-Level Hierarchical Partitioning[C]//Proceedings of the Usenix Annual Technical. USA： ASM， 2015： 375-386