基于Map/Reduce的分布“數(shù)據(jù)排序算法分析

余晟雋,宮學(xué)慶,祝君,錢衛(wèi)寧

(華東師范大學(xué)數(shù)據(jù)科學(xué)與工程研究院,上海200062)

基于Map/Reduce的分布“數(shù)據(jù)排序算法分析

余晟雋,宮學(xué)慶,祝君,錢衛(wèi)寧

(華東師范大學(xué)數(shù)據(jù)科學(xué)與工程研究院,上海200062)

為了解決大規(guī)模數(shù)據(jù)的存儲(chǔ)與計(jì)算,近年來(lái)分布式系統(tǒng)得到了大量的應(yīng)用.如何在分布式系統(tǒng)中對(duì)大規(guī)模數(shù)據(jù)集進(jìn)行排序是影響許多應(yīng)用性能的基礎(chǔ)問(wèn)題,其中不僅涉及每個(gè)節(jié)點(diǎn)上排序算法的選擇,更重要的是設(shè)計(jì)協(xié)調(diào)各節(jié)點(diǎn)的分布式算法.本文總結(jié)了分布式系統(tǒng)中常用的分布式排序算法,對(duì)每種算法的執(zhí)行流程、代價(jià)模型和適用場(chǎng)景進(jìn)行了分析,并通過(guò)實(shí)驗(yàn)對(duì)分析結(jié)果進(jìn)行了驗(yàn)證.本文的工作可以幫助開(kāi)發(fā)人員選擇和優(yōu)化分布式環(huán)境下大規(guī)模數(shù)據(jù)排序的算法.

分布式系統(tǒng);排序算法;代價(jià)模型

0 引言

排序是計(jì)算機(jī)科學(xué)中的基礎(chǔ)問(wèn)題,傳統(tǒng)的排序算法研究多關(guān)注于集中式環(huán)境下算法的性能、資源消耗和穩(wěn)定性[1].近年來(lái),在很多領(lǐng)域中數(shù)據(jù)的規(guī)?？焖僭鲩L(zhǎng),已經(jīng)很難在集中式環(huán)境中進(jìn)行存儲(chǔ)和處理,Hadoop等分布式系統(tǒng)[2]逐漸成為大規(guī)模數(shù)據(jù)處理的主流平臺(tái).在分布式環(huán)境中對(duì)大規(guī)模數(shù)據(jù)進(jìn)行排序處理時(shí),不僅需要考慮單節(jié)點(diǎn)上排序算法的選擇,還需要考慮分布式系統(tǒng)的架構(gòu)、數(shù)據(jù)分布策略和分布式計(jì)算模型等因素的影響.在分布式系統(tǒng)中如何提高大規(guī)模數(shù)據(jù)排序處理的性能是一個(gè)值得研究的問(wèn)題.

本文關(guān)注于分布式系統(tǒng)中大規(guī)模數(shù)據(jù)排序算法的性能分析問(wèn)題,提出了單節(jié)點(diǎn)排序(Single Node Sort,SNS)、多節(jié)點(diǎn)歸并排序(Multiple Node Merge Sort,MNMS)和多節(jié)點(diǎn)分區(qū)排序(Multiple Partition Sort,MPS)3種排序算法.針對(duì)每種算法策略,將算法的執(zhí)行過(guò)程細(xì)分為磁盤I/O(Input/Output,I/O)、網(wǎng)絡(luò)I/O和排序計(jì)算等多個(gè)階段,給出了算法的代價(jià)模型,并討論了數(shù)據(jù)分布和數(shù)據(jù)分片大小等因素對(duì)算法的影響.在實(shí)驗(yàn)分析中,我們采用Map/Reduce計(jì)算模型[3]分別實(shí)現(xiàn)了3種排序算法,并在Sorting Benchmark[4]的數(shù)據(jù)集上驗(yàn)證了分析的正確性.本文的工作能夠幫助開(kāi)發(fā)人員在分布式環(huán)境下選擇和優(yōu)化排序算法.

