動態標簽匹配DLMS調度器設計與實現

毛韋　竹翠

摘 要：針對Hadoop集群節點性能差異大、資源分配隨機、執行時間過長的問題，提出一種將節點性能標簽（簡稱節點標簽）和作業類別標簽（簡稱作業標簽）進行動態匹配的調度器。節點初始分類并賦予原始節點標簽，節點檢測自身性能指標生成動態節點標簽，作業根據部分運行信息進行分類并生成作業標簽，資源調度器將節點資源分配給對應標簽的作業。實驗結果表明，相對于YARN中自帶的調度器，其在作業執行時間上有很大縮短。

關鍵詞：Hadoop；資源調度器；動態匹配；動態標簽

DOI：10.11907/rjdk.171392

中圖分類號：TP319 文獻標識碼：A 文章編號：1672-7800（2017）009-0095-05

Abstract：In order to solve the problem of big performance difference， random resource allocation， and long time execution in Hadoop cluster nodes，this paper propose a scheduler which allocates the Node Performance Label（NPL） and Job Category Label（JCL） dynamically.The node makes the classification of initialization and is assigned original node label. The node detects its own performance metrics to generate dynamic node labels. The job is classified based on part of the run information to generate the job label. The resource scheduler assigns the node resource to the corresponding label job. The experimental results shows that the scheduler has a certain degree of shorten in the time of job execution compared with one that comes with YARN.

Key Words：hadoop； scheduler； dynamic matching； dynamic label

0 引言

早期Hadoop版本由于將資源調度管理和MapReduce框架整合在一個模塊中，導致代碼的解耦性較差，不能較好地進行擴展，不支持多種框架。Hadoop開源社區設計實現了一種全新架構的新一代Hadoop系統，該系統為Hadoop2.0版本，將資源調度抽取出來構建了一個新的資源調度框架，即新一代的Hadoop系統YARN。眾所周知，在某一確定的環境下，合適的調度算法能夠在滿足用戶作業請求的同時，有效提升Hadoop作業平臺的整體性能和系統資源利用率。在YARN中默認自帶3種調度器：先入先出（fifo）、公平調度器（Fair Scheduler）和計算能力調度器（Capacity Scheduler）。Hadoop默認采用fifo調度器，該算法采用先進先出的調度策略，簡單易實現，但是不利于短作業的執行，不支持共享集群和多用戶管理；由Facebook提出的公平調度算法考慮了不同用戶與作業資源配置需求的差異，支持用戶公平共享集群的資源，但是作業資源的配置策略不夠靈活，容易造成資源浪費，并且不支持作業搶占；雅虎提出的計算能力調度算法支持多用戶共享多隊列，計算能力靈活，但是不支持作業搶占，易陷入局部最優[1-2]。

然而在企業生產中，隨著企業數據量的加大，每年集群都會加入一些新節點，但是集群節點的性能差異很顯著，這種異構集群在企業生產環境中很普遍。設想如果將一個計算量很大的機器學習任務分配在CPU計算能力很差的機器節點上，顯然會影響作業的整體執行時間。Hadoop自帶的3種資源調度器并沒有很好地解決該問題。本文提出一種節點性能和作業類別標簽動態匹配的資源調度算法（DLMS），將CPU性能較好的機器貼上CPU標簽，將磁盤IO性能較好的機器貼上IO標簽或者是兩者都一般的普通標簽。作業根據分類可以貼上CPU標簽、IO標簽任務或者普通標簽，然后進入不同的標簽隊列，調度器盡可能將相應標簽節點的資源分配給相應的標簽作業，從而減少作業運行時間，提高系統資源利用率，提高系統整體效率。

1 相關研究

Hadoop集群異構調度相關研究已較多，Quan Chen等[3-4]提出動態MapReduce調度算法SAMR，能夠動態計算任務的進展，自動適應不斷變化的環境。但是該算法并沒有考慮到數據的本地化問題，同時也缺少一種動態參數調優機制。Aysan Rasooli等[5]提出異構環境中的動態調度算法，通過估計作業到達率和平均作業執行時間決定調度策略。該文討論了不同作業的資源需求，把任務分配到不同的隊列中，設計了一個MapReduce負載預測機制來檢測MapReduce過程的負載類型，并對其分類后放入相應的隊列中。針對需要頻繁IO的作業類型，HUI Kang等[6]提出MapReduce組調度算法，以降低上下文切換次數和解決IO阻塞問題。寧菲菲、鄭均輝等[7]探討了三隊列作業調度算法，提出節點分類的概念，基于Hadoop1.0框架，沒有考慮到節點運行作業時性能的變化等情況。以上這些算法都是針對異構集群提出的優化調度算法，都沒有將節點和作業進行動態匹配的調度，缺少一種更加通用簡單高效的解決方法，增加了用戶的學習使用成本。endprint

