基于兩階分區(qū)的MapReduce實(shí)驗(yàn)室系統(tǒng)負(fù)載均衡研究

鄭文麗，熊貝貝，2，程立勛，蔡伊娜，2，包先雨

(1.深圳市檢驗(yàn)檢疫科學(xué)研究院，廣東 深圳 518045；2.深圳海關(guān)食品檢驗(yàn)檢疫技術(shù)中心，廣東 深圳 518000)

0 引 言

在數(shù)字化信息時(shí)代背景下，通過實(shí)驗(yàn)室系統(tǒng)存儲(chǔ)和處理海量數(shù)據(jù)，已成為日常檢驗(yàn)檢測(cè)工作不可或缺的一部分[1]。而隨著實(shí)驗(yàn)室系統(tǒng)智慧化的要求，改進(jìn)存儲(chǔ)和處理技術(shù)在大規(guī)模數(shù)據(jù)高效采集領(lǐng)域中越發(fā)重要。其中，MapReduce是一種處理大規(guī)模數(shù)據(jù)集的分布式框架，擁有拓展性強(qiáng)和容錯(cuò)率高的特點(diǎn)。通過Mapper和Reducer對(duì)海量的檢測(cè)數(shù)據(jù)進(jìn)行拆分和歸類，與傳統(tǒng)構(gòu)架相比，能更高效地處理數(shù)據(jù)。而Hadoop框架源于Apache Nutch，集合了HDFS(hadoop distributed file system)，YARN(yet another resource negotiator)和MapReduce，是具備高容錯(cuò)性、高拓展性和低成本特點(diǎn)的分布式軟件框架，廣泛應(yīng)用于各種領(lǐng)域，包括分布式搜索[2-3]、模式識(shí)別[4-5]、計(jì)算機(jī)機(jī)器學(xué)習(xí)[6-7]和文獻(xiàn)聚類[8-9]等。

MapReduce框架主要運(yùn)用于大型云平臺(tái)的海量數(shù)據(jù)庫存儲(chǔ)的應(yīng)用場(chǎng)景中，可解決數(shù)據(jù)采集高頻、數(shù)據(jù)類型多樣、數(shù)據(jù)來源復(fù)雜以及用戶需求響應(yīng)及時(shí)等難點(diǎn)痛點(diǎn)。同時(shí)，根據(jù)不同場(chǎng)景的業(yè)務(wù)應(yīng)用流程，云平臺(tái)的設(shè)計(jì)與開發(fā)通常采用分層式管理并按需優(yōu)化，這為大數(shù)據(jù)的分流工作提出了新的挑戰(zhàn)。王立俊等人[10]在設(shè)計(jì)氣象大數(shù)據(jù)云平臺(tái)時(shí)，其總體架構(gòu)共分為基礎(chǔ)設(shè)施層、數(shù)據(jù)管理與處理層和應(yīng)用層，實(shí)現(xiàn)了更快的數(shù)據(jù)采集以及更優(yōu)的數(shù)據(jù)處理功能。鑒于氣象數(shù)據(jù)屬于典型的時(shí)序性數(shù)據(jù)，其數(shù)據(jù)流量高達(dá)6萬次/秒，利用MapReduce這種高性能的分布式計(jì)算框架則可以大幅簡(jiǎn)化數(shù)據(jù)處理流程。在該云平臺(tái)中，其MapReduce任務(wù)包括了3個(gè)任務(wù)階段，分別為Map階段、Shuffle階段以及Reduce階段，這3個(gè)階段可連接為鏈?zhǔn)焦ぷ髁?。?shù)據(jù)流在Map階段，依據(jù)數(shù)據(jù)的不同屬性和特點(diǎn)進(jìn)行處理之后，形成不同長度的數(shù)據(jù)塊，以相應(yīng)的規(guī)則映射成鍵值對(duì)，這些鍵值對(duì)就是實(shí)例化的Map任務(wù)。經(jīng)過Map階段處理后的分片數(shù)據(jù)，生成鍵值對(duì)，輸送至Shuffle階段。在Shuffle階段，主要是對(duì)Map階段處理后的分片數(shù)據(jù)進(jìn)行規(guī)則化采集，即采用鍵值對(duì)列表的序號(hào)與鍵值對(duì)形成映射，該階段最終產(chǎn)生分片數(shù)據(jù)排序表。在Reduce階段接收該排序表后，便按該表的序號(hào)以及Reduce算法對(duì)分片數(shù)據(jù)進(jìn)行分類處理。通過各階段數(shù)據(jù)處理方式，不僅對(duì)批量數(shù)據(jù)實(shí)行了有針對(duì)性的調(diào)度管理，同時(shí)也對(duì)用戶隱藏了具體的管理流程，提升了處理效率，也降低了用戶的使用成本。

