摘  要： 針對(duì)傳統(tǒng)MapReduce框架中任務(wù)節(jié)點(diǎn)和工作節(jié)點(diǎn)的失效問(wèn)題，提出了在配置備份節(jié)點(diǎn)的分層主從式MapReduce框架中加入單元集群的處理方法。在改進(jìn)框架中，任務(wù)處理的最小單位是單元集群，當(dāng)單元集群中的某個(gè)工作節(jié)點(diǎn)失效或者超過(guò)時(shí)間闕值時(shí)，子任務(wù)節(jié)點(diǎn)則選擇該單元集群中的空閑工作節(jié)點(diǎn)來(lái)分配任務(wù)，并且不需要重新傳輸任務(wù)文件分塊，這既節(jié)省了工作節(jié)點(diǎn)重選擇的時(shí)間，又降低了網(wǎng)絡(luò)傳輸?shù)膲毫?。使用該框架針?duì)不同數(shù)量的數(shù)據(jù)塊進(jìn)行實(shí)驗(yàn)，工作節(jié)點(diǎn)的災(zāi)難恢復(fù)時(shí)間均可以節(jié)省25 ms左右，證明了單元集群的處理方法可以有效解決工作節(jié)點(diǎn)的失效問(wèn)題。
關(guān)鍵詞： Hadoop架構(gòu)；MapReduce框架；任務(wù)節(jié)點(diǎn)；工作節(jié)點(diǎn)；備份節(jié)點(diǎn)；節(jié)點(diǎn)失效；單元集群

    隨著互聯(lián)網(wǎng)的迅猛發(fā)展，基于互聯(lián)網(wǎng)的數(shù)據(jù)蘊(yùn)含著豐富的信息，云計(jì)算[1]就是在這樣一種對(duì)大規(guī)模計(jì)算能力的強(qiáng)烈需求環(huán)境下發(fā)展起來(lái)的。Hadoop[2]云計(jì)算平臺(tái)是Apache的一個(gè)分布式計(jì)算開(kāi)源框架，其核心是MapReduce[3]和Hadoop Distribute File System[4]。
1 MapReduce框架簡(jiǎn)介
    MapReduce是一種處理海量數(shù)據(jù)的并行編程框架，其主要思想是將要執(zhí)行的問(wèn)題進(jìn)行分割，數(shù)據(jù)被分割后通過(guò)Map函數(shù)的程序?qū)?shù)據(jù)映射到不同區(qū)塊，分配給PC集群來(lái)處理，再通過(guò)Reduce函數(shù)的程序?qū)⒔Y(jié)果匯總，輸出要得到的結(jié)果。
1.1 算法流程
    用戶程序調(diào)用MapReduce函數(shù)時(shí)會(huì)引起如下操作：
    (1)利用MapReduce函數(shù)庫(kù)把輸入文件分成M塊，每塊大概16 MB～64 MB，并在集群中的Worker節(jié)點(diǎn)上執(zhí)行程序的備份；
    (2)Master節(jié)點(diǎn)找出空閑的Worker節(jié)點(diǎn)并為它們分配子任務(wù)；
    (3)被分配到Map任務(wù)的Worker節(jié)點(diǎn)讀取已經(jīng)分割好的文件塊，計(jì)算生成中間結(jié)果，中間結(jié)果暫時(shí)緩沖到內(nèi)存；
    (4)中間結(jié)果周期性地寫(xiě)入本地，中間文件通過(guò)分區(qū)函數(shù)分成R個(gè)分區(qū)，并將它們?cè)诒镜卮疟P(pán)的位置信息發(fā)送給Master節(jié)點(diǎn)；
    (5)被分配到Reduce任務(wù)的Worker節(jié)點(diǎn)根據(jù)Master提供的中間文件位置信息遠(yuǎn)程讀取相應(yīng)Worker節(jié)點(diǎn)的中間文件；
    (6)執(zhí)行Reduce，計(jì)算并輸出結(jié)果。