2 調度算法

2.1 整體架構

本文提出的調度框架將集群節點進行初始分類并賦予相應的標簽。NodeManager發送心跳前作自我檢測并對原始標簽進行動態調整，并使用機器學習分類算法對作業進行分類并賦予相應的標簽，根據用戶設定的作業優先級、作業等待時間等屬性動態實現作業排序，并將相應標簽的資源分配給相應標簽隊列中的作業。YARN調度作業整體框架流程如圖1所示。

各步驟解釋如下：①用戶向YARN提交應用程序，其中包括用戶程序、啟動ApplicationMaster命令等；②ResourceManager為該應用程序分配第一個Container，并與對應的NodeManager通信，要求它啟動應用程序的ApplicationMaster；③ApplicationMaster向ResourceManager注冊后，為各任務申請資源，并監控它們的運行狀態，直到運行結束；④NodeManager發送心跳前作自我檢測生成動態的節點標簽，并向ResourceManager匯報資源；⑤作業分類進入不同的標簽隊列中，進行優先級排序等待分配資源；⑥ApplicationMaster通過RPC協議向ResourceManager申請和領取資源；⑦根據NodeManager匯報的節點標簽和資源，調度器將此節點的資源分配給對應標簽隊列的作業；⑧ApplicationMaster申請到資源后，便與對應的NodeManager通信，要求它啟動任務；⑨NodeManager為任務設置好運行環境（環境變量、JAR包、二進制程序等）后，將任務啟動命令寫到腳本中，并通過運行該腳本啟動任務；⑩各任務通過某個RPC協議向ApplicationMaster匯報自己的狀態和進度，可以在任務失敗時重新啟動任務；應用程序運行完成后，ApplicationMaster向ResourceManager注銷并關閉自己。

該調度器主要模塊包括：①集群節點原始分類及其動態分類標簽；②Map執行信息的獲取與回傳；③多優先級隊列；④作業分類；⑤數據本地性；⑥調度算法。

2.1.1 集群節點原始分類及其動態類別標簽

根據節點運行單個任務的時間與集群中所有節點運行時間平均值的大小關系將節點大致分成CPU型節點、磁盤IO型節點、普通型節點。分類方法如下：

（1）設節點集為N={Ni|i∈[1，n]}，n為節點數量。

（2）在每臺節點上都執行一個相同任務量的3種類型的作業并記錄作業執行時間，Tcpu（i）表示在第Ni個節點上執行CPU作業的花費時間，Tio（i）表示在第Ni個節點上執行IO作業的花費時間，Tcom（i）表示在第Ni個節點上執行普通作業的花費時間。

（3）計算每種作業的集群平均時間，計算公式如下：Avgj=∑n1Tj（i）n i∈[1，n]，j∈{cpu，io，com}

（1） 計算出每個節點在此種作業下與平均時間的時間差，如果Tcpu（i）Avgcpu，可以為此節點貼上普通型原始標簽，通過比較后很可能每臺節點上的標簽會有多個，選擇節省時間最多的標簽為此節點的最后標簽。

在集群節點運行過程中，如果一個節點的負載過大，即使該節點原始標簽是CPU型標簽，但在此環境下CPU性能優勢已經不能體現，因此采取動態標簽的方法，在nodeManager向resourceManager發送心跳時動態檢測節點機器的CPU和IO使用率，如果超過了一定的閾值，將此節點標簽貼上普通標簽，由此實現動態標簽。

2.1.2 Map信息回傳與獲取

