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

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

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

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

	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)

為什么會(huì)選擇排序？

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

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

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

因此，基于上述原因，當(dāng)我們尋找Spark的測(cè)量標(biāo)準(zhǔn)和提升辦法時(shí)，排序這個(gè)最苛刻的工作負(fù)載成為了對(duì)比的不二之選。

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

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

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

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

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

 

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

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

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

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

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

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

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

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

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

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

總結(jié)

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

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