Hadoop 平臺下計算能力調度算法的改進與實現

戴小平，張宜力

DAI Xiaoping,ZHANG Yili

安徽工業大學 計算機學院，安徽 馬鞍山243002

Computer Science School,Anhui University of Technology,Ma’anshan,Anhui 243002,China E-mail：jszhangyili@163.com

1 引言

云計算是一種新的計算模型，是隨著并行計算、分布式計算和網格計算技術的發展而來的，在學術界和工業界產生了巨大的影響，Google、Amazon、Microsoft 等大公司是云計算的先行者[1]在眾多的云計算解決方案中，Google提出的基于分布式文件系統GFS 建立的廉價集群實現MapReduce 編程思想[2-3]的方案因其應用簡單、高效的特點得到了廣泛支持，解決了許多大規模數據的計算問題。而Hadoop 作為Google 相關思想開源實現的一個分布式系統，除了成本少，有較高的效率和伸縮性外，安全性也得到了很好的保障[4-5]。

Hadoop 平臺中作業調度器對平臺性能至關重要，作業調度技術作為Hadoop 平臺的核心技術之一，不僅對Hadoop 平臺的整體性能有影響，還對系統資源是否有效利用起到了至關重要的作用。但是目前的幾種作業調度算法都存著資源浪費，作業響應時間過長，沒有考慮作業配置及不能夠滿足作業多樣性的服務要求等缺點[6]。

本文基于計算能力調度算法實現了一種基于優先級的計算能力加權調度算法，能夠有效地解決目前Hadoop的作業調度算法中存在的一些問題，并在Hadoop平臺的實驗中表現良好。

2 MapReduce編程思想

MapReduce是Google公司的核心計算模型，它將復雜的運行于大規模集群上的并行計算過程高度地抽象到了兩個函數：Map（映射）和Reduce（歸約）。簡單的MapReduce 的過程如圖1 所示，MapReduce 首先將用 戶輸入的數據通過Split 方法將其分成若干Block（每個Block的默認大小為64 MB，每一個Block由一個Mapper管理，在整個MapReduce 中Reducer 是通過配置來得到的，而Mapper 的數量則是由Block 的數量來決定）每一個Mapper 將運行在該Block 所在的本機中，所有的臨時數據和中間數據也都會存儲在本機中，當所有的Mapper運行完成后，Reducer 便會開始運行，將所有Mapper 的輸出作為它的輸入，最后將結果輸出到GFS 中[7]。

圖1 給出了MapReduce 執行過程，分為Map 階段以及Reduce 階段，這兩個階段都使用了集群中的所有節點。在這兩個階段之間還有一個中間的分類階段，即將中間結果包含相同Key 交給同一個Reduce 函數去執行。MapReduce操作相關的類型如下：

map(keyin,valuein)→list(keyout,valueintermediate)

reduce(keyout,list(valueintermediate))→list(valueout)

圖1 Google MapReduce執行過程

3 Hadoop 的作業調度算法

Hadoop的調度器旨在調度各種提交到Hadoop的作業，使作業有個合理的執行順序。它不僅要對作業進行調度，提高系統的吞吐量，使系統處理盡可能多的作業，提高執行效率，而且還要考慮系統的利用率，不使系統空閑下來，另外還得兼顧用戶的感受提高作業響應時間。

早期Hadoop 的作業調度算法按照作業提交順序來運行，使用的就是FIFO 調度算法。后來FaceBook 和Yahoo分別提設計了公平調度算法，計算能力調度算法。

公平調度算法目的是能夠滿足多用戶多類型作業的并行運行[8]。該算法的設計思想是讓所有的作業隨著時間的推移，都能平均獲取等同的共享資源。算法以池的形式管理作業，默認情況下每個用戶都有一個獨立的作業池，用戶能獲得相等份額的資源而不管用戶提交了多少作業。在公平調度算法的具體實現中，為了選擇合適的作業執行，公平調度算法還定義了作業赤字作為選擇依據，即一個作業在理想情況下應該獲得的計算資源量與它實際獲得的計算資源了之間的差距。公平調度算法會周期性地觀察各個作業中有多少任務已經在上一時間段內執行，并將該結果與它應得的資源份額作對比更新該作業的赤字值。一旦有空閑的TaskTracker 出現，它會被首先分配給當前具有最高赤字的作業使用。

