阿里云破紀(jì)錄的背后：377秒是如何煉成的？

作者：阿里云“飛天”團隊 2015-11-04 15:07:43
云計算 10月28日，Sort Benchmark官方宣布，阿里云用377秒完成了100TB的數(shù)據(jù)排序，打破了此前Apache Spark創(chuàng)造的1406秒紀(jì)錄。因此在各種圈子里也掀起了討論：這件事情有多難？怎么做到的？對普通人意味著什么等等?；谶@些原因，阿里云“飛天”團隊發(fā)表此文，希望從阿里云的角度回答大家的疑問。

10月28日，Sort Benchmark官方宣布，阿里云用377秒完成了100TB的數(shù)據(jù)排序，打破了此前Apache Spark創(chuàng)造的1406秒紀(jì)錄。在含金量***的GraySort和MinuteSort兩個評測系統(tǒng)中，阿里云分別在通用和專用目的排序類別中創(chuàng)造了4項世界紀(jì)錄。

消息一出，整個技術(shù)圈都沸騰了，特別是對云計算高度關(guān)注的互聯(lián)網(wǎng)、計算機行業(yè)。阿里云打破世界紀(jì)錄，再次點燃了大家對分布式計算的熱情。同時，大數(shù)據(jù)、云計算的各種圈子里也掀起了討論：這件事情有多難？怎么做到的？對普通人意味著什么等等。

基于這些原因，我們發(fā)表此文，希望從阿里云的角度回答大家的疑問。

這件事情有多難？

SortBenchmark的出現(xiàn)，是希望能用最簡單的方法，評估出不同的計算模型，計算平臺的計算能力優(yōu)劣？而排序是最基礎(chǔ)的計算問題，任何一本數(shù)據(jù)結(jié)構(gòu)和算法的計算機教材，首先要講的，就是各種排序算法。所以排序，當(dāng)之無愧的成為這個簡單，但直接有效的benchmark。

SortBenchmark競賽最早的紀(jì)錄追溯到1987年，當(dāng)時都是單機的比賽。如何造出***大的機器，如何盡量壓榨單臺機器的性能是大家的主要工作。

但從1998年開始，大家的策略和思路發(fā)生了改變，分布式計算開始成為主流。大家的工作重點也轉(zhuǎn)變?yōu)椋喝绾斡行д{(diào)度成百上千乃至幾萬臺機器上的CPU、內(nèi)存、網(wǎng)絡(luò)、磁盤IO等物理資源，最快完成海量數(shù)據(jù)的排序。這就像軍隊里，管好幾個人，你可以當(dāng)班長；管好幾十個人，你可以當(dāng)排長；但要管好幾萬人，你才能當(dāng)將軍。

而且，對大規(guī)模集群做線性擴展，遠比大家想象得困難。正如，一個班長說“我只有幾個人，所以我才是班長，但如果你現(xiàn)在給我?guī)兹f人，我馬上就是將軍了”，大家會覺得好笑一樣。當(dāng)規(guī)模不斷擴大，系統(tǒng)的各種瓶頸都會逐漸出現(xiàn)，原來能處理所有消息，能做出各種調(diào)度決定，現(xiàn)在發(fā)現(xiàn)忙不過來；如果找出下級代理，可能又會發(fā)現(xiàn)代理做出的決定和處理總不是***的。
這還只是一種資源的調(diào)度，當(dāng)計算需要多種資源***配合時，你可能會發(fā)現(xiàn)內(nèi)存是有效調(diào)度了，但是會影響網(wǎng)絡(luò)的使用；網(wǎng)絡(luò)可能用好了，但是又影響了磁盤的有效利用。調(diào)度不好時，各個維度可能互相沖突。

當(dāng)你把資源調(diào)度得差不多了，你可能發(fā)現(xiàn)其實這個計算任務(wù)如果從機器A上換到機器B上運行，時間會短很多。或者機器A本來很適合，但是碰巧機器A壞了，就像幾千人的軍隊打仗，有人臨陣脫逃很正常。諸如此類的問題，隨著規(guī)模的不斷擴大，會急劇復(fù)雜化?？梢哉f，規(guī)模每增加一個數(shù)量級，分布式計算平臺需要處理問題就會完全不同。而如何利用大量低端機器達到高性能，正是云計算技術(shù)的核心挑戰(zhàn)。