本文后續(xù)內(nèi)容組織如下:第1節(jié)對(duì)排序相關(guān)的研究工作進(jìn)行綜述;第2節(jié)對(duì)分布式場(chǎng)景中影響排序算法性能的因素進(jìn)行分析;第3節(jié)對(duì)3種排序算法進(jìn)行詳細(xì)的介紹和分析;第4節(jié)是對(duì)3種排序算法的實(shí)驗(yàn)分析;第5節(jié)給出不同算法適用場(chǎng)景的結(jié)論.

1 相關(guān)工作

分布式系統(tǒng)快速發(fā)展,近年來(lái)基于分布式系統(tǒng)的應(yīng)用相繼出現(xiàn).Facebook[5]基于Hadoop平臺(tái)構(gòu)建了一個(gè)實(shí)時(shí)系統(tǒng)來(lái)完成其新消息推送,為開(kāi)發(fā)者提供數(shù)據(jù)分析工具、統(tǒng)計(jì)內(nèi)部軟硬件狀態(tài)等需求.Twitter[6]使用Hadoop平臺(tái),幫助公司能夠更快地分析和處理數(shù)據(jù).對(duì)數(shù)據(jù)進(jìn)行排序在這樣的系統(tǒng)中是常見(jiàn)的操作.

分布式環(huán)境,數(shù)據(jù)集通常是按特定的策略被劃分為多個(gè)分片,分布存儲(chǔ)于不同的節(jié)點(diǎn)中,每個(gè)節(jié)點(diǎn)上只保存整個(gè)數(shù)據(jù)集的一部分.為了對(duì)整個(gè)數(shù)據(jù)集進(jìn)行排序,人們需要編寫分布式算法來(lái)協(xié)調(diào)多個(gè)節(jié)點(diǎn)共同完成排序任務(wù).Jim Grey發(fā)起的Sort Benchmark推動(dòng)了分布式環(huán)境中大規(guī)模數(shù)據(jù)排序問(wèn)題的研究,處理能力已經(jīng)有了非常大的進(jìn)步.2009年,Yahoo公司[7]使用Map/Reduce計(jì)算模型對(duì)大規(guī)模數(shù)據(jù)集進(jìn)行排序,在3 452個(gè)節(jié)點(diǎn)的集群上達(dá)到了0.578 TB/min的處理能力;到了2015年,阿里巴巴集團(tuán)實(shí)現(xiàn)的Fuxi Sort系統(tǒng)[8],采用Map/Sort模型在3 377個(gè)節(jié)點(diǎn)的集群上達(dá)到了15.9 TB/min的數(shù)據(jù)排序能力.對(duì)于分布式排序算法的分析能夠幫助開(kāi)發(fā)人員在分布式環(huán)境下選擇和優(yōu)化排序算法.

以上的工作,或是作為排序操作的調(diào)用者,或是基于特定的場(chǎng)景提供快速有效的排序算法.本文提出了3種排序算法,針對(duì)每種算法策略,分析其執(zhí)行代價(jià),討論了在不同場(chǎng)景下排序算法的優(yōu)劣情況.

2 算法性能影響因素

在分布式系統(tǒng)中,排序算法效率不僅僅取決于內(nèi)存排序算法的實(shí)現(xiàn),系統(tǒng)中其他因素對(duì)排序算法的效率也起著至關(guān)重要的作用.

2.1 并行程度

這里的并行程度主要指兩個(gè)方面:一方面是計(jì)算節(jié)點(diǎn)是否需要從存儲(chǔ)節(jié)點(diǎn)獲取到所有的數(shù)據(jù)分片才可以開(kāi)始進(jìn)行排序操作;另一方面是是否可以有多個(gè)計(jì)算節(jié)點(diǎn)協(xié)同進(jìn)行排序操作.當(dāng)節(jié)點(diǎn)需要獲取到所有的數(shù)據(jù)分片才可以進(jìn)行計(jì)算時(shí),往往會(huì)伴隨大量的網(wǎng)絡(luò)等待時(shí)間,這樣顯然會(huì)降低排序的性能.而多個(gè)節(jié)點(diǎn)的協(xié)同計(jì)算往往需要有一個(gè)主控節(jié)點(diǎn)來(lái)負(fù)責(zé)整體的調(diào)度,可能會(huì)出現(xiàn)負(fù)載不均等問(wèn)題.

2.2 待排序數(shù)據(jù)集的分布