計算能力調度算法[9]中可以定義多個隊列，當用戶提交了一個作業，該作業會被放到一個隊列中。各個隊列根據配置文件分配到相應數量的TaskTracker 資源用于處理Map 操作和Reduce 操作。當系統中出現了空閑的TaskTracker，算法會首先選擇一個具有最多空閑空間的隊列。當一個隊列被選中時，調度器就會根據該隊列中作業的優先級和提交時間進行選擇合適的作業。計算能力調度算法雖然吸取了公平算法的不足，根據作業的性能分配資源，但這種分配策略過于簡單，容易陷入局部最優，同時該算法配置起來也是比較繁瑣的，而且對服務水平協議也不能很好的支持。

在Hadoop 中作業調度器是一個可以插拔的模塊，因此針對Hadoop 現存調度器存在的問題，以及根據自己的實際應用要求修改或者重新設計調度器。國內外有很多專家學者對Hadoop 的作業調度算法進行了深入的研究。Torabzadeh 等提出了在兩階段裝配式流水作業調度中基于云理論的模擬退火算法，其中給出相關函數，加入相應指標，并證明了該算法使總運行時間和平均運行時間減少[10]。Hadoop 的公平調度算法中存在著調度公平性與數據本地化的沖突。Zaharia 和Borthakur等人在公平調度算法的基礎上提出了Delay Scheduler策略，延遲調度的設計目標是盡可能小地減少公平性，并提高作業的數據本地化，最終提高系統的吞吐量[11-12]。延遲調度的思想主要是針對小作業提出的，能很大程度提高小作業的吞吐量，從而提高整個系統的吞吐量。彭艦等提出了一種能夠找到總完成時間和平均完成時間最短的算法[13]。鄧自立針對Hadoop 的FIFO 調度算法的不足給出了一個基于優先級的加權輪轉調度算法（PBWRR），這種算法避免了作業的長期等待，又考慮到了作業的優先級[14]。

4 基于優先級的計算能力加權調度算法

4.1 算法的思想與設計

在原算法中當某個Tasktracker 上出現空閑slot 時，調度器依次選擇一個queue、（選中queue 中的）job、（選中job 中的）task，并將該slot 分配給該task[15]。下面介紹選擇queue、job 和task 所采用的策略。

（1）選擇queue：將所有queue 按照資源使用率由小到大排序，依次進行處理，直到找到一個合適的job。

（2）選擇job：在當前queue 中，所有作業按照作業提交時間和作業優先級進行排序，調度依次考慮每個作業，選擇符合兩個條件的job：①作業所在的用戶未達到資源使用上限；②該TaskTracker 所在的節點剩余的內存足夠該job 的task 使用。

（3）選擇task，考慮task 的locality 和資源使用情況。

計算能力調度算法應用于較多小作業時性能較差，容易陷入局部最優，本文的算法主要是在計算能力調度的基礎上優化步驟（2）中的作業選擇策略，優化后的策略更加公平，避免選擇策略陷入局部最優，從而提高了整個系統的效率和用戶滿意度。

在原算法中作業隊列是按照作業提交時間和作業優先級進行排序，然后選擇隊列頭部的作業。在本算法中首先根據作業權重對每個隊列的作業進行排序，然后將這個空閑的slot 分配給選中隊列的第一個作業的task（該作業必須滿足步驟（2）中的兩個條件）。可以看出，作業的權重是選擇的重要參考依據。

考慮到TaskTracker主動請求task的模式以及Hadoop任務調度體系非搶占模式的特點，為了使調度的任務避免長期的等待，同時各個任務調度權重又能根據實際情況進行調整，可以通過下面一系列推導得到權重的公式。

