基于HadoopMapReduce作業(yè)調(diào)度方法研究

王玉芹

摘要：信息技術(shù)飛速發(fā)展，以及如今的互聯(lián)網(wǎng)的快速性的發(fā)展，云計算作為現(xiàn)在最為流行的計算模式，很好達到了傳統(tǒng)的互聯(lián)網(wǎng)式服務(wù)模式。關(guān)系到集群運行得很好的性能以及整個集群，能否有效利用各個的調(diào)度系統(tǒng)資源的關(guān)鍵，更好地來滿足信息用戶的需求。但是隨著不同應(yīng)用的環(huán)境變化，Hadoop 的調(diào)度算法很難充分的適應(yīng)用戶的應(yīng)用式環(huán)境變化的多種性需求，因此改進現(xiàn)有的業(yè)務(wù)調(diào)度式流程算法非常的重要。本文針對在數(shù)據(jù)的區(qū)域性方面初始作業(yè)的調(diào)度算法的不足性，在引入預(yù)先提取技術(shù)的基礎(chǔ)上，設(shè)計了一種基于資源提前取出的Hadoop MapReduce 作業(yè)調(diào)度算法。

關(guān)鍵詞：云計算；業(yè)務(wù)調(diào)度式流程數(shù)據(jù)的區(qū)域性；資源提前取出；Hadoop MapReduce.

中圖分類號：TP391 文獻標識碼：A 文章編號：1009-3044（2018）27-0010-02

1 概述

互聯(lián)網(wǎng)時代，大量的用戶信息數(shù)據(jù)需要得到及時迅速的處理，因此互聯(lián)網(wǎng)系統(tǒng)開發(fā)和操作人員需要研究一個能夠處理大數(shù)據(jù)并存儲大量異構(gòu)數(shù)據(jù)的數(shù)據(jù)庫和能夠計算不同結(jié)構(gòu)的算法。Google 云計算系統(tǒng)的關(guān)鍵技術(shù)由Hadoop開源實現(xiàn)并得到廣泛支持及應(yīng)用。彈性計算云[1]（elastic compute cloud，簡稱 EC2）系統(tǒng)是由推出Amazon的，為用戶提供信息存儲服務(wù)是該系統(tǒng)主要功能 ，所以說Amazon 是云計算服務(wù)商業(yè)化的先行者。云計算最為核心部分做簡單的介紹：MapReduce[2]模型、Google File System[3]簡稱GFS和BigTable。通過Map及Reduce兩個操作分布式，計算模型MapReduce[4]， 充分利用集群的效用最后進行分布式計算從而達到簡化的目的。首先，分布式的文件系統(tǒng)Google File System將文件分為若干相等的塊，再建立多個副本。BigTable。Hadoop是一種開源框架，作為GFS以及Google MapReduce的開源實現(xiàn)。可靠性上有一個好的業(yè)務(wù)調(diào)度流程算法顯得至關(guān)重要。

2 調(diào)度算法

本節(jié)主要討論的是資源提前取出的業(yè)務(wù)調(diào)度流程的算法，資源提前取出是為了解決以前的問題而被引入的。第一步：預(yù)先選出將要被候選的節(jié)點；第二步：非本地的以及待提前取出的map任務(wù)被預(yù)選出。第三步：資源提前取出被實現(xiàn)。非本地map任務(wù)在不用等待的情況下就可以被分配到存儲有輸入的數(shù)據(jù)的節(jié)點，這樣就大大提升了數(shù)據(jù)的區(qū)域性。對比于Hadoop原生的三種作業(yè)調(diào)度算法，本文基于 Hadoop0.2版本，又重新改進了算法，從而實現(xiàn)了更好的數(shù)據(jù)區(qū)域性。

提前選出節(jié)點以及預(yù)先選出任務(wù)及提前取出資源是該算法的三大步驟。

2.1 提前選出節(jié)點