當(dāng)需要多個(gè)節(jié)點(diǎn)共同完成本次排序操作時(shí),往往需要對(duì)數(shù)據(jù)按照一定的規(guī)則來(lái)重新劃分.如果事先無(wú)法知曉待排序數(shù)據(jù)分布,一般需要通過(guò)采樣來(lái)獲取數(shù)據(jù)的分布,并以此作為劃分任務(wù)的依據(jù).劃分策略是否能均勻劃分?jǐn)?shù)據(jù)對(duì)排序效率的影響主要體現(xiàn)在兩個(gè)方面:首先,合理的劃分可以使得各個(gè)節(jié)點(diǎn)之間的運(yùn)算負(fù)載大致相同,不會(huì)出現(xiàn)大量數(shù)據(jù)被分配至同一節(jié)點(diǎn),使得少數(shù)節(jié)點(diǎn)負(fù)載過(guò)高的情況;其次,可以通過(guò)改變數(shù)據(jù)劃分的策略來(lái)減少數(shù)據(jù)在節(jié)點(diǎn)間的傳輸量,從而起到提升效率的作用.如果在排序之前就已經(jīng)知曉待排序的數(shù)據(jù)范圍,以及待排序數(shù)據(jù)的分布,那么就可以對(duì)數(shù)據(jù)進(jìn)行更加合理的劃分以提升排序的效率.

2.3 副本使用

如前所述,分布式系統(tǒng)常常通過(guò)冗余存儲(chǔ)數(shù)據(jù)分片的方式來(lái)保證系統(tǒng)的可靠性.在多個(gè)節(jié)點(diǎn)參與運(yùn)算的情況下,計(jì)算的過(guò)程中可以讀取數(shù)據(jù)分片副本的形式來(lái)減少網(wǎng)絡(luò)的傳輸,以此來(lái)優(yōu)化排序的性能.

考慮如下場(chǎng)景,數(shù)據(jù)分片di的主副本存儲(chǔ)在節(jié)點(diǎn)N1中,di的第二副本存儲(chǔ)在節(jié)點(diǎn)N2中.當(dāng)節(jié)點(diǎn)N2計(jì)算需要使用到數(shù)據(jù)分片di時(shí),如果不能使用副本,則需要到N1節(jié)點(diǎn)中獲取數(shù)據(jù)分片di;但是如果可以使用副本,則僅需要在本地讀取數(shù)據(jù)分片di的副本即可.通過(guò)這樣的方式來(lái)減少網(wǎng)絡(luò)間的傳輸,以此來(lái)提升效率.

2.4 數(shù)據(jù)分片大小

在分布式場(chǎng)景下,如果設(shè)置較大的數(shù)據(jù)分片大小,那么在讀取數(shù)據(jù)時(shí),以最小化磁盤尋道時(shí)間的代價(jià),來(lái)提升系統(tǒng)的性能.同時(shí)數(shù)據(jù)在網(wǎng)絡(luò)中傳輸時(shí),能夠最小化網(wǎng)絡(luò)建立連接的代價(jià),幫助進(jìn)一步提升系統(tǒng)性能.但是如果數(shù)據(jù)分片的大小設(shè)置得過(guò)大,又會(huì)導(dǎo)致單個(gè)子任務(wù)需要處理的任務(wù)過(guò)多,進(jìn)而降低系統(tǒng)的性能.

2.5 硬件配置

硬件配置主要分為節(jié)點(diǎn)間的網(wǎng)絡(luò)配置和節(jié)點(diǎn)內(nèi)的硬件配置.節(jié)點(diǎn)間的網(wǎng)絡(luò)配置方面,如今,機(jī)房普遍能夠配置千兆交換機(jī),且PC(Personal Computer,個(gè)人計(jì)算機(jī))也能夠配置千兆網(wǎng)卡.在千兆網(wǎng)的環(huán)境下,極限帶寬為125 M/s.當(dāng)網(wǎng)絡(luò)傳輸速度成為瓶頸時(shí),可以使用更高性能的硬件來(lái)提升性能,如萬(wàn)兆網(wǎng)、InfiniBand等.節(jié)點(diǎn)內(nèi)的硬件配置主要指的是節(jié)點(diǎn)計(jì)算機(jī)的體系結(jié)構(gòu),例如CPU(Central Processing Unit,中央處理器)架構(gòu)中使用CMP(Chip Multiprocessors,單片多核架構(gòu))、SMP(Symmetrical Multi-Processing,對(duì)稱多處理架構(gòu))等,通過(guò)線程的并行,使得同一時(shí)間內(nèi),節(jié)點(diǎn)能夠進(jìn)行更多的運(yùn)算.此外還有SSD(Solid State Drives,固態(tài)硬盤)等的應(yīng)用,由于沒(méi)有了傳統(tǒng)磁盤尋道時(shí)間的消耗,可以大大提升系統(tǒng)隨機(jī)讀寫的性能.