MapReduce除了在云平臺(tái)系統(tǒng)運(yùn)用方面有很多理論研究基礎(chǔ)，在其框架的改良優(yōu)化方面也有著廣泛的研究成果，胡東明等人[11]在MapReduce框架下提出了一種負(fù)載均衡的Top-k連接查詢算法。該算法不僅在MapReduce框架下實(shí)現(xiàn)了Top-k連接查詢算法，還通過提前終止機(jī)制和負(fù)載均衡機(jī)制來增強(qiáng)其數(shù)據(jù)連接處理性能。由于Map任務(wù)和Reduce任務(wù)可以并行處理的特點(diǎn)，能盡量避免了鏈接MapReduce作業(yè)的初始化開銷，降低了數(shù)據(jù)處理成本。該算法流程包括直方圖構(gòu)建、提前終止機(jī)制、數(shù)據(jù)過濾、負(fù)載均衡機(jī)制4個(gè)步驟組成，其中提前終止、數(shù)據(jù)過濾都在Map階段實(shí)現(xiàn)，Reduce階段則通過Top-k連接查詢算法完成數(shù)據(jù)清洗。在數(shù)據(jù)過濾階段，Map任務(wù)會(huì)處理每個(gè)記錄，因?yàn)獒槍?duì)每個(gè)作業(yè)會(huì)形成不同的過濾機(jī)制。通過了數(shù)據(jù)過濾機(jī)制的記錄則進(jìn)入到Reduce階段，并使用啟發(fā)式任務(wù)調(diào)度算法對(duì)每個(gè)記錄的Reduce任務(wù)進(jìn)行數(shù)據(jù)調(diào)度，將這些任務(wù)依次分配給連接總數(shù)較低的Reducer進(jìn)行處理。該算法利用了Map階段輸出的鍵值對(duì)以及連接值進(jìn)行分組，將其按照自定義分區(qū)程序的結(jié)果分配給Reduce任務(wù)。MapReduce的并行化處理大大降低了算法的總執(zhí)行時(shí)間，但采取不同Top-k算法則產(chǎn)生不同的任務(wù)執(zhí)行時(shí)間，經(jīng)比較，最終選用了P-TKJ算法，相較于RSJ算法獲得了更快的執(zhí)行速度，并隨著數(shù)據(jù)集的增大，其效率的優(yōu)勢(shì)越明顯，使得MapReduce框架下不僅實(shí)現(xiàn)了海量數(shù)據(jù)的Top-k連接查詢算法，還提高了CPU的利用效率。