一般在 Hadoop 中，不會有這一步，Pull通信模型在JobTracker 與以TaskTracker 之間得到了采用。作業(yè)在執(zhí)行時間上的減少，通過該算法來實現(xiàn)生成的。在這個策略的范圍內(nèi)，通過下面的方法可以調(diào)整運行之中的任務(wù)的數(shù)據(jù)的區(qū)域性：網(wǎng)絡(luò)狀況是處在運行的過程中的、和集群的負載動態(tài)，作業(yè)的運行時間實現(xiàn)了減少。不要對數(shù)據(jù)的區(qū)域性進行追求，而是要對整體性能在該算法中實現(xiàn)通盤的考慮。不足的是：在Hadoop平臺上還不能實現(xiàn)該算法。算法實現(xiàn)的是預(yù)先提出節(jié)點，能射都是的節(jié)點的提前取出。

Pull通信模型被JobTracker到TaskTracker的傳輸中使用。在與TaskTracker 的通信上不會主動化。輪詢實現(xiàn)了TT的預(yù)選。正好有空閑的節(jié)點被加入預(yù)選的隊伍中，這時因為有空閑計的算槽（slots）的節(jié)點，第一種情況相對于第二種更簡單。另一種是輪詢到的沒有空閑的計算槽（slots）的節(jié)點，需要判斷節(jié)點。具有下面的判斷條件：此節(jié)點的計算槽（slots）被釋放的快慢，即計算槽（slots）中的任務(wù)正被執(zhí)行的快慢性。最大概率釋放忙碌槽（slots）的是當(dāng)前任務(wù)中工作快的節(jié)點。

此算法主要的運算步驟如下：

第一步：執(zhí)行中的任務(wù)還需要多少時間來執(zhí)行完畢，這一問題TT（TaskTracker）給出了計算結(jié)果。

progressRate（T）=progress（T）/（currentTime-dispatchTime（T）） 任務(wù)在目前為止執(zhí)行的進度是progress（T），并且系統(tǒng)當(dāng)前的時間是currentTime ，同時調(diào)度任務(wù)T的時間是dispatchTime（T）。則可以一個任務(wù)的剩余時間被估算為 timeLeft?？梢酝ㄟ^任務(wù)的進度增長率這一途徑來計算剩余的所需時間。下面所示：

timeLeft=（1-progress（T））/progressRate（T））

第二步：此步驟實現(xiàn)的是一個數(shù)據(jù)塊經(jīng)過 Hadoop 在TT（TaskTracker）之間傳輸時一共花費的時間。通常的情況下，Hadoop有固定的文件塊大小，大多數(shù)的情況下被設(shè)置為64M，在數(shù)據(jù)塊間的數(shù)據(jù)傳輸，需要花費的時間被記為timePerBlock，下面所示：

timePerBlock=blockSize/tranRate

blockSize在一般的情況是64M，集群網(wǎng)絡(luò)在理論上的傳輸速率是tranRate，一般是100Mbps。

第三步：進行了對比，實現(xiàn)了節(jié)點的選取。此節(jié)點被心跳信號輪詢時，判斷該節(jié)點上是否有此map 任務(wù)處在正在運行的任務(wù)中，該節(jié)點則會被加入預(yù)選的隊列里，因為數(shù)據(jù)提前取出是在此任務(wù)完成之前進行的。當(dāng)存在這樣的任務(wù)節(jié)點，滿足timePerBlock 小于timeLeft，則在candidateTTs的集合中加入了此計算節(jié)點，如果不滿足，此次提前取出的操作被退出。

按照相差時間的多少對此隊列進行由小到大的排序，這樣做的目的是實現(xiàn)了從預(yù)選的隊列中，來進行節(jié)點的預(yù)選。每隔一個心跳時段對已經(jīng)排好序的隊列進行一次更新，時間的有效性得到了提前選出節(jié)點的保證。

2.2 預(yù)選作業(yè)任務(wù)

候選的計算節(jié)點集合在上面得到了講解，作業(yè)任務(wù)的提前取出在這一節(jié)中得到了介紹。需要被預(yù)期的數(shù)據(jù)塊在map任務(wù)被選擇之后才能確認。提前取出工作的成功完成是因為下面的條件決定的：每一個map 任務(wù)所處理的數(shù)據(jù)塊不一樣，因此首先要做的是預(yù)選出待提前取出的 map 任務(wù)。