阿里云的“飛天”分布式計算平臺于2013年正式上線了5000臺的單集群規(guī)模，現(xiàn)在生產(chǎn)線上的規(guī)模更大。關(guān)于如何支持這么大的規(guī)模，可以參考VLDB 2014上伏羲發(fā)表的文章，這不是本文的重點。本文接下來會重點介紹在支持如此大規(guī)模計算集群后，我們還做了哪些事情，讓一萬億條記錄，100TB數(shù)據(jù)的排序能在不到7分鐘完成。

阿里云如何做到的？

“飛天”是阿里云的分布式計算平臺，不僅承擔(dān)著阿里集團內(nèi)部所有的離線數(shù)據(jù)處理任務(wù)，同時也提供阿里云公共云服務(wù)的基礎(chǔ)平臺支撐?！帮w天”系統(tǒng)的關(guān)鍵模塊包括：(a)Pangu-分布式文件系統(tǒng)，負責(zé)存儲和管理計算中心的數(shù)據(jù)文件；(b)Fuxi-分布式調(diào)度系統(tǒng)，負責(zé)管理計算中心的集群資源，調(diào)度分布式系統(tǒng)中運行的在線和離線應(yīng)用。Fuxi提供了一種名為FuxiJob的大數(shù)據(jù)批處理框架，能處理任意的基于DAG(有向無環(huán)圖)描述的用戶計算任務(wù)。

Fuxi已經(jīng)部署在了阿里巴巴多個計算中心的數(shù)十萬服務(wù)器上，單個集群的規(guī)模超過5000臺機器。任何可以用DAG描述的離線數(shù)據(jù)處理作業(yè)都可以用Fuxi Job來執(zhí)行，包括但不限于MapReduce作業(yè)和更加復(fù)雜的機器學(xué)習(xí)作業(yè)。Job的輸入輸出文件以及運行過程中的臨時文件都存儲在Pangu中，依賴Pangu提供的文件副本和locality配置來獲取更好的性能，同時提高數(shù)據(jù)的可靠性。

接下來我們重點介紹基于“飛天”系統(tǒng)開發(fā)的Fuxisort程序。我們在GraySort和MinuteSort兩項比賽中使用相同的程序，程序中的優(yōu)化將在后續(xù)章節(jié)中詳細介紹。

? 概述
首先，程序會對待排序數(shù)據(jù)進行采樣，以確定數(shù)據(jù)各分片的范圍。如圖1所示，除了采樣之外，整個數(shù)據(jù)排序過程分兩大階段：map階段和sort階段。兩個階段都包含多個并行的任務(wù)。

圖 1. FuxiSort流程圖

在map階段，map任務(wù)通過Pangu的ChunkServer進程從本地磁盤中讀入數(shù)據(jù)分片，然后對輸入數(shù)據(jù)進行RangePartition分配給不同的sort，分配后的數(shù)據(jù)通過網(wǎng)絡(luò)直接傳輸給sort任務(wù)。
在sort階段，所有的sort任務(wù)周期性地將map任務(wù)發(fā)過來的數(shù)據(jù)讀入內(nèi)存，當(dāng)內(nèi)存緩沖區(qū)滿的時候，進行基于快速排序算法的內(nèi)存排序，內(nèi)存排序的結(jié)果數(shù)據(jù)將會被寫入Pangu的temporary文件（這種文件存放在本地，不會做多份的拷貝）。當(dāng)sort任務(wù)接收完所有的map數(shù)據(jù)后，會將所有在內(nèi)存中排好序的數(shù)據(jù)以及之前寫入temporary文件中的數(shù)據(jù)一起做歸并排序，歸并排序的最終結(jié)果輸出到Pangu中。當(dāng)FuxiSort所有的sort任務(wù)都執(zhí)行完后，會生成多個的Pangu文件，它們在全局也是有序的。

? 實現(xiàn)和優(yōu)化

a）輸入數(shù)據(jù)采樣。為了降低數(shù)據(jù)傾斜帶來的性能影響，我們對輸入數(shù)據(jù)做了采樣，根據(jù)采樣結(jié)果來確定RangePartition的邊界，從而保證每個sort任務(wù)處理的數(shù)據(jù)量盡量接近。
舉例說明，假設(shè)輸入數(shù)據(jù)被分成了X個文件，首先，我們在每個文件里隨機選取Y個位置，從每個位置開始連續(xù)讀取Z個數(shù)據(jù)樣本，***共得到X * Y * Z個樣本。然后，我們對這些樣本數(shù)據(jù)進行排序，排序后樣本數(shù)據(jù)被均分為S份，這里S為sort任務(wù)的個數(shù)，這樣就得到每個sort任務(wù)待處理數(shù)據(jù)的范圍邊界。由于樣本是均分的，可以使得每個sort任務(wù)都處理了幾乎相等的數(shù)據(jù)量。