3 分布式排序算法

考慮分布式系統(tǒng)中的一個(gè)典型場(chǎng)景:有待排序數(shù)據(jù)集被按照一定的劃分策略分成了A、B、C、D 4個(gè)數(shù)據(jù)分片,設(shè)置副本個(gè)數(shù)為3,冗余地存儲(chǔ)在具有m個(gè)存儲(chǔ)節(jié)點(diǎn)的分布式存儲(chǔ)系統(tǒng)中,其中A′、B′、C′、D′分別表示各分片的數(shù)據(jù)副本.為了盡可能詳盡、全面地描述排序算法在分布式環(huán)境中的代價(jià),我們定義以下符號(hào)來(lái)表示排序算法的各個(gè)子操作的代價(jià)(表1).

表1 符號(hào)定義Tab.1Symbol definition

3.1 單節(jié)點(diǎn)排序(SNS)

假設(shè)數(shù)據(jù)存儲(chǔ)在多個(gè)節(jié)點(diǎn)中,但是負(fù)責(zé)計(jì)算的節(jié)點(diǎn)之間沒(méi)有并行計(jì)算的能力,只有當(dāng)前被連接的節(jié)點(diǎn)能夠提供計(jì)算并對(duì)對(duì)客戶端提供服務(wù).在這樣的場(chǎng)景下對(duì)進(jìn)行數(shù)據(jù)排序,流程的主要步驟如圖1所示,各節(jié)點(diǎn)將數(shù)據(jù)讀入內(nèi)存,并通過(guò)網(wǎng)絡(luò)傳輸至排序的節(jié)點(diǎn),在該節(jié)點(diǎn)上進(jìn)行排序.

圖1 集中式內(nèi)存排序Fig.1Single node sorting

在這樣的場(chǎng)景下,根據(jù)表1的符號(hào)定義,可以認(rèn)為存儲(chǔ)數(shù)據(jù)節(jié)點(diǎn)的代價(jià)為

計(jì)算節(jié)點(diǎn)的代價(jià)為

由于計(jì)算節(jié)點(diǎn)需要在得到所有數(shù)據(jù)節(jié)點(diǎn)傳輸過(guò)來(lái)的數(shù)據(jù)分片后才可以進(jìn)行排序操作,因此,我們可以得到

這樣我們可以認(rèn)為排序的總代價(jià)為

3.2 ?節(jié)點(diǎn)歸并排序(MNMS)

當(dāng)存儲(chǔ)數(shù)據(jù)的節(jié)點(diǎn)同時(shí)也擁有計(jì)算能力的時(shí)候,可以采用如圖2所示的算法.各節(jié)點(diǎn)先對(duì)存儲(chǔ)在本地的數(shù)據(jù)進(jìn)行排序,待所有的存儲(chǔ)節(jié)點(diǎn)都對(duì)本地的數(shù)據(jù)排好序之后,再傳送至某一個(gè)處理節(jié)點(diǎn)進(jìn)行歸并排序.

圖2 分布式歸并排序算法Fig.2Multiple node sorting

在這樣的排序場(chǎng)景下,根據(jù)表1所述的符號(hào)定義,可以將具有存儲(chǔ)數(shù)據(jù)的計(jì)算節(jié)點(diǎn)排序的代價(jià)歸結(jié)為

對(duì)客戶端進(jìn)行響應(yīng)的計(jì)算節(jié)點(diǎn)代價(jià)為

