1/10計算資源，1/3耗時，Spark顛覆MapReduce保持的排序記錄_數(shù)據(jù)分析師

在過去幾年，Apache Spark的采用以驚人的速度增加著，通常被作為MapReduce后繼，可以支撐數(shù)千節(jié)點規(guī)模的集群部署。在內(nèi)存中數(shù)據(jù)處理上，Apache Spark比MapReduce更加高效已經(jīng)得到廣泛認識；但是當數(shù)據(jù)量遠超內(nèi)存容量時，我們也聽到了一些機構(gòu)在Spark使用上的困擾。因此，我們與Spark社區(qū)一起，投入了大量的精力做Spark穩(wěn)定性、擴展性、性能等方面的提升。既然Spark在GB或TB級別數(shù)據(jù)上運行良好，那么它在PB級數(shù)據(jù)上也應(yīng)當同樣如此。

為了評估這些工作，最近我們與AWS一起完成了一個Sort Benchmark（Daytona Gray類別）測試，一個考量系統(tǒng)排序100TB數(shù)據(jù)（萬億條記錄）速度的行業(yè)基準測試。在此之前，這項基準測試的世界記錄保持者是雅虎，使用2100節(jié)點的Hadoop MapReduce集群在72分鐘內(nèi)完成計算。而根據(jù)測試結(jié)果得知，在使用了206個EC2節(jié)點的情況下，Spark將排序用時縮短到了23分鐘。這意味著在使用十分之一計算資源的情況下，相同數(shù)據(jù)的排序上，Spark比MapReduce快3倍！

此外，在沒有官方PB排序?qū)Ρ鹊那闆r下，我們首次將Spark推到了1PB數(shù)據(jù)（十萬億條記錄）的排序。這個測試的結(jié)果是，在使用190個節(jié)點的情況下，工作負載在短短不到4小時內(nèi)完成，同樣遠超雅虎之前使用3800臺主機耗時16個小時的記錄。同時，據(jù)我們所知，這也是公用云環(huán)境首次完成的PB級排序測試。

	Hadoop World Record	Spark 100 TB	Spark 1 PB
Data Size	102.5 TB	100 TB	1000 TB
Elapsed Time	72 mins	23 mins	234 mins
# Nodes	2100	206	190
# Cores	50400	6592	6080
# Reducers	10,000	29,000	250,000
	1.42 TB/min	4.27 TB/min	4.27 TB/min
Rate/node	0.67 GB/min	20.7 GB/min	22.5 GB/min
Sort Benchmark Daytona Rules	Yes	Yes	No
Environment	dedicated data center	EC2 (i2.8xlarge)	EC2 (i2.8xlarge)

為什么會選擇排序？

排序的核心是shuffle操作，數(shù)據(jù)的傳輸會橫跨集群中所有主機。Shuffle基本支持了所有的分布式數(shù)據(jù)處理負載。舉個例子，在一個連接了兩個不同數(shù)據(jù)源的SQL查詢中，會使用shuffle將需要連接數(shù)據(jù)的元組移動到同一臺主機；同時，類似ALS等協(xié)同過濾算法同樣需要依賴shuffle在網(wǎng)絡(luò)中發(fā)送用戶或產(chǎn)品的評級（ratings）和權(quán)重（weights）。

大部分數(shù)據(jù)管道開始時都會有大量的原始數(shù)據(jù)，但是在管道處理過程中，隨著越來越多不相干數(shù)據(jù)被過濾，或者中間數(shù)據(jù)被更簡潔的表示，數(shù)據(jù)量必然會減少。在100TB原始數(shù)據(jù)的查詢上，網(wǎng)絡(luò)上shuffle的數(shù)據(jù)可能只有100TB的一小部分，這種模式也體現(xiàn)在MapReduce的命名。

然而，排序卻是非常有挑戰(zhàn)的，因為數(shù)據(jù)管道中的數(shù)據(jù)量并不會減少。如果對100TB的原始數(shù)據(jù)進行排序，網(wǎng)絡(luò)中shuffle的數(shù)據(jù)必然也是100TB。同時，在Daytona類型的基準測試中，為了容錯，不管是輸入數(shù)據(jù)還是輸出數(shù)據(jù)都需要做備份。實際上，在100TB的數(shù)據(jù)排序上，我們可能會產(chǎn)生500TB的磁盤I/O及200TB的網(wǎng)絡(luò)I/O。

因此，基于上述原因，當我們尋找Spark的測量標準和提升辦法時，排序這個最苛刻的工作負載成為了對比的不二之選。

產(chǎn)生如此結(jié)果的技術(shù)實現(xiàn)

在超大規(guī)模工作負載上，我們投入了大量的精力來提升Spark。從細節(jié)上看，與這個基準測試高度相關(guān)的工作主要有3個： 