對于GraySort而言，我們有20000個輸入文件（X），每個輸入文件選取300個位置（Y），每個位置讀取1個樣本（Z），最終我們選取6000000條樣本進行排序，并均分為20000份（sort任務(wù)個數(shù)），map任務(wù)將根據(jù)上述樣本來進行RangePartition，保證 sort任務(wù)處理的數(shù)據(jù)盡量均勻。整個采樣過程大約耗時35秒。對于MinuteSort而言，3350個輸入文件，我們在每個文件里選取900個數(shù)據(jù)作為樣本，總的樣本數(shù)量為3015000，排序后分成10050份。整個采樣過程耗時4秒。對于IndySort，則不需要這個采樣過程。

b）IO 雙buffer。map階段，F(xiàn)uxiSort在一個I/O buffer中處理數(shù)據(jù)，同時Pangu在另一個buffer中執(zhí)行數(shù)據(jù)讀入操作。這兩個buffer的角色會周期性地進行切換，這樣就能保證處理數(shù)據(jù)操作和I/O操作能并行起來，從而能夠大幅降低任務(wù)的Latency。

圖2. FuxiSort各階段啟動順序

c)流水線操作。如圖2所示，為了進一步降低整體Latency，我們把排序過程的每個階段分解成許多小的步驟，并且盡可能地將這些小的步驟重疊起來執(zhí)行。這些分解出來的小步驟如下所示：

數(shù)據(jù)采樣；
Job啟動；
MapTask讀輸入數(shù)據(jù)；
MapTask發(fā)送數(shù)據(jù)至SortTask；
SortTask接收數(shù)據(jù)；
SortTask將內(nèi)存中的數(shù)據(jù)進行排序，當(dāng)內(nèi)存裝不下時，將排好序的數(shù)據(jù)dump到臨時文件中；
SortTask將內(nèi)存中的有序數(shù)據(jù)和臨時文件中的有序數(shù)據(jù)做merge sort；
SortTask寫最終輸出文件。

FuxiSort將數(shù)據(jù)采樣過程和Job啟動過程并行起來執(zhí)行，在Job啟動階段做的主要工作包括任務(wù)的分發(fā)，以及一些其他的數(shù)據(jù)管理工作，比如收集所有SortTask的網(wǎng)絡(luò)地址，并且通知所有的MapTask。當(dāng)數(shù)據(jù)采樣過程結(jié)束時，采樣程序會將每個分區(qū)的界限存放在Pangu上，并且會建立另一個通知文件存放在Pangu上，用來標(biāo)志采樣結(jié)束。一旦任務(wù)分發(fā)完成，每個MapTask就開始周期性地檢查通知文件是否存在。一旦檢查到通知文件存在，也就意味著采樣程序產(chǎn)生的各分區(qū)界限可用，MapTask就會立刻讀取這些分區(qū)界限，并且根據(jù)這些界限進行數(shù)據(jù)分發(fā)。
步驟（3）（4）和（5）在map階段并行執(zhí)行，步驟（7）和（8）在sort階段并行執(zhí)行。

在步驟(6)中，只有當(dāng)分配給task的內(nèi)存已經(jīng)全部填滿，才會進行排序和dump，由于在排序過程中，內(nèi)存被全部占用，沒有剩余內(nèi)存可以接收新的數(shù)據(jù)，因此步驟(5)會被阻塞。為了緩解這個問題，我們將步驟(5)和(6)并行起來，一旦內(nèi)存使用超過一定量值，就開始做排序，這樣，步驟(6)會被提前執(zhí)行，而步驟(5)也不會被阻塞。當(dāng)內(nèi)存全部占滿時，我們將內(nèi)存中已經(jīng)排好序的數(shù)據(jù)進行歸并，并dump到臨時文件中。顯然，開始做排序的內(nèi)存閾值越低，步驟(6)開始得越早。在我們的實驗中，當(dāng)接收到的數(shù)據(jù)占用分配給Task內(nèi)存的1/10時，開始執(zhí)行步驟(6)。通過這種方法，我們將I/O和計算并行起來，并且沒有明顯的延遲，雖然這種方法可能會需要merge更多的臨時文件，但在我們的場景中沒有因此導(dǎo)致明顯的overhead。
圖2說明了每一步所花費的時間以及在執(zhí)行過程中這些步驟之前的重合部分。