1.2 Hadoop中的MapReduce
    Hadoop中對(duì)MapReduce的處理是在原算法中加入了兩個(gè)處理：Split(分割)和Shuffle(混合)，框架如圖1所示。

    圖1中，Split過(guò)程可以看作是Input的再次分割，用以控制分塊的大小，Shuffle過(guò)程可以看作是在執(zhí)行Map任務(wù)的Worker節(jié)點(diǎn)上執(zhí)行本地的Reduce過(guò)程。
2 MapReduce框架的相關(guān)改進(jìn)
2.1 MapReduce Online
    MapReduce Online框架[5]將Map任務(wù)產(chǎn)生的中間數(shù)據(jù)在執(zhí)行映射任務(wù)的Worker節(jié)點(diǎn)和執(zhí)行規(guī)約任務(wù)的Worker節(jié)點(diǎn)間通過(guò)管道進(jìn)行傳輸，使得下游數(shù)據(jù)元素可以在上游數(shù)據(jù)元素完成執(zhí)行前開(kāi)始消耗數(shù)據(jù)，增加了并行機(jī)會(huì)，提高了利用率，減少了響應(yīng)時(shí)間。
2.2 Map-Balance-Reduce
    Map-Balance-Reduce模型[6]針對(duì)MapReduce模型中存在的多個(gè)Reduce任務(wù)之間完成時(shí)間差別較大的問(wèn)題而提出。通過(guò)添加Balance任務(wù)對(duì)Map任務(wù)處理完成的中間數(shù)據(jù)進(jìn)行均衡操作，使分配到Reduce任務(wù)節(jié)點(diǎn)的數(shù)據(jù)比較均衡，從而確保Reduce任務(wù)的完成時(shí)間基本一致。
2.3 Shuffle階段的優(yōu)化與重構(gòu)
    該方案在執(zhí)行Map任務(wù)的Worker節(jié)點(diǎn)上壓縮產(chǎn)生的中間文件、重構(gòu)遠(yuǎn)程數(shù)據(jù)拷貝傳輸協(xié)議（從Http協(xié)議到UDT協(xié)議）以及Reduce端內(nèi)存分配優(yōu)化3個(gè)方面[7]來(lái)優(yōu)化和重構(gòu)Shuffle階段，從而達(dá)到提高M(jìn)apReduce的計(jì)算性能。
3 傳統(tǒng)MapReduce框架中節(jié)點(diǎn)失效的問(wèn)題
    MapReduce框架不同于其他傳統(tǒng)系統(tǒng)框架設(shè)計(jì)的一點(diǎn)在于其將框架中的每個(gè)節(jié)點(diǎn)都當(dāng)作一種不穩(wěn)定的資源來(lái)對(duì)待，每個(gè)節(jié)點(diǎn)的失效都當(dāng)作是一種系統(tǒng)常態(tài)并且有著相應(yīng)的處理方式。
3.1 Master節(jié)點(diǎn)的失效問(wèn)題
    MapReduce框架中，唯一的Master節(jié)點(diǎn)管理數(shù)據(jù)分塊、Worker節(jié)點(diǎn)分配、記錄節(jié)點(diǎn)狀態(tài)以及選擇空閑節(jié)點(diǎn)等眾多事務(wù)，一旦Master節(jié)點(diǎn)失效，“重新執(zhí)行”的操作對(duì)之前的工作是一種極大的浪費(fèi)并且大大延長(zhǎng)了任務(wù)執(zhí)行時(shí)間，這對(duì)需要進(jìn)行海量數(shù)據(jù)計(jì)算的系統(tǒng)是不能接受的。參考文獻(xiàn)[8]提出了“配置備份節(jié)點(diǎn)的分層主從式MapReduce框架”的解決方案，該方案可以在一定程度上緩解Master節(jié)點(diǎn)的任務(wù)壓力并且提高了Master節(jié)點(diǎn)的安全性，具體框架如圖2所示。

    在這個(gè)MapReduce框架中，各個(gè)節(jié)點(diǎn)的工作職責(zé)是：
    (1)主任務(wù)節(jié)點(diǎn)：主要對(duì)集群進(jìn)行管理以及第一次任務(wù)的委派，在任務(wù)委派完之后不需要擔(dān)心任務(wù)的完成情況，只需要專心地完成與子任務(wù)節(jié)點(diǎn)的聯(lián)系，并對(duì)元數(shù)據(jù)進(jìn)行管理，同時(shí)應(yīng)對(duì)節(jié)點(diǎn)失效的問(wèn)題。
    (2)子任務(wù)節(jié)點(diǎn)：主要與用戶程序進(jìn)行通信，完成用戶程序發(fā)出的任務(wù)并將結(jié)果返回給用戶程序；它們直接受主任務(wù)節(jié)點(diǎn)的管理，也直接管理工作節(jié)點(diǎn)，起到承上啟下的作用。
    (3)工作節(jié)點(diǎn)：直接受子任務(wù)節(jié)點(diǎn)的管理，完成分配的任務(wù)。
