大數(shù)據(jù)分析平臺

鄭緯民　陳文光

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

關(guān)鍵詞：大數(shù)據(jù)；分布與并行處理；并行編程；容錯；可擴(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ā)展，人類可以產(chǎn)生、收集、存儲越來越多的數(shù)據(jù)，并利用這些數(shù)據(jù)進(jìn)行決策，從而出現(xiàn)了大數(shù)據(jù)的概念。大數(shù)據(jù)的定義很多，比較流行的定義是Gartner公司提出的簡稱為3V的屬性，即數(shù)據(jù)量大（Volume），到達(dá)速度快（Velocity）和數(shù)據(jù)種類多（Variety）。大數(shù)據(jù)分析利用數(shù)據(jù)驅(qū)動的方法，在科學(xué)發(fā)現(xiàn)、產(chǎn)品設(shè)計(jì)、生產(chǎn)與營銷、社會發(fā)展等領(lǐng)域具有應(yīng)用前景。

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

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

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

1 以MapReduce/Spark為

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

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

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

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

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

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

重要性

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

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

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

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

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

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

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

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

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

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

未來的大數(shù)據(jù)問題會呈現(xiàn)兩種趨勢：

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

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

針對這類應(yīng)用，顯然性能優(yōu)化的大數(shù)據(jù)分析處理平臺能夠獲得更好的性價比。

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

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