假設用V表示一個輪轉周期內處理數據的總量，一般情況下Map任務的數量遠遠超過Reduce任務的數量。因此，用系統能夠同時運行的Map 任務數量的能力來表示系統的所能處理任務數量的能力，記為taskcapacity。而且每個Map 任務所處理的數據量為輸入數據的分塊，大小即為HDFS中數據塊的大小，記為blockSize（一般設為64 MB），所以每一個輪轉周期系統所處理的數據量為：

用avgtasksize表示某job劃分的任務平均大小，jobsize表示某job的大小，tasknumber表示job的任務數，則有：

如果job[i]的權重為weight[i] 為了保證優先級的作用和有效性，某個job 的優先級與所有job 優先級之和的比值應該基本等于這個job 執行的數據量與隊列一個輪轉周期的數據量的比值，即有：

于是就可以得到：

根據實際情況對權重進行動態的調整避免算法陷入局部最優，這里的實際情況就是隊列中作業的等待時間以及作業完成度（未完成task/全部的task）。因此計算作業權重要將作業的等待時間以及作業的完成度考慮進去，由此可得：

其中，ctime表示當前時間，stime表示作業提交時間，utask表示未完成任務數，atask代表任務的總數。

權重是同一隊列中作業排列的依據，在公式（5）中taskcapacity、blockSize和avgtasksize對同一隊列的作業來說都是定值所以可以簡化權重公式為：

4.2 算法的實現

（1）數據結構設計

jobQueue 的調度隊列中每個元素jobInfo 的數據結構為：

class JobSchedulingInfo {

JobId jobId；

JobPriority priority；

long startTime；

int taskNum=0；

int finshedTaskNum=0；

double weight=0；

}

本文算法在JobSchedulingInfo 中增加了幾個參數，解析如下：

taskNum：作業的任務數量；

finshedTaskNum：作業已完成的任務數；

weight：作業的權重。

（2）權重的計算方法

權重的計算要基于作業的優先級、作業的提交時間以及作業的完成量，其中優先級對應的值如表1 所示。

表1 優先級對應的具體值

算法的偽代碼如下：

（3）算法的其他內容描述

本算法中要將JobQueue 中作業根據作業的權重進行排序，實現的關鍵部分就是增加一個新的比較器，通過比較器來對隊列排序。比較器偽代碼描述如下：

5 實驗結果及分析

實驗采用三臺虛擬機，操作系統為Linux Ubuntu 10.04.2，Java Version為1.6.0_10-rc2，Hadoop版本為0.20.2。將其中一臺作為主機Master，其余的兩臺分別作為Slave。使用最常用的WordCount（從文本數據統計單詞次數）作業來測試調度的改進。算法部分參數配置如表2 所示。

表2 算法參數配置

實驗1將八個作業傳到集群上運行，作業的詳細信息如表3。

FIFO 調度器中得到的運行時間如圖2 所示。

將Hadoop0.20.2 計算能力調度器插入到集群中。同樣將上面的八個作業傳到集群上運行得到的運行時間如圖3 所示。

將本文的調度算法插入到集群中，得到的運行時間如圖4 所示。

表3 作業詳細信息

圖2 FIFO 調度器運行結果圖

圖3 計算能力調度器運行結果圖

圖4 本文算法運行結果圖

對比實驗結果可以看出三個調度算法的執行順序均不相同，時長不一。FIFO 算法是按照作業的優先級和進入隊列先后時間順序執行的，沒有考慮作業的長短以及不同作業的需求，執行總時間最長，計算能力調度算法和本文算法執行順序不同是因為算法的選擇作業策略不同。對比圖3 和圖4，可以看出圖4 中的部分小作業執行時間提前，執行順序并沒有嚴格按照作業的優先級進行執行；還可以看出雖然部分作業的執行時間延長了但是等待作業執行時間提前了。這是因為在作業執行的尾段，作業的完成度較高而等待作業的完成度較低，調度器動態調整了作業的權重，等待作業的權重這時候有可能大于運行的作業，這樣就減少了等待作業的響應時間。同時可以看出，優化后的作業選擇策略使得總時間比計算能力調度算法減少了兩分多鐘。

實驗2測試更一般的情況，提交八個作業到集群上運行。作業的優先級和大小是不盡同的，集群分別使用FIFO、本文算法和計算能力調度算法，完成時間對比如圖5 所示。