在實(shí)驗(yàn)室系統(tǒng)中引入MapReduce框架[12]，可以極大簡(jiǎn)化分布式程序，集中精力于數(shù)據(jù)處理的任務(wù)本身，提高實(shí)驗(yàn)室系統(tǒng)數(shù)據(jù)處理的效率[13]。實(shí)驗(yàn)室系統(tǒng)在執(zhí)行MapReduce任務(wù)時(shí)，數(shù)據(jù)是以鍵值對(duì)輸入，并在Mapper節(jié)點(diǎn)時(shí)進(jìn)行聚合處理，最終根據(jù)哈希值函數(shù)Shuffle至各Reducer節(jié)點(diǎn)進(jìn)行Sort和Reduce。在風(fēng)力發(fā)電行業(yè)中，其實(shí)驗(yàn)室監(jiān)測(cè)數(shù)據(jù)包含設(shè)備的狀態(tài)參數(shù)、氣象環(huán)境、地理信息等，數(shù)據(jù)體量較大，類型多。雖然Hadoop平臺(tái)能夠滿足大數(shù)據(jù)的基本需求，但在運(yùn)營效率方面仍有待提高。王林童等人[14]提出了基于MapReduce的多源數(shù)據(jù)并行關(guān)聯(lián)查詢的優(yōu)化方法，主要是針對(duì)風(fēng)電大數(shù)據(jù)進(jìn)行存儲(chǔ)預(yù)處理，盡量使具有關(guān)聯(lián)關(guān)系的數(shù)據(jù)存儲(chǔ)在同一存儲(chǔ)節(jié)點(diǎn)中，之后采用哈希分桶算法對(duì)數(shù)據(jù)存儲(chǔ)進(jìn)行優(yōu)化，并在查詢時(shí)利用MapReduce框架的并行性特點(diǎn)，最終對(duì)數(shù)據(jù)可以采用并行優(yōu)化的查詢與計(jì)算。具體而言，該系統(tǒng)首先對(duì)具有相同屬性(時(shí)間、地點(diǎn)、設(shè)備號(hào)等)的風(fēng)電數(shù)據(jù)進(jìn)行歸并的預(yù)處理，在關(guān)聯(lián)查詢的基礎(chǔ)上數(shù)據(jù)清洗之后得到風(fēng)電時(shí)序數(shù)據(jù)合并表。合并表按照關(guān)聯(lián)字段的哈希值分配到不同的“桶”中，同一個(gè)“桶”中的數(shù)據(jù)即存儲(chǔ)到同一個(gè)數(shù)據(jù)節(jié)點(diǎn)中，實(shí)現(xiàn)本地化的存儲(chǔ)優(yōu)化。當(dāng)用戶使用該系統(tǒng)進(jìn)行數(shù)據(jù)查詢時(shí)，設(shè)計(jì)在Map階段完成風(fēng)電數(shù)據(jù)的查詢、過濾、篩選等操作，盡量減少Reduce階段的操作，此時(shí)就大大降低了傳輸時(shí)延，提升了數(shù)據(jù)查詢效率。文獻(xiàn)[15]則利用MapReduce框架，構(gòu)建了一個(gè)基于物聯(lián)網(wǎng)的細(xì)胞生物學(xué)智慧云實(shí)驗(yàn)室系統(tǒng)，該模型共包括四層：實(shí)驗(yàn)設(shè)備層、IT基礎(chǔ)設(shè)施層、控制層以及數(shù)據(jù)分析層。實(shí)驗(yàn)設(shè)備層主要工作為自動(dòng)采集不同實(shí)驗(yàn)儀器設(shè)備上產(chǎn)生的數(shù)據(jù)，采集后的實(shí)驗(yàn)數(shù)據(jù)交由IT基礎(chǔ)設(shè)施層進(jìn)行存儲(chǔ)，在控制層可對(duì)這些存儲(chǔ)的數(shù)據(jù)進(jìn)行查詢，而在數(shù)據(jù)分析層，則針對(duì)數(shù)據(jù)種類應(yīng)用不同的分析方式進(jìn)行篩選；通過4個(gè)層級(jí)的相互作用，為實(shí)時(shí)根據(jù)用戶的需求進(jìn)行針對(duì)性調(diào)整和優(yōu)化提供了極大的便利。

在這個(gè)過程中，如果原始數(shù)據(jù)分布不均，易出現(xiàn)數(shù)據(jù)傾斜的問題，引發(fā)Reducer節(jié)點(diǎn)負(fù)載不均衡，導(dǎo)致整個(gè)MapReduce任務(wù)的執(zhí)行時(shí)間過長。因此，基于MapReduce的實(shí)驗(yàn)室系統(tǒng)負(fù)載均衡成為近期國內(nèi)外學(xué)者們的研究熱點(diǎn)。

1 基于兩階分區(qū)的MapReduce負(fù)載均衡算法

針對(duì)原始數(shù)據(jù)分布不均導(dǎo)致的數(shù)據(jù)傾斜問題，國內(nèi)外學(xué)者從不同角度提出了解決方案。杜鵑等人提出了一種利用快速無偏分層圖抽樣算法的負(fù)載均衡算法，在小規(guī)模數(shù)據(jù)上運(yùn)行良好，但未在大型真實(shí)的復(fù)雜數(shù)據(jù)集中運(yùn)用、驗(yàn)證[16]。陶永才等人提出了MR-LSP(MapReduceon-line load balancing mechanismbasedon sample partition)算法，對(duì)原始數(shù)據(jù)進(jìn)行采樣分析，通過分析結(jié)果進(jìn)行負(fù)載均衡的分區(qū)分配Reducer節(jié)點(diǎn)的策略，但該算法忽略了數(shù)據(jù)采樣率不高時(shí)的情況[17]。在馬青山等人[18]提出的DSJA(data skew join algorithm)算法從數(shù)據(jù)關(guān)系表中的連接鍵出現(xiàn)頻率的角度出發(fā)，區(qū)分?jǐn)?shù)據(jù)是否產(chǎn)生連接傾斜的情況，再分配到相應(yīng)的Reducer節(jié)點(diǎn)中進(jìn)行處理，但該算法僅考慮了Reduce端輸出的負(fù)載均衡，未考慮Map端到Reduce端的輸入階段的負(fù)載均衡處理。M.A.Irandoost 等人提出的LAHP(learning automata hash partitioner)算法[19]是根據(jù)學(xué)習(xí)自動(dòng)機(jī)策略，在作業(yè)執(zhí)行階段，對(duì)數(shù)據(jù)鍵值進(jìn)行調(diào)配各個(gè)Reducer節(jié)點(diǎn)的數(shù)據(jù)量；但該算法只考慮了數(shù)據(jù)偏度高，而未考慮數(shù)據(jù)采樣率高的情形，且只優(yōu)化了執(zhí)行階段，忽略了計(jì)算時(shí)間對(duì)流程的影響。Elaheh Gavagsaz 等[20]提出的可拓展的高偏度數(shù)據(jù)的隨機(jī)采樣模型SBaSC(sorted-balance algorithm using scalable simple random sampling )對(duì)高偏度數(shù)據(jù)進(jìn)行了隨機(jī)采樣，通過采樣中鍵值近似分布推算數(shù)據(jù)的分布，從而進(jìn)行負(fù)載平衡的分配；該算法在基于Spark的網(wǎng)絡(luò)數(shù)據(jù)中得到了較好的效果，但未能泛化至不同的數(shù)據(jù)集，存在一定的局限性。黃偉建[21]等使用并行隨機(jī)抽樣貪心算法，縮短了采樣階段的執(zhí)行時(shí)間，但當(dāng)MapReduce輸入數(shù)據(jù)量較大時(shí)，準(zhǔn)確性不夠且負(fù)載效果較差。