3.2 Worker節(jié)點(diǎn)的失效問(wèn)題
    對(duì)于Worker節(jié)點(diǎn)的狀態(tài)判定，MapReduce框架是讓Master節(jié)點(diǎn)周期性地ping每個(gè)Worker節(jié)點(diǎn)，當(dāng)無(wú)法得到Worker節(jié)點(diǎn)的應(yīng)答時(shí)，Master節(jié)點(diǎn)就認(rèn)為這個(gè)節(jié)點(diǎn)是失效的?？梢詫⑹У腤orker節(jié)點(diǎn)上的任務(wù)以及所處狀態(tài)處理措施分為4類，如表 1 所示。

    針對(duì)Worker節(jié)點(diǎn)的失效問(wèn)題，采用引入單元集群的處理方法，該方案可以方便Master節(jié)點(diǎn)對(duì)Worker節(jié)點(diǎn)的管理并降低Worker節(jié)點(diǎn)失效的影響。
4 改進(jìn)型MapReduce框架設(shè)計(jì)
4.1 單元集群
    在傳統(tǒng)MapReduce框架中，龐大的機(jī)器集群是進(jìn)行Map任務(wù)和Reduce任務(wù)的計(jì)算資源，集群中除了Master節(jié)點(diǎn)之外的每個(gè)機(jī)器都可以是Worker節(jié)點(diǎn)。通常情況下，這個(gè)集群的數(shù)量是以萬(wàn)為計(jì)數(shù)單位的。為了讓Master節(jié)點(diǎn)更好地管理這樣一個(gè)龐大的機(jī)器集群，引入了“單元集群”的概念。
4.1.1 基本概念
    單元集群就是擁有X個(gè)Worker節(jié)點(diǎn)的一個(gè)小機(jī)器集群，整個(gè)機(jī)器集群由多個(gè)單元集群組成，單元集群從Master節(jié)點(diǎn)處接受X個(gè)任務(wù)，該單元集群中Worker節(jié)點(diǎn)也都獲取這X個(gè)任務(wù)所對(duì)應(yīng)的文件分塊并將其存儲(chǔ)在自己的本地磁盤(pán)中，每個(gè)Worker節(jié)點(diǎn)執(zhí)行這X個(gè)任務(wù)中其中一個(gè)，如節(jié)點(diǎn)1執(zhí)行任務(wù)1，節(jié)點(diǎn)2執(zhí)行任務(wù)2，…，節(jié)點(diǎn)X執(zhí)行任務(wù)X。
    Master節(jié)點(diǎn)對(duì)其管理的單元集群設(shè)置一個(gè)“任務(wù)執(zhí)行時(shí)間闕值T”，單元集群中的Worker節(jié)點(diǎn)的3個(gè)狀態(tài)以及處理措施如下：
    (1)正常狀態(tài)：正常執(zhí)行完X個(gè)任務(wù)，然后開(kāi)始向Mas-