Hadoop作業通常分成Map階段和Reduce階段，通常大作業的Map數量有上百個甚至更多，一個作業的時間主要消耗在Map階段計算上，但是每個Map又是完全相同的執行內容，因而會收集作業運行的第一個Map進程的運行信息，這些信息在Nodemanage向ResourceManager發送心跳時傳遞到調度器中，由調度器進行作業分類。當生產環境中同一種類型作業每天都會運行，即用戶已經知道作業所屬標簽，可以在命令行或者代碼中為作業設置作業類型標簽，在調度時調度器會進行檢查，如果用戶已經對作業貼上標簽，就省去分類環節，直接進行調度。設Map的運行信息為，M={Min，Mout，Rate，Acpu，Mcpu，Zcpu，Mrate}，它包括以下需要收集的信息：Min表示Map輸入數據量，Mout表示Map輸出數據量，Rate表示輸入數據量/輸出數據量，Acpu表示CPU平均使用率，Mcpu表示CPU中位數，Zcpu表示CPU使用率超過90%的平均數，MRate表示內存使用量，這些數據將成為以后該作業分類的特征屬性。在實驗過程中發現，單純計算CPU的平均時間不能很好反映作業特征，CPU型作業的CPU使用率大于90%的次數較多，其類型作業的CPU使用率大于90%的次數相對較少。因此，將此信息也加入到Map回傳的信息中。

2.1.3 多級優先級隊列

在調度器中新建5個隊列：原始隊列、等待優先級隊列、CPU優先級隊列、IO優先級隊列、普通優先級隊列。用戶提交的作業首先進入原始隊列中，先運行作業部分Map并收集這部分Map運行信息，作業進入等待優先級隊列中根據作業的分類類別標簽進入到對應標簽的隊列中。在隊列優先級方面采取用戶自定義雙層權重的設計方法，設作業的大小屬性所占權重為worthNum，該屬性可以分成3個等級Ni∈{long，mid，short}。作業的擁有者屬性所占權重為worthUser，該屬性可以分成兩個等級Ui∈{root，others}；作業的緊急程度所占權重為worthEmogence，該屬性可分成3個等級Ei∈{highPrority，midPrority，lowPrority}；作業的等待時間所占權重為worthWait，等待的計算公式為Jwait=nowTime-submitTime，賦予相應權重，最后計算出每個任務的優先級數，然后在相應的隊列中進行排序。上述5種任務屬性權重相加和為100%，具體公式如下：worthNum+worthUser+endprint

worthEmogence+worthWait=1

（2） 最后權重計算公式：finalWorth=worthNum*Ni+worthUser*Ui+

worthEmogence*Ei+worthWait*Jwait

（3）2.1.4 作業分類

在作業分類上，選擇簡單、使用比較普遍而且分類效果較好的樸素貝葉斯分類器進行分類[8]。如果用戶在命令行和任務代碼中已經添加作業類型，則此步驟會省掉，直接進入相應的隊列中等待分配資源。樸素貝葉斯分類器算法具體分類步驟如下：

（1）分別計算一個作業在某些條件下是CPU、IO還是普通型作業下的條件概率：P（job=cpuV1，V2…Vn）

P（job=ioV1，V2…Vn）

P（job=comV1，V2…Vn） 其中，job表示作業類別標簽，為作業的屬性特征。

（2）根據貝葉斯公式P（BA）=P（AB）/P（A）得：P（job=cpuV1，V2…Vn）=

P（V1，V2…Vnjob=cpu）P（job=cpu）P（V1，V2…Vn）

（4） 假設V之間相對獨立，根據獨立假設：P（V1，V2…Vn|job=cpu）=∏n1P（Vi|job=cpu）（5） （3）實際計算中P（V1，V2…Vn）與作業無關可忽略不計，因此最終可得：P（job=cpuV1，V2…Vn）=

P（job=cpu）∏n1P（Vijob=cpu）

（6） 同理有：P（job=ioV1，V2…Vn）=

P（job=io）∏n1P（Vijob=io）

（7）

P（job=comV1，V2…Vn）=

P（job=com）∏n1P（Vijob=com）

（8） 作業是CPU型作業、IO型作業還是普通型作業取決于哪個概率值更大[9]。

2.1.5 數據本地性

Hadoop中遵循一個原則是“移動計算比移動數據更好”[9]，移動到放置數據的計算節點要比將數據移動到一個計算節點更加節省成本，性能更好。關于數據本地性，本文采取了延時降級調度策略。該策略具體思想如下：

