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

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

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

此外，在沒有官方PB排序對比的情況下，我們首次將Spark推到了1PB數據（十萬億條記錄）的排序。這個測試的結果是，在使用190個節點的情況下，工作負載在短短不到4小時內完成，同樣遠超雅虎之前使用3800臺主機耗時16個小時的記錄。同時，據我們所知，這也是公用云環境首次完成的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操作，數據的傳輸會橫跨集群中所有主機。Shuffle基本支持了所有的分布式數據處理負載。舉個例子，在一個連接了兩個不同數據源的SQL查詢中，會使用shuffle將需要連接數據的元組移動到同一臺主機；同時，類似ALS等協同過濾算法同樣需要依賴shuffle在網絡中發送用戶或產品的評級（ratings）和權重（weights）。

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

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

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

產生如此結果的技術實現

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

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

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

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

 

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

更多的技術細節

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

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

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

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

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

Spark只能在內存中大放異彩？

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

如本次測試所示，Spark可以勝任集群內存大小N倍的數據集處理。

總結

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

   CDA數據分析師