d)網(wǎng)絡(luò)通信優(yōu)化。在map task和sort task之前有明顯的網(wǎng)絡(luò)通信流量，每個網(wǎng)絡(luò)包到達后都會產(chǎn)生CPU中斷。如果對中斷的處理被綁定到一個指定的CPU內(nèi)核上，當(dāng)這個CPU內(nèi)核忙于排序時，對中斷的處理會被延遲，這就可能導(dǎo)致請求超時，甚至丟包。通過設(shè)置”sm_affinity”，可以將網(wǎng)絡(luò)中斷產(chǎn)生的負載均衡到所有的CPU內(nèi)核上，請求超時和丟包的比率明顯下降。

圖三. 實時計算框架

e) 對MinuteSort的進一步優(yōu)化。由于MinuteSort的執(zhí)行時間要求限制在60秒內(nèi)，一般離線作業(yè)的調(diào)度開銷就變得不容忽視。為了降低這些開銷，我們在Fuxi的準(zhǔn)實時Job模型上執(zhí)行MinuteSort，F(xiàn)uxi準(zhǔn)實時Job模型是為了降低調(diào)度產(chǎn)生的overhead，使內(nèi)存計算獲得很高的性能而開發(fā)的。Figure 3說明了準(zhǔn)實時Job模型的框架。在典型的生產(chǎn)環(huán)境中，準(zhǔn)實時系統(tǒng)是一個長期運行的service，會在集群部署過程中被啟動，并且在每臺機器上啟動一個不退出的worker進程。系統(tǒng)啟動之后，用戶可以向準(zhǔn)實時系統(tǒng)的調(diào)度器提交各種job，并且可以獲得job在運行期間的狀態(tài)。sort benchmark競賽要求與排序直接相關(guān)的啟動和退出過程也需要包含在最終的時間里，為了遵守這一規(guī)則，我們在提交MinuteSort job之前，先通過程序去啟動準(zhǔn)實時系統(tǒng)worker，在job運行結(jié)束后，再將worker進程停掉，在最終提交的結(jié)果中，包含了worker啟動和停止所用的時間。

準(zhǔn)實時系統(tǒng)針對的場景是在中等規(guī)模大小的數(shù)據(jù)集(不超過10TB)上，對延遲敏感的數(shù)據(jù)處理過程，在這種規(guī)模的數(shù)據(jù)集下，包括輸入和輸出在內(nèi)的所有records都可能被cache在內(nèi)存中。在我們的實驗中，我們只在準(zhǔn)實時系統(tǒng)中運行MinuteSort。

對普通人意味著什么？

從2009年阿里云誕生那天起，我們的愿景就是打造一個自研的、通用的、大規(guī)模分布式計算底層系統(tǒng)，讓計算像電一樣成為公共服務(wù)，“飛天”平臺是承載這一理念的技術(shù)核心。
FuxiSort打破Sort Benchmark排序比賽世界紀(jì)錄是阿里云6年技術(shù)沉淀的直接體現(xiàn)，是所有技術(shù)人的驕傲。

但這僅僅是開始。技術(shù)本身不是目的，阿里云在任何技術(shù)上的進步，都會通過云產(chǎn)品對外輸出，讓中國乃至全世界的云計算客戶收益。比如本次參賽的FuxiSort，通過開放數(shù)據(jù)處理服務(wù)（Open Data Processing Service， 簡稱ODPS）對外商用。ODPS是由阿里云自主研發(fā)，提供針對TB/PB級數(shù)據(jù)、實時性要求不高的分布式處理能力，應(yīng)用于數(shù)據(jù)分析、挖掘、商業(yè)智能等領(lǐng)域。阿里巴巴的離線數(shù)據(jù)業(yè)務(wù)都運行在ODPS上（詳情參考http://www.aliyun.com/product/odps/ ）。

阿里云將借助技術(shù)創(chuàng)新，不斷提升計算能力與規(guī)模效益，希望更多的合作伙伴、中小企業(yè)、開發(fā)者能夠受益于云計算帶來的便利和價值，共同將云計算變成真正意義上的公共服務(wù)和普惠科技。

當(dāng)前名稱：阿里云破紀(jì)錄的背后：377秒是如何煉成的？
網(wǎng)頁URL：http://www.5511xx.com/article/coeceei.html