第一步：執(zhí)行中的任務(wù)還需要多少時間來執(zhí)行完畢，這一問題TT（TaskTracker）給出了計算結(jié)果。

progressRate（T）=progress（T）/（currentTime-dispatchTime（T））

任務(wù)在目前為止執(zhí)行的進度是progress（T），并且系統(tǒng)當(dāng)前的時間是currentTime ，同時調(diào)度任務(wù)T的時間是dispatchTime（T）。則可以一個任務(wù)的剩余時間被估算為 timeLeft?？梢酝ㄟ^任務(wù)的進度增長率這一途徑來計算剩余的所需時間。下面（2）所示：

timeLeft=（1-progress（T））/progressRate（T））

第二步：此步驟實現(xiàn)的是一個數(shù)據(jù)塊經(jīng)過 Hadoop 在TT（TaskTracker）之間傳輸時一共花費的時間。通常的情況下，Hadoop有固定的文件塊大小，大多數(shù)的情況下被設(shè)置為64M，在數(shù)據(jù)塊間的數(shù)據(jù)傳輸，需要花費的時間被記為timePerBlock，下面所示：

timePerBlock=blockSize/tranRate blockSize在一般的情況是64M，集群網(wǎng)絡(luò)在理論上的傳輸速率是tranRate，一般是100Mbps。

第三步：進行了對比，實現(xiàn)了節(jié)點的選取。此節(jié)點被心跳信號輪詢時，判斷該節(jié)點上是否有此map 任務(wù)處在正在運行的任務(wù)中，該節(jié)點則會被加入到預(yù)選的隊列里，因為數(shù)據(jù)提前取出是任務(wù)完成之前進行的。當(dāng)存在這樣的任務(wù)節(jié)點，滿足timePerBlock 小于timeLeft，則在candidateTTs的集合中加入了此計算節(jié)點提前取出的操作被退出。

第一步：提前取出候選的節(jié)點targetTT，最近得到更新的候選列表被讀取到，排名靠前的TaskTracker得到了選取，任務(wù)的剩余的時間上是最小，并且在完成任務(wù)之前，計算節(jié)點的提前取出會有足夠的時間來實現(xiàn)。

第二步：完成任務(wù)的選取。首先要選取失敗的任務(wù)failedMaps，這個失敗任務(wù)是從正在進行作業(yè)之中選擇出來的。在作業(yè)的過程中，優(yōu)先級最高的任務(wù)就是失敗任務(wù)。當(dāng)failedMaps 不為空時，對于失敗次數(shù)最多的任務(wù)，是從列表中被選取的，相對于上一個步驟這個任務(wù)被選取出來的，篩選targetTT 的數(shù)據(jù)的區(qū)域性。整個的算法在任務(wù)為node-local時結(jié)束。此任務(wù)被加入到該任務(wù)中，是該任務(wù)不是node-local時進行的，記為toPrefetchMap。

第三步：當(dāng)toPrefetchMap還處于空的狀態(tài)時，任務(wù)的選擇是按照選取出來的targetTT的數(shù)據(jù)的區(qū)域性實現(xiàn)的，整個算法的結(jié)束是在nonRunningCache存在node-local任務(wù)的條件下進行的。node-local 任務(wù)在nonRunningCache中不存在的話 ，toPrefetcheMap中加入了rack-local 的任務(wù)，是按照優(yōu)先級進行選擇的。完成了本次作業(yè)任務(wù)的選擇，在該算法沒有結(jié)束時，實現(xiàn)下一步的操作，即進行數(shù)據(jù)塊間的資源的預(yù)選。

2.3 提前取出塊間數(shù)據(jù)資源