為解決上述問題，Xu等[22]提出的兩階分區(qū)算法在數(shù)據(jù)偏度較低時(shí)通過CC(cluster combination)調(diào)度方案實(shí)現(xiàn)負(fù)載平衡，在數(shù)據(jù)偏度較高時(shí)則通過CSC(cluster split combination)調(diào)度方案，可比傳統(tǒng)的MapReduce減少60%的運(yùn)算時(shí)長。在該算法中，一個(gè)數(shù)據(jù)處理作業(yè)被分為了兩個(gè)階段，一是制定分區(qū)方案，二是執(zhí)行MapReduce作業(yè)。在制定分區(qū)方案中，CC調(diào)度方案主要思路是選擇具有最大數(shù)量的鍵值對(duì)的數(shù)據(jù)塊調(diào)度給當(dāng)前最小工作量的鍵值對(duì)的Reducer，這種啟發(fā)式算法使用了標(biāo)準(zhǔn)差這一評(píng)價(jià)指標(biāo)來自適應(yīng)地衡量所有Reducer的負(fù)載量。當(dāng)標(biāo)準(zhǔn)差高于設(shè)定的閾值后，將對(duì)數(shù)據(jù)塊進(jìn)行再分配。其分配的方式基于對(duì)數(shù)據(jù)塊所含鍵值對(duì)數(shù)量的降序排序，并對(duì)Reducer也進(jìn)行工作量進(jìn)行降序排列，接下來則將兩個(gè)表中的首位進(jìn)行匹配，以確保整個(gè)MapReduce系統(tǒng)的負(fù)載均衡。CC調(diào)度方案可以解決數(shù)據(jù)量輕度傾斜的情況，但面對(duì)數(shù)據(jù)量極度傾斜的情況，就無法實(shí)現(xiàn)較好的效果。針對(duì)這樣的情況，該算法又提出了第二種情況的方法，即CSC調(diào)度方案。面對(duì)包含著較多數(shù)據(jù)記錄的數(shù)據(jù)塊，CSC采用的是劃分的方式，將較大的數(shù)據(jù)塊劃分成較小的數(shù)據(jù)塊。由于在MapReduce框架下，每個(gè)數(shù)據(jù)塊都對(duì)應(yīng)著一個(gè)Reducer，因此在對(duì)較大數(shù)據(jù)進(jìn)行劃分的時(shí)候，需要額外分配一個(gè)Reducer。與CC類似，CSC在進(jìn)行數(shù)據(jù)塊分割的時(shí)候也是采用啟發(fā)式算法解決這個(gè)NP-Hard(non-deterministic polynomial hard)問題，偏度較大的數(shù)據(jù)塊將被分成n塊(n為Reducer的總數(shù))，以確保整個(gè)MapReduce系統(tǒng)保持負(fù)載平衡。在分區(qū)任務(wù)完成之后，該算法便執(zhí)行MapReduce任務(wù)。在MapReduce任務(wù)階段主要是對(duì)數(shù)據(jù)集進(jìn)行采樣，不同的數(shù)據(jù)則有不同的特點(diǎn)，因此，不同的數(shù)據(jù)集則采用不同的采樣方法，如對(duì)預(yù)訓(xùn)練過的數(shù)據(jù)集，則采用區(qū)間分布；對(duì)于全新的數(shù)據(jù)集，則采用隨機(jī)抽樣的方式。該算法為了具備更強(qiáng)的泛化性能，采取了隨機(jī)抽樣的方式，并利用Map任務(wù)和Reduce任務(wù)的并行處理特點(diǎn)，先對(duì)數(shù)據(jù)集進(jìn)行預(yù)處理，利用鍵值中位數(shù)估值的概率進(jìn)行分布。首先對(duì)數(shù)據(jù)集進(jìn)行均勻的數(shù)據(jù)采樣，再針對(duì)采樣中出現(xiàn)鍵值的概率分布進(jìn)行統(tǒng)計(jì)，基于統(tǒng)計(jì)結(jié)果，擬合不同的概率分布，并根據(jù)相應(yīng)概率分布對(duì)數(shù)據(jù)塊進(jìn)行抽樣，以解決數(shù)據(jù)偏度和采樣率都較高的問題。雖然該兩階分區(qū)算法在解決數(shù)據(jù)偏度和采樣率都較高的問題中都有相當(dāng)好的性能表現(xiàn)，但在執(zhí)行MapReduce作業(yè)，該算法也存在著以下的缺陷：