由于需要等到所有存儲(chǔ)節(jié)點(diǎn)完成數(shù)據(jù)處理之后,響應(yīng)客戶端的計(jì)算節(jié)點(diǎn)才會(huì)開(kāi)始排序計(jì)算,因此排序的總代價(jià)可以歸結(jié)為

3.3 ?節(jié)點(diǎn)分區(qū)排序(MPS)

當(dāng)節(jié)點(diǎn)具有并行計(jì)算能力,可采用如圖3所示的算法.將數(shù)據(jù)按照一定的范圍進(jìn)行劃分,每個(gè)節(jié)點(diǎn)處理一定范圍內(nèi)的數(shù)據(jù),當(dāng)節(jié)點(diǎn)獲取到屬于該范圍的所有數(shù)據(jù)后,對(duì)數(shù)據(jù)進(jìn)行排序操作.

在這樣的排序場(chǎng)景下,根據(jù)表1所述的符號(hào)定義,可以將Map任務(wù)的執(zhí)行代價(jià)歸結(jié)為

Reduce任務(wù)的執(zhí)行代價(jià)歸結(jié)為

圖3 分布式分區(qū)排序算法Fig.3Multiple partition sorting

由于不同的系統(tǒng)對(duì)Map任務(wù)、Reduce任務(wù)的并行不一樣,如在Hadoop中只需要在有一個(gè)Map任務(wù)完成之后就可以開(kāi)啟Reduce任務(wù);而在Spark中,需要當(dāng)所有的Map任務(wù)完成之后才可以開(kāi)啟Reduce任務(wù).這里排序的總代價(jià)可以歸結(jié)為

3.4 算法效率分析

對(duì)于集中式的排序方法,由于系統(tǒng)中提供的計(jì)算模塊的節(jié)點(diǎn)沒(méi)有協(xié)同的計(jì)算能力,僅有對(duì)客戶端提供服務(wù)的節(jié)點(diǎn)能夠?qū)Υ鎯?chǔ)的數(shù)據(jù)進(jìn)行處理,因此只能夠用集中式的內(nèi)存排序來(lái)滿足排序的功能.如果計(jì)算節(jié)點(diǎn)的內(nèi)存排序算法采用類似于插入排序的算法,那么計(jì)算節(jié)點(diǎn)將不再需要等待所有的數(shù)據(jù)分片到達(dá)計(jì)算節(jié)點(diǎn)后才進(jìn)行排序操作.但是由于插入排序相比于快速排序等方式會(huì)增加CPU的使用,因此在節(jié)點(diǎn)中具體采用哪種內(nèi)存排序算法需要根據(jù)具體的硬件水平來(lái)進(jìn)行衡量:當(dāng)網(wǎng)絡(luò)傳輸速率過(guò)慢,而CPU等資源充裕的場(chǎng)景下,可以選擇插入排序算法對(duì)數(shù)據(jù)進(jìn)行排序;然而當(dāng)數(shù)據(jù)量過(guò)大,甚至超出單個(gè)計(jì)算節(jié)點(diǎn)的內(nèi)存大小時(shí),需要使用外部排序才能完成排序操作,其間帶來(lái)的大量磁盤I/O勢(shì)必會(huì)成為性能的一大瓶頸.

對(duì)于分布式歸并排序方法,適用于系統(tǒng)中存儲(chǔ)數(shù)據(jù)的節(jié)點(diǎn)有簡(jiǎn)單計(jì)算的模塊,但不具備各節(jié)點(diǎn)協(xié)同計(jì)算能力的情況.相比于集中式內(nèi)存排序算法,歸并排序可以流水線式地輸出結(jié)果,在數(shù)據(jù)庫(kù)系統(tǒng)中進(jìn)行排序歸并連接時(shí),能獲得良好的應(yīng)用;且當(dāng)數(shù)據(jù)量過(guò)大,超出單個(gè)計(jì)算節(jié)點(diǎn)的內(nèi)存大小時(shí),由于每個(gè)數(shù)據(jù)分片都已經(jīng)是有序的,在最終的計(jì)算節(jié)點(diǎn)上進(jìn)行外部排序的效果會(huì)好于集中式內(nèi)存排序算法.但是,與集中式內(nèi)存排序相同,由于計(jì)算壓力都被分配在一個(gè)節(jié)點(diǎn)上,在待排序數(shù)據(jù)量過(guò)大的場(chǎng)景下,均不能獲得良好的性能.