為每個作業增加一個延時時間屬性，設Ti為第i個作業當前的延時時間，i∈[1，n]，n為集群的節點數目，初始化為0，Tlocal表示本地節點延時時間閾值，Track表示機架節點延時時間閾值。當調度器分配資源給作業時，如果作業的執行節點和數據輸入節點不在一個節點上，此時自增1，表示該作業有一次延時調度，則可以將此資源分配給其它合適的作業，直到當Ti>Tlocal時，作業的本地性就會降低為機架本地性，此時只要是本機架內的節點都可以將資源分配給該作業；當Ti>Track時，作業的本地性降低為隨機節點。其中Tlocal和Track都采用配置文件的方式由用戶根據集群情況自行配置。采用延時的調度策略可以保證在一定的延時時間內獲得較好的本地性。

2.1.6 DLMS調度算法

DLMS調度算法的基本思想是預先分配部分作業執行，根據作業回傳信息對作業進行分類，然后將節點標簽的資源分配給相應隊列中的任務，基本流程：

步驟1：當節點通過心跳向資源管理匯報資源時，如果原始隊列不為空，則遍歷原始隊列中的作業，將已經在命令行或者程序中指定了作業類型標簽的作業分配到相應的標簽優先級隊列中，原始隊列移除此作業。

步驟2：如果原始隊列不為空，則將此節點上的資源分配給原始隊列，作業進入等待隊列中等待下次分配資源，原始隊列移除此作業，此輪分配結束。

步驟3：如果等待優先級隊列不為空，則對等待優先級隊列中的作業進行分類進入相應的標簽優先級隊列。

步驟4：如果節點性能標簽相應的作業類別隊列不為空，則將此節點的資源分配給此隊列，此輪分配結束。

步驟5：設置查看資源訪問次數變量，如果超過集群的數量，則將節點的資源按CPU、IO、普通、等待優先級順序將資源分配給相應的隊列，此輪調度結束。此步驟可以防止出現如下情況：CPU隊列作業過多，導致CPU型節點資源已經耗盡，而其它標簽的節點還有資源，但是作業無法分配資源。

算法流程如圖2所示。

DMLS調度算法流程偽代碼如下：

算法1：

When a heartbeat is received from Node i to RM

If 初始隊列不為空

遍歷原始隊列中的作業將已經貼上標簽的作業放入相應的標簽隊列中

if applicats中有app不為空

為app分配ApplicationMaster

endif

endif

if 等待隊列不為空

分類得到app類型

根據app類型進入不同的隊列中

在等待隊列中將app移除

Endif

根據節點標簽的類型將節點資源分配給CPU、IO、COM優先級隊列之一

if（調度次數變量 >= 總節點數目）

依次將資源分配給CPU、IO、COM隊列

調度次數變量=0；

endif

算法2：

assignContiners（applications，node）

for（app ： applications）

獲取app用戶、大小、緊急程度、等待時間對應的權重

計算app最后的總權重

endforendprint

根據計算的權重值為app排序

for（app ： applications）

fifo為app分配資源

endfor

3 實驗驗證

本文通過實驗驗證DLMS調度器的實際效果。實驗環境為5臺PC機搭建而成的Hadoop完全分布式集群，集群的節點機器配置統一為操作系統Ubuntu-12.04.1，JDK1.6，Hadoop2.5.1，內存2G，硬盤50G。其中NameNode的CPU核數為2，dataNode1的CPU核數為2，dataNode2的CPU核數為4，dataNode3的CPU核數為2，dataNode4的CPU核數為4。

3.1 節點原始標簽分類

首先準備數據量為128M的wordCount（IO型），kmeans（CPU型）作業各一個，分別在4臺節點上面運行6次，記錄作業運行時間。

表1中，s表示時間單位秒，avg表示該節點運行相應標簽任務的平均時間，allAvg表示所有節點運行相應標簽任務的總平均時間，rate的計算公式如下：rate=avg-allAvgallAvg×100%

（9） 其中，負號表示平均時間相對于總平均時間的減少，正號表示平均時間相對于總平均時間的增加。

由圖2可以看出，DataNode1運行兩個任務都是節省時間的。采取節省最多的CPU作業作為機器的原始標簽，DataNode4為IO標簽，DataNode2、DataNode3為普通機器。

3.2 實驗及結果分析

使用可以明顯區分作業類型的幾種作業，WordCount在Map階段需要大量地讀取數據和寫入中間數據，Map階段和Reduce階段基本上沒有算數計算，因此將這種作業定性為IO型作業，Kmeans在Map階段和Reduce階段都需要大量計算點和點之間的距離，并沒有太多中間數據的寫入，因此將這種作業定性為CPU型作業，TopK在Reduce階段沒有大量的數據寫入磁盤，也沒有大量的計算，只涉及簡單的比較，因此可認為是普通型任務。