1)由于兩次MapReduce任務(wù)采用了串行的執(zhí)行方式，導(dǎo)致整體任務(wù)的執(zhí)行時(shí)間長。

2)在采樣率較高時(shí)，由于抽樣數(shù)據(jù)會(huì)在執(zhí)行階段進(jìn)行重復(fù)的MapReduce作業(yè)，處理時(shí)間過長。

3)執(zhí)行CSC調(diào)度方案時(shí)，資源利用率較低，導(dǎo)致出現(xiàn)了部分Reducer節(jié)點(diǎn)出現(xiàn)空轉(zhuǎn)的情況。

為了解決上述算法執(zhí)行時(shí)間較長及負(fù)載不均衡的問題，本文提出了一種基于MapReduce框架下兩階分區(qū)的改進(jìn)算法。該算法主要工作為：

1)對(duì)MapReduce任務(wù)的流程進(jìn)行并行化處理。

2)在采樣率高的情況下，將采樣階段輸出數(shù)據(jù)進(jìn)行回收。

3)在原始數(shù)據(jù)偏度較高的情況下，提出ICSC(improved cluster split combination)算法。本文提出的算法可以在算法速率提高且執(zhí)行時(shí)間縮短的同時(shí)，有針對(duì)性地解決高偏度，高采樣率情況下數(shù)據(jù)傾斜的問題。改進(jìn)后兩階段的流程如圖1和圖2所示。

圖1 采樣階段流程

圖2 執(zhí)行階段流程

在采樣階段，與優(yōu)化前的兩階分區(qū)算法不同的是，之前是先進(jìn)行采樣數(shù)據(jù)進(jìn)行數(shù)據(jù)統(tǒng)計(jì)再劃分?jǐn)?shù)據(jù)塊，優(yōu)化后則是先對(duì)數(shù)據(jù)塊進(jìn)行劃分，并對(duì)劃分后的數(shù)據(jù)塊進(jìn)行采樣后與數(shù)據(jù)集的概率分布進(jìn)行比對(duì)，當(dāng)某一個(gè)數(shù)據(jù)塊的采樣率很高(超過設(shè)定的閾值)時(shí)，就不需要再重新對(duì)數(shù)據(jù)塊進(jìn)行采樣而直接使用采樣出的數(shù)據(jù)。低于采樣率閾值的數(shù)據(jù)塊則需要再進(jìn)一步采樣，但也是增加而非重新分配。這樣的做法不僅能夠降低時(shí)間開銷，同時(shí)也降低了系統(tǒng)對(duì)于數(shù)據(jù)取用的I/O成本。

而在MapReduce執(zhí)行階段，本文算法除了使用其并行處理的特點(diǎn)，在數(shù)據(jù)被劃分為不同數(shù)據(jù)塊后被分配給不同的Mapper，由于數(shù)據(jù)塊在采樣階段就出現(xiàn)了處理速度不一致的情況，因此如果采用串行方式則會(huì)大大拖累不需要重采樣的數(shù)據(jù)塊，增加時(shí)間開銷。本文算法通過直接處理和計(jì)算系統(tǒng)負(fù)載均衡的方式并行處理數(shù)據(jù)塊，并在分配好數(shù)據(jù)塊的Reducer之后再進(jìn)行不同數(shù)據(jù)偏度方式進(jìn)行調(diào)度。對(duì)于偏度較低的數(shù)據(jù)塊，本文依然沿用了兩階分區(qū)算法中的CC調(diào)度方式，而針對(duì)偏度較高的數(shù)據(jù)塊則采用了優(yōu)化了CSC調(diào)度方式的ICSC方法，具體的過程在第2節(jié)中進(jìn)行詳細(xì)地闡述。