對(duì)于分布式并行排序算法,適用于計(jì)算節(jié)點(diǎn)有并行計(jì)算能力的系統(tǒng).采用分布式并行排序算法,由于事先不知道待排序數(shù)據(jù)的分布,需要通過(guò)采樣獲取待排序數(shù)據(jù)的分布后對(duì)數(shù)據(jù)的范圍進(jìn)行劃分.常用的采樣算法有:隨機(jī)采樣,根據(jù)采樣率隨機(jī)地選取待排序的數(shù)據(jù);頭部采樣,根據(jù)需要采樣的數(shù)值x,選取待排序數(shù)據(jù)的前x條數(shù)據(jù);等間隔采樣,根據(jù)需要采樣的數(shù)值x與數(shù)據(jù)總量X,等距地選取x條記錄.

由于集中式排序和分布式歸并排序的數(shù)據(jù)分片,最終都只是傳輸?shù)教幚砉?jié)點(diǎn)上進(jìn)行總體的排序,所以是否使用數(shù)據(jù)副本對(duì)最終的排序效率影響不大.而對(duì)于分布式并行排序,由于之前的分析,可以通過(guò)使用副本減少網(wǎng)絡(luò)傳輸,進(jìn)而提升一定的排序性能.

對(duì)于待排序數(shù)據(jù)在進(jìn)行排序操作之前就已經(jīng)是有序的場(chǎng)景下,集中式內(nèi)存排序算法和分布式歸并排序算法需要的僅是傳輸并掃描一遍數(shù)據(jù),代價(jià)較小.分布式并行排序則根據(jù)采樣算法的不同可能呈現(xiàn)出不同的效果:隨機(jī)采樣由于其隨機(jī)性,能夠得到大致均勻的數(shù)據(jù)劃分;頭部采樣由于被采樣數(shù)據(jù)本身就已經(jīng)是有序的,反而造成了劃分的不均;等間隔采樣在這樣的場(chǎng)景下能發(fā)揮出最好的性能,做出最準(zhǔn)確的劃分.

4 實(shí)驗(yàn)分析

為了驗(yàn)證分析結(jié)論的正確性,我們搭建了7個(gè)節(jié)點(diǎn)的Hadoop集群,節(jié)點(diǎn)間通過(guò)千兆以太網(wǎng)連接.每個(gè)節(jié)點(diǎn)的配置為2顆Intel(R)Xeon E5-2650 CPU、128 G內(nèi)存和SSD存儲(chǔ),軟件環(huán)境包括Red Hat Enterprise Linux Server release 6.2、Hadoop 2.7.2和JDK 1.7.0 79.實(shí)驗(yàn)中使用的數(shù)據(jù)集由Sort Benchmark的數(shù)據(jù)生成器gensort產(chǎn)生,數(shù)據(jù)集規(guī)模分為20 GB、40 GB和80 GB 3種.

實(shí)驗(yàn)分為3組:第一組用于對(duì)比3種排序算法對(duì)不同規(guī)模數(shù)據(jù)集的排序性能;第二組測(cè)試數(shù)據(jù)分片大小對(duì)排序性能的影響;最后一組實(shí)驗(yàn)用于分析影響分布式分區(qū)算法性能的因素.在實(shí)驗(yàn)中我們使用監(jiān)控工具nmon for Linux[9]來(lái)獲取排序算法執(zhí)行過(guò)程中各節(jié)點(diǎn)的資源使用情況. 4.13種排序算法性能比較

如圖4所示,顯示了3種排序算法在不同數(shù)據(jù)量場(chǎng)景下的運(yùn)行時(shí)間.其中,SNS為單節(jié)點(diǎn)排序算法,MNMS為多節(jié)點(diǎn)歸并排序算法,MPS為多節(jié)點(diǎn)分區(qū)算法.在各數(shù)據(jù)量上,分布式分區(qū)算法的運(yùn)行時(shí)間均要小于SNS與MNMS.由于SNS和MPS最終的排序只由一個(gè)節(jié)點(diǎn)處理,處理節(jié)點(diǎn)伴隨著大量的網(wǎng)絡(luò)I/O,同時(shí)計(jì)算節(jié)點(diǎn)還需要對(duì)大量數(shù)據(jù)進(jìn)行處理,所以排序的運(yùn)行時(shí)間要長(zhǎng)于分布式分區(qū)算法.