ter節(jié)點(diǎn)請(qǐng)求新的任務(wù)或者接受其他單元集群的任務(wù)；
    (2)超過(guò)時(shí)間闕值狀態(tài)：當(dāng)單元集群中的Worker節(jié)點(diǎn)i執(zhí)行當(dāng)前任務(wù)i的時(shí)間超過(guò)了T，則啟動(dòng)該單元集群中的空閑節(jié)點(diǎn)k執(zhí)行任務(wù)i，因?yàn)樵搯卧褐械拿總€(gè)節(jié)點(diǎn)初始化時(shí)已經(jīng)獲取了任務(wù)i對(duì)應(yīng)的文件分塊，所以節(jié)點(diǎn)k可立即開(kāi)始執(zhí)行任務(wù)i。當(dāng)節(jié)點(diǎn)k也超過(guò)時(shí)間闕值T時(shí)，處理方法一樣，直到該任務(wù)在任一節(jié)點(diǎn)上執(zhí)行完成；
    (2)失效狀態(tài)：當(dāng)單元集群中的Worker節(jié)點(diǎn)j失效時(shí)，Master節(jié)點(diǎn)優(yōu)先在該單元集群中尋找空閑節(jié)點(diǎn)（或已經(jīng)執(zhí)行完本身任務(wù)的節(jié)點(diǎn)）來(lái)執(zhí)行這個(gè)失效節(jié)點(diǎn)j上的任務(wù)j，同樣也不需要進(jìn)行文件分塊的重新獲取，如果失效節(jié)點(diǎn)j所在的單元集群中沒(méi)有空閑節(jié)點(diǎn)或已執(zhí)行完任務(wù)的節(jié)點(diǎn)，Master節(jié)點(diǎn)就去尋找空閑的單元集群的Worker節(jié)點(diǎn)來(lái)執(zhí)行該任務(wù)。
4.1.2 主要改進(jìn)
    單元集群的引入能夠很好地解決Worker節(jié)點(diǎn)失效和工作效率不高的問(wèn)題，單元集群所帶來(lái)的優(yōu)化如下：
    (1)單元集群中的Worker節(jié)點(diǎn)失效或者執(zhí)行時(shí)間超過(guò)闕值時(shí)，Master的處理方案是先從失效節(jié)點(diǎn)所在的單元集群中尋找合適的節(jié)點(diǎn)，節(jié)省了在整個(gè)機(jī)器集群中尋找合適節(jié)點(diǎn)的時(shí)間；
    (2)單元集群中的節(jié)點(diǎn)在初始化時(shí)都獲取了X個(gè)任務(wù)文件分塊，正常工作的節(jié)點(diǎn)可以立即執(zhí)行失效節(jié)點(diǎn)的任務(wù)而不需要重新獲取文件分塊，這降低了網(wǎng)絡(luò)傳輸?shù)膲毫?，Worker節(jié)點(diǎn)失效或者執(zhí)行時(shí)間超過(guò)闕值不會(huì)對(duì)系統(tǒng)效率有太大影響；
    (3)Master節(jié)點(diǎn)的執(zhí)行單位從單個(gè)節(jié)點(diǎn)變成單元集群，其只需要考慮單元集群的失效率和工作效率。只要單元集群中還有正常工作的Worker節(jié)點(diǎn)，Master節(jié)點(diǎn)就認(rèn)為該單元集群仍然處于正常工作狀態(tài)。
4.1.3 單元集群的X的取值
    首先，X不宜偏大，單元集群中的每個(gè)節(jié)點(diǎn)都需要存儲(chǔ)這個(gè)單元集群所要執(zhí)行任務(wù)對(duì)應(yīng)的文件分塊，節(jié)點(diǎn)過(guò)多會(huì)在初始化時(shí)網(wǎng)絡(luò)傳輸過(guò)多的文件分塊給每個(gè)節(jié)點(diǎn)，將增加每個(gè)節(jié)點(diǎn)的網(wǎng)絡(luò)傳輸壓力并且延長(zhǎng)初始化時(shí)間從而降低系統(tǒng)執(zhí)行效率。單元集群在未執(zhí)行完成分配的所有任務(wù)之前是不接受其他單元集群任務(wù)的，X偏大就有可能導(dǎo)致單元集群的節(jié)點(diǎn)不能得到充分利用；其次，X不宜偏小，單元集群節(jié)點(diǎn)偏少時(shí)，若單元集群中出現(xiàn)節(jié)點(diǎn)失效或者運(yùn)行效率低下的情況，且該單元集群中其他的節(jié)點(diǎn)也在執(zhí)行自己的任務(wù)，而這個(gè)任務(wù)又不能及時(shí)交給其他單元集群去執(zhí)行，則失效節(jié)點(diǎn)所負(fù)責(zé)的任務(wù)執(zhí)行就會(huì)成為系統(tǒng)效率的瓶頸，從而導(dǎo)致整個(gè)系統(tǒng)的效率偏低，而且單元集群節(jié)點(diǎn)偏少也在一定程度上增大了單元集群的失效率。