通過兩組實驗進行驗證，第一組實驗設置調度器為fifo，在500M、1G和1.5G的數據量下分別運行WordCount、Kmeans、Topk作業各3次，記錄每個作業3次的平均時間作為最終時間，切換調度器為Capacity和DLMS調度器做同樣的實驗操作，實驗中記錄DLMS調度器下每種作業的Container在集群的分布，Container表示集群資源的劃分單位，記錄了作業分片在集群中運行的分布情況，YARN中每個Map和Reduce進程都是以一個Container來表示。Container在集群中每個節點分布比例表明節點執行作業任務量的比例。

圖3的橫坐標是作業數據量，縱坐標是WordCount、Kmeans、Topk這3種作業共同運行的總時間。在數據量增大的情況下，DLMS調度器相比于其它調度器節省約10%～20%的時間。

DLMS會將相應節點標簽的資源分配給相應標簽的作業。作業的Map和Reduce以Container的形式在節點上運行，圖4—圖6是在DMLS調度器下不同數據量作業的Container數量。原始分類Node1是CPU型的標簽節點，Node2和Node3是普通標簽節點，Node4是IO標簽節點。WordCount是IO型作業，Topk是普通型作業，Kmeans是CPU型作業。從圖中可以看出，Container的分布規律為WordCount作業在Node4上分配的Container較多，Tokp在普通節點Node2和Node3上分布較多，Kmeans作業在Node1節點上分布較多。以上不同作業的Container在集群節點上的分布表明，DLMS調度器提高了相應節點標簽的資源分配給對應標簽作業的概率。

第二組實驗中，準備了5個作業，分別是128M和500M數據量的WordCount作業，128M和500M數據量的Kmeans作業，500M的Topk作業組成一個作業組，5個作業同時提交運行。在不同調度器的集群中模擬連續作業執行情況，記錄作業組執行完的總時間。作業組在不同的調度器下運行3次，記錄作業組運行完的總時間。

具體結果如圖7所示，可以看出本文提出的DLMS調度器相比于Hadoop自帶調度器執行相同作業組節省的時間較明顯，本文提出的DMLS調度器相比Hadoop自帶的Fifo調度器節省了大約20%的時間，比Capacity調度器節省了大約10%的運行時間。

4 結語

本文提出的DLMS調度器用兩個不同的實驗驗證該調度器的效率，從實驗數據上可以看出，無論是執行不同數據量的單個作業，還是執行由不同數據量組成的多個不同作業組成的作業組，相對于Hadoop自帶的其它調度器，其在執行時間上都大大減少，集群資源利用率得以提高。后續研究的重點是在本文提出的調度器中考慮Reduce的數據本地性。

參考文獻：

[1] 朱潔，趙紅，李雯睿.基于Hadoop的三隊列作業調度算法[J].計算機應用，2014，34（11）：3227-3230，3240.

[2] 鄧傳華，范通讓，高峰.Hadoop下基于統計最優的資源調度算法[J].計算機應用研究，2013，30（2）：417-419.

[3] CHEN Q， ZHANG D， GUO M， et al. SAMR： a self-adaptive mapreduce scheduling algorithm in heterogeneous environment[C]. IEEE， International Conference on Computer and Information Technology，2010：2736-2743.

[4] 陳全，鄧倩妮.異構環境下的動態的Map-Reduce調度[J].計算機工程與科學，2009，31（S1）：168-171，175.

[5] RASOOLI A， DOWN D G. An adaptive scheduling algorithm for dynamic heterogeneous Hadoop systems[C]. Conference of the Center for Advanced Studies on Collaborative Research，2011：30-44.

[6] KANG H， CHEN Y， WONG J L， et al. Enhancement of Xen′s scheduler for MapReduce workloads[C]. ACM International Symposium on High PERFORMANCE Distributed Computing，2011：251-262.

[7] 寧菲菲，鄭均輝.基于Hadoop平臺的三隊列作業調度算法[J].微型電腦應用，2015， 31（8）：19-21.

[8] 趙文濤，孟令軍，趙好好，等.樸素貝葉斯算法的改進與應用[J].測控技術，2016，35（2）：143-147.

[9] 楊洋.基于資源狀況的延時等待公平調度算法的研究[D].沈陽：東北大學，2014.

（責任編輯：孫 娟）endprint