圖4 算法性能比較Fig.4Performance comparison of algorithms

圖5所示,是在待排序數(shù)據(jù)量為40 G的場(chǎng)景下,3種排序算法在運(yùn)行過(guò)程中,某一數(shù)據(jù)節(jié)點(diǎn)以及某一計(jì)算的節(jié)點(diǎn)的資源監(jiān)控圖.由于在SNS以及MNMS的模擬實(shí)現(xiàn)中,計(jì)算節(jié)點(diǎn)上也存儲(chǔ)著待排序的數(shù)據(jù),因此其資源監(jiān)控圖中計(jì)算節(jié)點(diǎn)也含有數(shù)據(jù)節(jié)點(diǎn)的代價(jià).由圖5可以看出,無(wú)論是SNS還是MNMS,它們最終都是在某一個(gè)節(jié)點(diǎn)上完成所有的排序操作,因此這個(gè)節(jié)點(diǎn)在總的排序操作開(kāi)始時(shí),有大量的網(wǎng)絡(luò)傳輸來(lái)獲得待排序數(shù)據(jù),以及大量的磁盤I/O來(lái)完成排序后的數(shù)據(jù)寫入操作.而對(duì)于MPS來(lái)說(shuō),前期由于需要進(jìn)行采樣操作;所以伴隨著少量的磁盤I/O,之后無(wú)論是磁盤還是網(wǎng)絡(luò)的壓力,MPS均小于其他兩種算法.

圖5(a) SNS代價(jià)Fig.5(a)Cost of single node sort

圖5(b) MNMS代價(jià)Fig.5(b)Cost of mutiple node merge sort

圖5(c) MPS代價(jià)Fig.5(c)Cost of mutiple partition sort

4.2 數(shù)據(jù)分片大小的影響

如圖6所示,顯示了待排序數(shù)據(jù)量為40 G場(chǎng)景下,數(shù)據(jù)分片大小分別為64 M、128 M、256 M時(shí),執(zhí)行不同排序算法需要的時(shí)間,其中SNS為單節(jié)點(diǎn)排序算法,MNMS為多節(jié)點(diǎn)歸并排序算法,MPS為多節(jié)點(diǎn)分區(qū)算法.

圖6 不同數(shù)據(jù)分片大小的性能比較Fig.6Performance comparison of different block sizes

由于實(shí)驗(yàn)中集群采用SSD硬盤配置,沒(méi)有硬盤的尋道時(shí)間,所以3種排序算法在不同數(shù)據(jù)分片大小的場(chǎng)景下變化不大.SNS由于主要代價(jià)在最終節(jié)點(diǎn)的排序上,增大了數(shù)據(jù)分片的大小,使得網(wǎng)絡(luò)上有了些許提升,因此在算法運(yùn)行時(shí)間上有所減少.分布式歸并排序雖然在網(wǎng)絡(luò)上也得到了優(yōu)化,但是每個(gè)數(shù)據(jù)分片本地排序的時(shí)間也有稍許提升,因此總體沒(méi)有太大變化.分布式分區(qū)排序由于本地需要處理的數(shù)據(jù)量增多,所以運(yùn)行時(shí)間上有一定的增加.

4.3 分布式分區(qū)排序算法分析

實(shí)驗(yàn)使用了隨機(jī)數(shù)據(jù)、正態(tài)分布數(shù)據(jù),以及執(zhí)行排序操作前就已經(jīng)是有序的有序數(shù)據(jù)這3種不同的數(shù)據(jù)分布,測(cè)試采樣策略為隨機(jī)采樣、頭部采樣以及等間隔采樣3種不同的采樣策略.實(shí)驗(yàn)結(jié)果如圖7所示.

圖7 不同采樣策略的性能比較Fig.7Performance comparison of different sampling strategies