2 算法設(shè)計(jì)

2.1 兩階并行的設(shè)計(jì)

根據(jù)Apache官方文件所述，當(dāng)系統(tǒng)批處理數(shù)據(jù)且執(zhí)行MapReduce任務(wù)時(shí)，Mapper和Reducer的任務(wù)可以進(jìn)行一定程度的并行處理。其中，Mapper需完成Map的子任務(wù)，并將輸入的鍵值對(duì)進(jìn)行歸類。而Reducer需要完成的任務(wù)主要分為3個(gè)子階段：Shuffle、Sort和Reduce。Shuffle階段的職責(zé)為根據(jù)提供的哈希函數(shù)將Map任務(wù)的結(jié)果分別傳輸?shù)礁鱾€(gè)Reducer節(jié)點(diǎn)上，同時(shí)也是唯一一個(gè)可以和Map任務(wù)同時(shí)進(jìn)行的子任務(wù)。因此，Shuffle子任務(wù)的結(jié)束時(shí)間至少晚于Map任務(wù)的結(jié)束時(shí)間。Sort階段是各個(gè)Reducer將收到的數(shù)據(jù)進(jìn)行排序、統(tǒng)合的過程。最后的子階段Reduce則是對(duì)數(shù)據(jù)進(jìn)行計(jì)算，并將結(jié)果輸出到HDFS的過程。在Hadoop的MapReduce框架下，可根據(jù)Map的任務(wù)進(jìn)度，通過調(diào)整Slowstart參數(shù)，對(duì)Shuffle的開始時(shí)間進(jìn)行自定義。這種若參數(shù)設(shè)置為1.00，則為串聯(lián)運(yùn)行，Shuffle子任務(wù)將在Map任務(wù)全部完成后再執(zhí)行。

在兩階分區(qū)的算法中，由于執(zhí)行階段Shuffle子任務(wù)并不依賴于Map任務(wù)的完成，而是依賴于上一個(gè)階段——采樣階段的Reducer結(jié)果所產(chǎn)生的分區(qū)方案，因此采樣階段和執(zhí)行階段并非兩個(gè)相互獨(dú)立的任務(wù)，所以沒有利用調(diào)整Slowstart參數(shù)來進(jìn)行并行化。但在MapReduce任務(wù)中，Map任務(wù)的執(zhí)行時(shí)間往往遠(yuǎn)長于Shuffle任務(wù)的時(shí)間，所以若執(zhí)行階段的Map和Shuffle并行處理，會(huì)導(dǎo)致Shuffle任務(wù)的執(zhí)行時(shí)間被拉長?；诖?，本文提出在第一階段得到采樣結(jié)果后便開始第二階段的Shuffle任務(wù)，通過并行化流程來減少M(fèi)apper節(jié)點(diǎn)的空轉(zhuǎn)導(dǎo)致的資源浪費(fèi)。具體流程如圖3所示，實(shí)線箭頭標(biāo)識(shí)了串行工作時(shí)段，虛線箭頭標(biāo)示了并行工作時(shí)段。本文算法將原有的流程做兩方面更改：首先，讓執(zhí)行階段MapReduce的Map任務(wù)隨著采樣階段Map任務(wù)的結(jié)束立刻開始；其次，讓執(zhí)行階段MapReduce的Shuffle任務(wù)隨著采樣階段的分區(qū)方案的計(jì)算完成而立即啟動(dòng)，通過并行化的方式縮短整體任務(wù)的執(zhí)行時(shí)間。

圖3 MapReduce流程優(yōu)化前后對(duì)比

2.2 采樣階段的設(shè)計(jì)

在兩階分區(qū)算法中，第一次MapReduce任務(wù)的結(jié)果只用來規(guī)劃執(zhí)行階段各Reducer節(jié)點(diǎn)的負(fù)載分配，在數(shù)據(jù)計(jì)算完畢后分區(qū)處理后則被舍棄。該算法的設(shè)計(jì)在數(shù)據(jù)采樣率較高的時(shí)候，由于對(duì)同樣的數(shù)據(jù)進(jìn)行相同的多次操作，就會(huì)造成計(jì)算資源嚴(yán)重浪費(fèi)，增加數(shù)據(jù)使用開銷。針對(duì)這樣的問題，本文對(duì)上述算法進(jìn)行了以下改進(jìn)：將第一階段數(shù)據(jù)塊采樣輸出的結(jié)果重新Shuffle至對(duì)應(yīng)的Reducer節(jié)點(diǎn)上。因此，到了執(zhí)行階段的MapReduce，則可直接使用采樣數(shù)據(jù)塊，而不必對(duì)同一數(shù)據(jù)塊進(jìn)行重復(fù)作業(yè)，使采樣數(shù)據(jù)得到充分利用，且節(jié)省了存儲(chǔ)資源與計(jì)算資源。