數(shù)據(jù)塊間資源的提前取出機制是這一節(jié)我們介紹的重點，從步驟一以及步驟二中分別選擇出待提前取出的節(jié)點和任務(wù)，然后以待提前取出任務(wù)的元數(shù)據(jù)信息為依據(jù)把待提前取出map任務(wù)的數(shù)據(jù)塊從待提前取出的節(jié)點 targetTT 上面提前取出出來。待提前取出資源所在的源節(jié)點可以根據(jù)待提前取出map任務(wù)元數(shù)據(jù)的信息以及固定的策略來確定：輸入數(shù)據(jù)被分為了大小為64M的分片split，map任務(wù)和數(shù)據(jù)塊是一一對應(yīng)，每個數(shù)據(jù)塊又和每個數(shù)據(jù)分片 split一一對應(yīng)。一個數(shù)據(jù)塊里面又包含了按照固定策略在機器節(jié)點上分布的3個副本。因此要提前取出資源的話首先要選取和targetTT 距離最為相近的機器上面的節(jié)點數(shù)據(jù)分片來進行提前取出。具體算法如下：

步驟1：源節(jié)點集合 sourceTTs的確定：把組成源節(jié)點集合的位置信息從 toPretchMap 中的 TaskInProgress 類之中讀取出來，然后依據(jù)某些一定的策略選擇某個適合源節(jié)點來進行提前取出操作。

步驟2：選擇某一個源節(jié)點，這個源節(jié)點的集合中節(jié)點和targetTT之間的距離可以從Hadoop 集群中的網(wǎng)絡(luò)拓撲的信息來獲取，而這一網(wǎng)絡(luò)拓撲信息可以從Hadoop集群中conf 目錄之下的 topology.data 文件獲得，之后選擇和targetTT 最為相近的源節(jié)點作為sourceTT。targetTT和sourceTT這兩個節(jié)點之間的距離可以有下述幾種情況：兩個節(jié)點位于同一節(jié)點，距離記為0；兩個節(jié)點在同一機架上，距離為2；兩個節(jié)點在相同的數(shù)據(jù)中心但在不同的機架上，距離記為4；最后一種是兩個節(jié)點在不同數(shù)據(jù)中心之上，這種情況，距離就記為6。

步驟 3：把待提前取出map 任務(wù)的所需要數(shù)據(jù)從步驟 2 中選取的合適源節(jié)點 sourceTT上傳送到targetTT 節(jié)點之上。

步驟 4：主節(jié)點元信息文件更新，map 任務(wù)中出現(xiàn)了空閑的計算槽（slots）時，給主節(jié)點發(fā)送“心跳”信息來要求新任務(wù)，就是toPrefetchMap。

3 總結(jié)

通過資源提前取出技術(shù)的引入很大改進了任務(wù)調(diào)度時的數(shù)據(jù)的區(qū)域性。此研究方法絡(luò)資源，而且儉省經(jīng)濟負擔(dān)，為用戶提供了方便，從而在運算速度和存儲功能上也得到了很好的提升。

參考文獻：

[1] Amazon EC2[EB/OL].http：//aws.amazon.come/ec2， 2011.

[2] JEFFREY DEAN， SANJAY GHEMAWAT. MapReduce： simplified data processing on largec！usters[J]. Communications of the ACM， 2008，51（1）：107-113.

[3] Sanjay Ghemawat， Howard Gobioff， Shun-Tak Leung. The Google file system. In Proceedings ofthe nineteenth ACM symposium on Operating systems principles. New York： ACM， 2003：29-43.

[4] Fay Chang， Jeffrey Dean， Sanjay Ghemawat， et al. Bigtable： A Distributed Storage System forStructured Data[J]. ACM Transactions on Computer Systems， 2008，26（2）：1-26.

[5] 王凱，吳泉源，楊樹強.一種多用戶 MapReduce 集群的作業(yè)調(diào)度算法的設(shè)計與實現(xiàn)[J].計算機與現(xiàn)代化，2010：23-28.

[6] 羅國瑋，蘭瑞樂.基于云計算的高?？蒲袑嶒炂脚_構(gòu)建研究[J].實驗技術(shù)與管理.2012（4）. 115-117. [通聯(lián)編輯： ]