由圖7可以看出,采樣效率上由于隨機(jī)采樣需要掃描更多的數(shù)據(jù),所以效率最低,頭部采樣的效率最高.如果待排序數(shù)據(jù)原本已經(jīng)是有序的,由于排序計(jì)算時(shí)間的減少,排序算法的運(yùn)行時(shí)間要少于其他情況.同時(shí)在數(shù)據(jù)有序的場(chǎng)景下,等間隔采樣能在較少的時(shí)間內(nèi)對(duì)數(shù)據(jù)進(jìn)行均勻的劃分.

5 總結(jié)

本文描述了在分布式場(chǎng)景下3種不同的排序算法,描述了它們與傳統(tǒng)的單節(jié)點(diǎn)內(nèi)存排序算法的不同之處,分析了排序算法不同階段的代價(jià),并結(jié)合分布式場(chǎng)景的特點(diǎn)討論了數(shù)據(jù)分片大小、數(shù)據(jù)副本、數(shù)據(jù)分布等問(wèn)題對(duì)不同算法的影響.通過(guò)實(shí)驗(yàn)對(duì)比驗(yàn)證了分析的正確性.在分布式場(chǎng)景下,為了能夠更快地對(duì)數(shù)據(jù)進(jìn)行處理需要充分考慮系統(tǒng)架構(gòu)特點(diǎn)、系統(tǒng)參數(shù)設(shè)置、各節(jié)點(diǎn)的硬件水平等諸多因素,綜合評(píng)選,選擇最適合的排序算法.

［1］KNUTH D E.The Art of Computer Programming:Sorting and Searching[M].2nd ed.Indianapolis:Addison-Wesley Professional,1998.

［2］BORTHAKUR D.The hadoop distributed file system:Architecture and design[J].Hadoop Project Website, 2007,11:1-10.

［3］DEAN J,GHEMAWAT S.MapReduce:Simplified data processing on large clusters[J].Communications of the ACM,2008,51(1):107-113.

［4］CHRISNYBERG,MEHULSHAH.SortBenchmarkHomePage[EB/OL].(2015)[2016-04-20]. http://sortbenchmark.org/.

［5］BORTHAKUR D,GRAY J,SARMA J S,et al.Apache Hadoop goes realtime at Facebook[C]//Proceedings of the 2011 ACM SIGMOD International Conference on Management of Data.ACM,2011:1071-1080.

［6］MANE S B,SAWANT Y,KAZI S,et al.Real time sentiment analysis of twitter data using hadoop[J].International Journal of Computer Science and Information Technolo,2014,5(3):3098-3100.

［7］O’MALLEY O,MURTHY A C.Winning a 60 second dash with a yellow elephant[J].Proceedings of Sort Benchmark,2009,1810(9):1-9.

［8］WANG J,WU Y,CAI H,et al.Fuxi Sort[EB/OL].(2015)[2016-04-20].http://sortbenchmark.org/FuxiSort2015.pdf.

［9］GRIFFITHS N.Nmon performance:A free tool to analyze AIX and Linux performance[EB/OL].(2003-11-04) [2016-04-20].http://www.ibm.com/developerworks/aix/library/au-analyze aix/.

(責(zé)任編輯:李藝)

Sorting algorithm analysis of distributed data based on Map/Reduce

YU Sheng-jun,GONG Xue-qing,ZHU jun,QIAN Wei-ning
(Institute for Data Science and Engineering,East China Normal University, Shanghai200062,China)

Distributed system has been widely applied in recent years to tackle the storage and calculation of big data.Sorting of large-scale dataset in the distributed system has become the fundamental problem to affect a varieties of application performances which is not only concerning about the selection of sorting algorithm at each node,but also about the development of distributed algorithms to coordinate at each node.This paper summarizes the common distributed sorting algorithms which are applied in the distributed system.Analysis has been conducted to the implementation process,cost model and applicable field of each algorithm.And the analysis results have been verified by experiments.This work can help developers choose and optimize the big data sorting algorithm in distributed environments.

distributed system;sorting algorithm;cost model

TP311

A

10.3969/j.issn.1000-5641.2016.05.014

1000-5641(2016)05-0121-10

2016-05

國(guó)家自然科學(xué)基金(61332006);國(guó)家863計(jì)劃項(xiàng)目(2015AA015307)

余最雋,男,碩士研究生,研究方向?yàn)榉植际綌?shù)據(jù)庫(kù).E-mail:sjyu@obase.com.cn.