首先及最關(guān)鍵的，在Spark 1.1中我們引入了一個全新的shuffle實現(xiàn)，也就是基于排序的shuffle（SPARK-2045）。在此之前，Spark做的是基于哈希的shuffle實現(xiàn)，它需要在內(nèi)存中同時保持P（reduce的分割數(shù)量）個緩沖區(qū)。而在基于排序的shuffle下，任何時候系統(tǒng)只使用一個緩沖區(qū)。這個操作將顯著地減少內(nèi)存開銷，因此同一個場景下可以支撐數(shù)十萬任務(wù)（我們在PB排序中使用了2.5萬個任務(wù)）。

其次，我們修訂了Spark的網(wǎng)絡(luò)模型，通過JNI（SPARK-2468）使用基于Netty的Epoll本地端口傳輸。同時，新的模型還擁有了獨立的內(nèi)存池，繞過了JVM的內(nèi)存分配器，從而減少垃圾回收造成的影響。

最后但同樣重要的是，我們建立了一個外部shuffle服務(wù)（SPARK-3796），它與Spark本身的執(zhí)行器完全解耦。這個新的服務(wù)基于上文所述的網(wǎng)絡(luò)模型，同時，在Spark本身的執(zhí)行器忙于GC處理時，它仍然可以保證shuffle文件處理的繼續(xù)執(zhí)行。

 

通過這三項改變，我們的Spark集群在map階段單 節(jié)點可以支撐每秒3GB的IO吞吐，在reduce階段單節(jié)點可以支撐1.1GB，從而榨干這些機器間10Gbps的網(wǎng)絡(luò)帶寬。

更多的技術(shù)細節(jié)

TimSort：在Spark 1.1版本中，我們將默認排序算法從 quicksort轉(zhuǎn)換到TimSort，它是合并排序和嵌入排序的一個衍生。在大部分現(xiàn)實世界數(shù)據(jù)集中，TimSort比quicksort更加高效，在部分排序數(shù)據(jù)中表現(xiàn)則更為優(yōu)秀。不管在map階段還是Reduce階段，我們都使用了TimSort。

緩存位置的利用：在排序基準測試中，每條記錄的大小都是100字節(jié)，而被排序的鍵是前10個字節(jié)。在排序項目的性能分析階段，我們注意到緩存命中率不如人意，因為每次比較都需要進行一個隨機的對象指針查詢。為此，我們重新設(shè)計了記錄在內(nèi)存的布局，用16字節(jié)長度（兩個長整形）的記錄來表示每條記錄。在這里，前10個字節(jié)代表了排序的鍵，后4個字節(jié)則代表了記錄的位置（鑒于字節(jié)順序和符號，這點并不容易發(fā)現(xiàn)）。這樣一來，每個比較只需要做一次緩存查詢，而且它們都是連續(xù)的，從而避免了隨機的內(nèi)存查詢。

使用TimSort和新的布局方式來利用緩存命中，排序所占用的CPU時間足足減少了5倍。

大規(guī)模下的容錯機制：在大規(guī)模下，許多問題都會暴露。在這個測試過程中，我們看到因為網(wǎng)絡(luò)連通問題出現(xiàn)的節(jié)點丟失，Linux內(nèi)核自旋，以及因為內(nèi)存碎片整理造成的節(jié)點停滯。幸運的是，Spark的容錯機制非常好，并且順利的進行故障恢復(fù)。

AWS的能量：如上文所述，我們使用了206個i2.8xlarge實例來跑這個I/O密集測試。通過SSD，這些實例交付了非常高的I/O吞吐量。我們將這些實例放到一個VPC放置組中，從而通過單SR-IOV增強網(wǎng)絡(luò)性能，以獲得高性能（10Gbps）、低延時和低抖動。 

Spark只能在內(nèi)存中大放異彩？

這個誤解一直圍繞著Spark，特別是剛進入社區(qū)中的新人更是如此認為。不錯，Spark因為內(nèi)存計算的高性能聞名，然而Spark的設(shè)計初衷和理念卻是一個通用的大數(shù)據(jù)處理平臺——不管是使用內(nèi)存還是磁盤。在數(shù)據(jù)無法完全放入內(nèi)存時，基本上所有的Spark運算符都會做一些額外的處理。通俗來說，Spark運算符是MapReduce的超集。

如本次測試所示，Spark可以勝任集群內(nèi)存大小N倍的數(shù)據(jù)集處理。

總結(jié)

擊敗Hadoop MapReduce集群創(chuàng)造的大規(guī)模數(shù)據(jù)處理記錄不僅是對我們工作的一個證明，也是對Spark承諾的一個驗證——在任何數(shù)據(jù)體積，Spark在性能和擴展性上都更具優(yōu)勢。同時，我們也希望在用戶使用過程中，Spark可以帶來時間和開銷上的雙節(jié)省。

   CDA數(shù)據(jù)分析師