2.3 分區(qū)調(diào)度階段的設(shè)計(jì)

根據(jù)數(shù)據(jù)偏度情況，本文采用了兩種不同的分區(qū)調(diào)度算法。在偏度較低時(shí)，則繼續(xù)沿用Xu等人[22]提出的CC(cluster combination)調(diào)度算法，對(duì)采樣結(jié)果進(jìn)行分區(qū)。在偏度較高時(shí)，在執(zhí)行CC(cluster combination)調(diào)度算法，大數(shù)據(jù)塊會(huì)導(dǎo)致負(fù)載不均衡。因此，上述文獻(xiàn)中提出一種CSC(cluster split combination)調(diào)度算法來執(zhí)行分區(qū)。該調(diào)度算法先對(duì)大數(shù)據(jù)塊進(jìn)行拆分，由多個(gè)Reducer節(jié)點(diǎn)分別處理，當(dāng)所有Reducer完成各自的任務(wù)后，還需要一個(gè)Reducer進(jìn)行額外的合并任務(wù)。此時(shí)，其它Reducer會(huì)陷入空轉(zhuǎn)的狀態(tài)；同時(shí)，如果所有的Reducer都參與到最后一次Reduce任務(wù)中，任何一個(gè)節(jié)點(diǎn)的故障都會(huì)導(dǎo)致最后一次Reduce任務(wù)無法進(jìn)行，致使MapReduce任務(wù)執(zhí)行失敗，如圖4所示。

為此，本文提出的ICSC算法，如圖5所示。其主要思想是將Reducer節(jié)點(diǎn)的工作量進(jìn)行一定程度的壓縮，使得這些Reducer完成工作的總時(shí)間與其他不參與處理大數(shù)據(jù)分片的Reducer節(jié)點(diǎn)保持一致，減少節(jié)點(diǎn)的空轉(zhuǎn)時(shí)間，更好地利用計(jì)算資源，也達(dá)到了數(shù)據(jù)并行處理使得系統(tǒng)優(yōu)化的效果。同時(shí)，由于執(zhí)行時(shí)間對(duì)齊，該算法產(chǎn)生的結(jié)果并不會(huì)因?yàn)槿魏我粋€(gè)節(jié)點(diǎn)的故障而執(zhí)行失敗，意外產(chǎn)生的情況能夠獨(dú)立執(zhí)行作業(yè)進(jìn)行處理，提升了系統(tǒng)的穩(wěn)定性。為實(shí)現(xiàn)以上功能，ICSC算法的偽代碼如下：

ICSC(Clusters C, Reducers R){

average = C.size/R.count;

C.sort();

largeC = new Array();

for (i = 1; i <= C.size; i++){

if (C[i - 1].size >= average){

largeC.add(C[i - 1]);

C.remove(i - 1);

}

}

// n為參與到包含split的reduce任務(wù)的節(jié)點(diǎn)數(shù)量。

n = ceiling(largeC.size/average)

for (i = 1; i <= n; i++){

R[i - 1].assign(largeC.split(n));

}

R.sort();

while(!C.empty){

C.assign(R);

R.sort();

}

}

算法1 ICSC算法偽代碼

圖4 CSC調(diào)度

圖5 ICSC調(diào)度

3 試驗(yàn)與分析

3.1 試驗(yàn)驗(yàn)證

為模擬實(shí)驗(yàn)室系統(tǒng)的數(shù)據(jù)收集和處理過程，本試驗(yàn)采用WordCount算法和人工生成的數(shù)據(jù)，在同構(gòu)環(huán)境下以固定大小的輸入數(shù)據(jù)進(jìn)行實(shí)驗(yàn)驗(yàn)證本文提出的算法。由于實(shí)驗(yàn)室系統(tǒng)的實(shí)際應(yīng)用環(huán)境下，數(shù)據(jù)屬性中的偏度和采樣度是影響數(shù)據(jù)處理階段的主要因素，因此本文通過控制變量法，驗(yàn)證在不同數(shù)據(jù)偏度和采樣度的情況下，ICSC算法可在實(shí)驗(yàn)室系統(tǒng)的數(shù)據(jù)處理階段有效減少耗時(shí)，達(dá)到流程優(yōu)化的效果。