圖5 作業在不同調度算法下的完成時間對比

從圖5可以看出雖然本文算法在完成前幾個作業和原算法的用時上并沒有太大的優勢，但是隨著作業數的增加本文算法逐漸顯現出優勢。因此可以預測，在大的多用戶多任務的集群上本文算法的效率肯定將超過原算法。

實驗3將三個不盡相同的作業組上傳到集群上，并用本文算法和計算能力調度算法進行測試。每組八個作業，作業的優先級隨機分配，J-a 是小作業占多數的作業組，J-b 是大作業和小作業的數量相當的作業組，J-c 是大作業占多數的作業組，得到結果如圖6 所示。

圖6 三組作業在兩種調度算法下的時間比較

從圖6可以看出來不同類型的作業組在本文算法下運行的時間均比計算能力調度算法要少。在作業組中有較多小作業的時候本文算法可以減少總運行時間，當作業組中大作業占據絕大多數時，本文算法就退化為與計算能力調度算法類似。結合實驗對比圖可以得出本文算法減少了部分作業等待時間，優化了作業選擇策略，使得作業組的總運行時間減少，提高了用戶的滿意度。

6 結束語

作業調度器關系到Hadoop 平臺整體的性能和系統的資源使用效率，以及云計算服務的能力。本文提出了一種基于優先級的計算能力加權調度算法，當系統出現空閑的slot 時，作業的選擇策略不僅僅考慮作業的優先級而是同時將作業的等待時間和作業完成度動態的考慮進去。這種策略避免了作業調度器陷入局部最優，同時能夠減少某些相對較高優先級小作業的等待時間，在一定程度上優化了作業選擇策略，減少了總運行時間，提高了系統的使用效率。

[1] 劉鵬.云計算[M].2 版.北京：電子工業出版社，2011.

[2] Dean J，Ghemawat S.MapReduce：Simplified data processing on large clusters[J].Communications of the ACM，2008，51（1）：107-113.

[3] Ralf L.Google’s MapReduce programming model-revisited[J].Science of Computer Programming，2008，70（1）：1-30.

[4] Wbite T.Hadoop 權威指南[M].曾大聃，周傲英，譯.北京：清華大學出版社，2010.

[5] Apache Hadoop[EB/OL].[2013-09-01].http：//hadoop.apache.org/.

[6] 趙春燕.云環境下作業調度算法研究與實現[D].北京：北京交通大學，2009：20-58.

[7] 羅軍舟，金嘉暉，送愛波，等.云計算體系架構與關鍵技術[J].通信學報，2011，32（7）：3-21.

[8] Fair Scheduler[EB/OL].[2013-09-01].http：//hadoop.apache.org/common/docs/r0.20.2/fair_scheduler.html.

[9] CapacityScheduler[EB/OL].[2013-09-01].http：//hadoop.Apache.org/common/docs/r0.20.2/capacity_scheduler.html.

[10] Torabzadeh E，Zandieh M.Cloud theory-based simulated annealing approach for scheduling in the two-stage assembly flowshop[J].Advances in Engineering Software，2010，41（10）：1243-1298.

[11] Zaharia M，Borthakur D，Sen Sarma J，et al.Delay scheduling：A simple technique for achieving locality and fairness in cluster scheduling[C]//Proceedings of EuroSys’10.New York，NY，USA：ACM，2010：265-278.

[12] Matei Z，Dhruba B，Joydeep S，et al.Delay scheduling：A simple technique for achieving locality and fairness in cluster scheduling[C]//Proc of the 5th European Conference on Computer Systems，2010：265-278.

[13] 李建鋒，彭艦.云計算環境下基于改進遺傳算法的任務調度算法[J].計算機應用，2011，31（1）：184-186.

[14] 鄧自立.云計算中的網絡拓撲設計和Hadoop平臺研究[D].合肥：中國科學技術大學，2009.

[15] 陳艷金.MapReduce 模型在Hadoop 平臺下實現作業調度算法的研究和改進[D].廣州：華南理工大學，2011.