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

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

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

這件事情有多難？

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

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

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

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

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

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

阿里云如何做到的？

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

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

接下來(lái)我們重點(diǎn)介紹基于“飛天”系統(tǒng)開(kāi)發(fā)的Fuxisort程序。我們?cè)贕raySort和MinuteSort兩項(xiàng)比賽中使用相同的程序，程序中的優(yōu)化將在后續(xù)章節(jié)中詳細(xì)介紹。

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

圖 1. FuxiSort流程圖

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

 實(shí)現(xiàn)和優(yōu)化

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

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

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

圖2. FuxiSort各階段啟動(dòng)順序

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

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

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

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

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

圖三. 實(shí)時(shí)計(jì)算框架

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

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

對(duì)普通人意味著什么？

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

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

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