4.2 改進(jìn)框架設(shè)計(jì)
    改進(jìn)型MapReduce框架結(jié)合了“配置備份節(jié)點(diǎn)的分層主從式MapReduce框架”和“單元集群”，能夠同時(shí)解決Master節(jié)點(diǎn)和Worker節(jié)點(diǎn)失效問(wèn)題，如圖3所示。

    通過(guò)引入兩層Master節(jié)點(diǎn)來(lái)解決Master節(jié)點(diǎn)任務(wù)壓力過(guò)大的問(wèn)題，通過(guò)對(duì)Worker節(jié)點(diǎn)采取單元集群的處理來(lái)解決其失效和效率不高的問(wèn)題，以下通過(guò)實(shí)驗(yàn)驗(yàn)證這樣的改進(jìn)框架設(shè)計(jì)是合理且有效的。
5 實(shí)驗(yàn)結(jié)果分析
    實(shí)驗(yàn)所使用的測(cè)試系統(tǒng)是目前運(yùn)用MapReduce最廣泛的搜索領(lǐng)域，但實(shí)驗(yàn)中使用的測(cè)試數(shù)據(jù)相對(duì)于當(dāng)前搜索網(wǎng)站所涉及的數(shù)據(jù)要簡(jiǎn)單。
5.1 實(shí)驗(yàn)環(huán)境
    集群中主任務(wù)節(jié)點(diǎn)、子任務(wù)節(jié)點(diǎn)以及工作節(jié)點(diǎn)都使用相同的軟硬件，Hadoop為0.22.1版本，任務(wù)文件塊的數(shù)量設(shè)置為500、1 000、1 500，并且文件塊的大小設(shè)置為64 MB。
    針對(duì)傳統(tǒng)框架，使用4臺(tái)PC來(lái)模擬Master節(jié)點(diǎn)、Worker節(jié)點(diǎn)以及客戶端，其中Master節(jié)點(diǎn)和客戶端安裝在一臺(tái)計(jì)算機(jī)上，另外3臺(tái)PC作為Worker節(jié)點(diǎn)使用。
    針對(duì)改進(jìn)框架，使用15臺(tái)PC來(lái)模擬主任務(wù)節(jié)點(diǎn)、子任務(wù)節(jié)點(diǎn)、工作節(jié)點(diǎn)、備份節(jié)點(diǎn)以及客戶端，取單元集群的X=5，主任務(wù)節(jié)點(diǎn)和客戶端安裝在一臺(tái)PC上，2個(gè)子任務(wù)節(jié)點(diǎn)安裝在2臺(tái)PC上，2個(gè)備份節(jié)點(diǎn)分別安裝在2臺(tái)PC上，另外10臺(tái)PC分別作為2個(gè)單元集群中的Worker節(jié)點(diǎn)。