具體而言，試驗(yàn)以JAVA JDK 11.0.13編制程序模擬12節(jié)點(diǎn)的集群，在同構(gòu)環(huán)境下使用WordCount算法處理128 MB人工數(shù)據(jù)。在實(shí)驗(yàn)室系統(tǒng)的環(huán)境中，節(jié)點(diǎn)間的傳輸采用100 Mbps帶寬。測(cè)量時(shí)間使用currentTimeMillis函數(shù)來測(cè)量自采樣階段Mapper到Reducer完成最后一個(gè)任務(wù)為止的時(shí)間差值。 通過試驗(yàn)得出結(jié)果如表1和表2所示。

表1 數(shù)據(jù)偏度在不同方式下對(duì)耗時(shí)的影響

表2 采樣度在不同方式下對(duì)耗時(shí)的影響

3.2 比較分析

為了更好地比較ICSC算法與CC和CSC調(diào)度法的耗時(shí)效果。根據(jù)在上述結(jié)果(表1和表2)的基礎(chǔ)上分別以輸入數(shù)據(jù)偏度和采樣率為自變量，以各自的執(zhí)行時(shí)間為因變量，得出不同算法條件下，各自的偏度和采樣度的效率，如圖6和圖7所示。

圖6 偏度與時(shí)間對(duì)比

圖7 采樣度與時(shí)間對(duì)比

如圖6所見，當(dāng)數(shù)據(jù)偏度大于0.1時(shí)，ICSC算法采樣時(shí)間更少，且偏度愈大優(yōu)勢(shì)更明顯，通過測(cè)算，在處理偏度為1.5的數(shù)據(jù)時(shí)，對(duì)比CSC算法，ICSC算法可節(jié)約10%的執(zhí)行時(shí)間；但當(dāng)數(shù)據(jù)偏度小于0.1時(shí)，由于ICSC在最后需要進(jìn)行一次額外的Reduce以合并被拆分的大數(shù)據(jù)塊，因此執(zhí)行時(shí)間是會(huì)稍大于CC和CSC調(diào)度法所耗時(shí)間。

同樣在圖7中，可以看到在不同采樣度的環(huán)境下，ICSC與兩階分區(qū)中算法的執(zhí)行時(shí)間對(duì)比。3種算法均在采樣度為0.1左右時(shí)執(zhí)行效率最高；而在0.1及以上時(shí)，ICSC算法比其它兩種算法執(zhí)行時(shí)間更短。這是因?yàn)樵诒镜匦暂^高的環(huán)境下，ICSC通過節(jié)省第二輪MapReduce的數(shù)據(jù)處理量以達(dá)到更優(yōu)的執(zhí)行時(shí)間。

4 結(jié)束語

本文針對(duì)兩階分區(qū)算法中在實(shí)驗(yàn)室系統(tǒng)的實(shí)際數(shù)據(jù)處理場(chǎng)景中存在的不足，對(duì)其調(diào)度算法和執(zhí)行流程進(jìn)行改進(jìn)。實(shí)驗(yàn)結(jié)果證明，改進(jìn)后的算法在數(shù)據(jù)存在高偏度和高采樣度的情況下，均可有效地減少了Mapper和Reducer節(jié)點(diǎn)空轉(zhuǎn)的時(shí)間，弱化了Reducer節(jié)點(diǎn)間的依賴性，縮短MapReduce的執(zhí)行時(shí)間，優(yōu)化了實(shí)驗(yàn)室系統(tǒng)數(shù)據(jù)處理階段的流程，從而高效地實(shí)現(xiàn)MapReduce框架下的實(shí)驗(yàn)室系統(tǒng)中數(shù)據(jù)處理的負(fù)載均衡。但值得注意的是該算法在本地性較差的環(huán)境下仍有待改進(jìn)，即數(shù)據(jù)采集系統(tǒng)與數(shù)據(jù)處理系統(tǒng)之間需要一定時(shí)間開銷的情況，因?yàn)樵谶@樣條件的環(huán)境下，數(shù)據(jù)處理節(jié)點(diǎn)間的傳輸時(shí)間遠(yuǎn)大于MapReduce作業(yè)的計(jì)算時(shí)間，此時(shí)節(jié)點(diǎn)間的傳輸時(shí)間成為ICSC算法執(zhí)行時(shí)間的主要影響因素。而ICSC算法主要針對(duì)計(jì)算時(shí)間進(jìn)行優(yōu)化，在執(zhí)行時(shí)間的優(yōu)化效果較為遜色，這也是本研究下一步工作努力的目標(biāo)和方向。