5.2 實(shí)驗(yàn)數(shù)據(jù)及結(jié)果分析
    首先，在客戶端隨機(jī)生成50 000個(gè)整數(shù)(0～49 999)，在傳統(tǒng)框架下，將這些整數(shù)文件分成多個(gè)文件塊，并均分存儲(chǔ)到3個(gè)Worker節(jié)點(diǎn)上;若在改進(jìn)框架下，將這些整數(shù)文件分成多個(gè)文件塊，并均分這些文件塊到2個(gè)單元集群上；然后，客戶端將一個(gè)MapReduce任務(wù)（搜索最大值）交給機(jī)器集群來(lái)進(jìn)行實(shí)驗(yàn)。
    在傳統(tǒng)框架和改進(jìn)框架下的運(yùn)行結(jié)果如表2所示。
    Worker節(jié)點(diǎn)的災(zāi)難恢復(fù)時(shí)間是實(shí)驗(yàn)重點(diǎn)關(guān)注的參數(shù)指標(biāo)。節(jié)點(diǎn)的災(zāi)難恢復(fù)時(shí)間包括：重新選擇節(jié)點(diǎn)、IP地址的遷移、網(wǎng)絡(luò)傳輸時(shí)間以及數(shù)據(jù)塊的遷移時(shí)間，在單元集群改進(jìn)框架中，Worker節(jié)點(diǎn)的災(zāi)難恢復(fù)時(shí)間只包括重新選擇節(jié)點(diǎn)和IP地址的遷移這兩部分時(shí)間。從表2可以看出改進(jìn)框架相比較傳統(tǒng)框架，其Worker節(jié)點(diǎn)的災(zāi)難恢復(fù)時(shí)間降低了25 ms左右。
    通過(guò)測(cè)試數(shù)據(jù)可以得出：?jiǎn)卧旱母倪M(jìn)框架與傳統(tǒng)框架相比，降低了Worker節(jié)點(diǎn)的災(zāi)難恢復(fù)時(shí)間，從而在一定程度上解決了Worker節(jié)點(diǎn)的失效問(wèn)題。
    針對(duì)傳統(tǒng)MapReduce框架中存在的Master節(jié)點(diǎn)和Worker節(jié)點(diǎn)失效的問(wèn)題，采用了在“配置備份節(jié)點(diǎn)的分層主從式MapReduce框架”中加入單元集群的方案，單元集群包括多個(gè)Worker節(jié)點(diǎn)，這些Worker節(jié)點(diǎn)上都備份了所在單元集群所要處理的所有任務(wù)文件分塊，當(dāng)Worker節(jié)點(diǎn)失效或超過(guò)時(shí)間闕值時(shí)，Master節(jié)點(diǎn)會(huì)選擇其所在單元集群中的空閑節(jié)點(diǎn)來(lái)執(zhí)行失效節(jié)點(diǎn)的任務(wù)，這不僅將單一Worker節(jié)點(diǎn)的失效問(wèn)題轉(zhuǎn)移到單元集群上，而且也解決了當(dāng)出現(xiàn)節(jié)點(diǎn)失效時(shí)還需要重新傳輸任務(wù)文件分塊的問(wèn)題。單元集群的提出雖然在一定程度上解決了Master節(jié)點(diǎn)和Worker節(jié)點(diǎn)的失效問(wèn)題，但是因?yàn)樾枰诔跏蓟麄€(gè)系統(tǒng)時(shí)多傳輸一部分任務(wù)文件分塊，這樣就犧牲了一部分的系統(tǒng)性能，并且在單元集群的X取值上并沒(méi)有一個(gè)特定的方案，只能通過(guò)大量的實(shí)驗(yàn)測(cè)試才能得到合適的取值，這也是未來(lái)研究的一個(gè)方向。
參考文獻(xiàn)
[1] 陳康，鄭緯民.云計(jì)算：系統(tǒng)實(shí)例與研究現(xiàn)狀[J].軟件學(xué)報(bào)，2009，20(5)：1337-1348.
[2] WHITE T.Hadoop：the definitive guide[M].California：O′Reilly Media，2012.
[3] DEAN J，GHEMAWAT S.MapReduce: simplified data processing on large clusters[J].Communications of the ACM，2008，51(1)：107-113.
[4] BORTHAKUR D.HDFS architecture guide[DB/OL].Hadoop apache project.(2008-02-14).[2013-04-22].http://hadoop.apache.org/common/docs/current/hdfsdesign.pdf.
[5] CONDIE T，CONWAY N，ALVARO P，et al.MapReduce  online[C].Proceedings of the 7th USENIX Conference on  Networked Systems Design and Implementation，2010：21-21.
[6] 李玉林，董晶.基于Hadoop的MapReduce模型的研究與改進(jìn)[J].計(jì)算機(jī)工程與設(shè)計(jì)，2012，33(8)：3110-3116.
[7] 彭輔權(quán)，金蒼宏，吳明暉，等.MapReduce中shuffle優(yōu)化與重構(gòu)[J].中國(guó)科技論文，2012，7(4)：241-245.
[8] 周一可.云計(jì)算下MapReduce編程模型可用性的研究與優(yōu)化[D].上海：上海交通大學(xué)，2011.