基于獎懲共存收益模式的大數據作業調度器

胡 靜

(山西農業大學 軟件學院，山西 晉中 030801)

0 引 言

目前大數據計算平臺中已經存在的調度器如默認的FIFO調度器、公平調度器和計算能力調度器[1]等其它算法[2,3]調度作業時只考慮到了平臺資源利用率但沒有考慮作業截止時間和服務商收益問題。針對作業截止時間要求，部分研究學者[4-9]以作業截止期限為約束提出一種相應的作業調度算法以提高作業的執行效率與平臺資源利用率，但仍沒考慮服務商的收益問題。文獻[10-14]提出相關調度算法目的是使服務商的利潤最大化。文獻[15-17]均提出了新的調度方法，保證服務商的最大收益、服務的截止時間和平臺的資源利用率。在服務商指定時間內完成每個作業可獲得的收益模式和現有作業調度器下，服務商沒有考慮到作業截止時間的準確性和作業執行的迫切需求，并且也沒有考慮作業調度結果對平臺資源利用率的影響。

基于上述問題，本文提出了一種服務商收益模式—獎懲共存收益模式(RP Model)。該收益模式考慮用戶對服務商的獎勵政策，即當服務商在作業截止時間之前的一定時間范圍內完成作業時對服務商給予相應的獎勵值。此收益模式針對的是不確定作業截止時間是否準確的用戶，并且該類用戶對于作業執行完成有迫切需求。服務商為了滿足用戶較短的作業完成時間需求，盡量縮短每個作業的完成時間。當作業實際完成時間比用戶提供的截止時間短時，服務商將該作業的實際完成時間作為較準確的截止時間反饋給用戶，使得用戶以后提交相同的作業時就可以提供較準確的截止時間，并且推進了用戶處理作業群的整體行為。獎懲共存收益模式的提出實現了服務商與用戶的利益共贏。

本文以獎懲共存收益模式為目標，提出了一種高效的MapReduce作業調度器——最大收益完成時間資源利用率調度器(MPCRS)，以盡量縮短每個作業的完成時間，提高平臺資源利用率，使服務商獲得最大收益。

1 MPCRS調度器設計

本文是在Hadoop2.x的Yarn資源管理系統的基礎上提出的MPCRS。Yarn作為統一資源管理系統，核心組件為全局資源管理器(resource manager，RM)，負責整個平臺的資源管理和分配。它主要由兩個組件構成：作業調度器(Scheduler)和應用程序管理器ApplicationsManager(ASM)。調度器能夠根據容量、隊列等限制條件，將平臺中的資源按照作業資源需求分配給各個將要運行的作業。由于調度器是一個可插拔的組件，故本文將以滿足服務商最大收益、平臺最大資源利用以及作業最短完成時間為目標設計一種MPCRS作業調度器。圖1展示了在Yarn平臺中MPCRS的作用以及構成。

圖1 MPCRS的作用與構成

如圖1所示，該平臺允許多用戶提交多個帶有SLA約束的作業。RM中主要包含MPCRS和ASM兩種組件，其中MPCRS是由RP Model、基于任務執行時間的確定輪數算法(TRN)以及基于最大輪數的作業調度算法(MRNS)組成。RP Model將每個作業在不同時間段內可獲得的相應收益值作為TRN算法與MRNS算法的輸入值。為了盡量縮短每個作業的完成時間，TRN算法以RP Model下可獲得的收益值為標準，根據各個作業的Map和Reduce任務執行時間，確定出作業在不同獎懲階段的Map和Reduce最大輪數組合以及最大標準時間。MRNS算法得到最大輪數組合方案和最大標準時間值后，結合平臺中現有資源數量，考慮平臺資源利用率，制定出對于當前所有作業的調度策略(TS)。ASM將會根據TS策略啟動相應作業的(application master，AM)，AM向RM申請所需的資源，RM為其任務分配在各個(node manager，NM)上的Container資源，使得該作業得以開始運行。AM將作業分為Map和Reduce任務，由于MPCRS產生的TS策略是基于任務級別的，故為作業分配資源即為Map或Reduce任務分配資源。

2 MPCRS調度器實現

本文提出的MPCRS調度器是生成相應的作業調度策略，而具體任務分配方法仍然遵循MapReduce的動態分配原則，因此不會對平臺中負載均衡等其它性能特性造成影響。在滿足平臺資源利用率最大的條件下，雖盡量縮短每個作業的完成時間，但不可避免會造成個別作業超出截止期限，無法按時完成。出現這種情況時需根據服務商收益最大化的目標對作業的收益與賠付代價進行衡量評估，經過整體收益權衡后選擇放棄一些收益較少或賠付較少的作業。

2.1 獎懲共存收益模式-RP Model

在計算平臺中認為每個節點的處理能力都大致相同，由于平臺中的資源是統一進行管理和分配，故可動態分配的Container資源用R表示。用戶提交的作業j已知如下信息：j的截止時間j_deadline；在j_deadline之前j被完成，服務商可獲得的收益j_profit；如果j的實際完成時間j_completion的a倍比j_deadline短，用戶按照SLA中規定的獎勵比率α，支付除收益以外的獎勵值j_profit×α×a；j的實際完成時間j_completion超過j_deadline時，按照服務商與用戶簽訂的SLA賠付比率β進行賠償，即當j_deadlineb×j_deadline， 則需要賠付給用戶原收益的γ倍，即j_profit×γ；j具有的Map數量j_Nm和Reduce數量j_Nr；j中每一輪Map任務的執行時間j_Tm和每一輪Reduce任務的執行時間j_Tr。為了提高作業的執行效率和平臺資源利用率，在平臺資源充足的條件下，要盡量使同一個作業的多個任務并行執行，而每次并行執行任務的時間稱為一輪任務的執行時間。

根據以上關于作業的收益與賠付信息，結合平臺中可分配的資源數量，RP Model如下

(1)

(2)

式(1)表示完成所有作業后可獲得的收益和，其中每個作業的實際收益是由該作業在截止時間之前完成的收益j_profit與額外的獎勵或者懲罰f(j_profit) 組成。式(2)確保了將要運行的任務每次并行請求資源時資源數不超過平臺中可動態分配的資源總和。獎懲共存函數f(j_profit) 表示如下

(3)

其中，α是在作業實際完成時間比規定的截止期限短a倍時可獲得的獎勵比率，β是作業實際完成時間比規定的截止期限長但沒有超過截止時間的b倍時作業的賠付比率，γ是作業實際完成時間超過規定截止時間b倍時所要賠付的收益倍數。在考慮到盡量使每個作業都在截止期限之前完成，故設置賠付率值要比獎勵率值大，即α≤β≤1； 而作為長時間超出截止期限的懲罰設置γ≥1。 由于不能簡單的根據作業實際完成時間和截止時間值就規定用戶付出獎勵或服務商付出賠付，而應該是在兩者差值超過一定范圍后用戶才需要支付額外的獎勵值或服務商支付額外的賠付值，故設a,b>1。

2.2 基于任務執行時間的確定輪數算法——TRN

TRN算法根據每個作業的Map和Reduce任務執行時間，以RP Model中可獲得的收益值為標準，確定每個作業在不同獎懲階段的Map和Reduce最大輪數組合以及最大標準時間。在上節中介紹了用戶提交作業時已知的信息，其中包括作業j中每一輪Map任務的執行時間j_Tm和每一輪Reduce任務的執行時間j_Tr，并且j具有的Map數量j_Nm和Reduce數量j_Nr。

TRN算法的結果是要得到每個作業在不同獎懲階段時的任務最大輪數組合方案與最大標準時間。其中有關確定任務輪數的因素具體包括

(4)

(5)

(6)

在全部動態可分配資源只分配給一個作業的情況下，RN_m是作業j的Map任務的最少執行輪數；RN_r則是Reduce任務的最少執行輪數；式(6)表示作業j的最短執行時間tleast是由任務的最少輪數與每輪任務的執行時間組成。

算法1展示了如何在各獎懲階段獲得任務最大輪數組合方案以及最大標準時間。首先，算法根據作業的已知信息計算了每個作業的Map和Reduce任務的最少執行輪數、最短執行時間tleast以及在每個獎懲階段的最大標準時間tstandard(line(2)-line(5))； 其次，在每個最大標準時間時，比較了作業最短執行時間與最大標準時間的大小，根據獎懲共存函數f(j_profit) 的模式要求，最大標準時間的最小值一定是大于等于最短執行時間，故在每個獎懲階段都可以得到一種輪數組合方案A，最大限度地以空間換時間方案，即 (RN_m,RN_r,tstandard-tleast)(line(8))； 最后，根據剩余時間值的大小，找到能夠滿足執行一輪Map和一輪Reduce任務時間的最大變量值k，i，從而得到方案集 (planjm_t∪planjr_t) ——最大限度的以時間換空間方案。由于作業完成時間超過截止時間后服務商就要對用戶進行賠償，故要盡量保證每個作業的完成時間能夠達到最短，而任務的執行輪數會直接影響到作業的完成時間。同一獎懲階段中，在輪數不超過任務數量的前提下，即 (RN_m+k)

算法1：TRN algorithm

Input：j_Tm: the processing time of the round map task

j_Tr: the processing time of the round reduce task

j_Nm: the number of map tasks for a job

j_Nr: the number of reduce tasks for a job

f(j_profit): the reward or the punishment of a job

N: the number of jobs

R: dynamically allocate the number of resources

Output：Planj_t: the maximum round numbers plans

tstandard: the maximum standard time for a job

(1)forj=0;j

(2)RN_m←the minimum round numbers of map tasks

(3)RN_r←the minimum round numbers of reduce tasks

(4)tleast←the minimum processing time of a jobj

(5)tstandard∈T←the maximum standard time of every PR stage

(6)whiletstandarddo

(7)iftstandard≥tleastthen

(8) getting the round numbers plan A (RN_m,RN_r), the remaining time (tstandard-tleast)

(9)k,i≥1andk,iareintegersandk,iarethemaximumvaluesthatmeettherequirements

(10)if(tstandard-tleast)≥j_Tmkthen

(11) getting plan B (RN_m+k,RN_r), the remaining time (tstandard-tleast-j_Tm×k)

(12)if(tstandard-tleast-j_Tm×k)≥j_Tr×ithen

(13) getting plan C (RN_m+k,RN_r+i), the remaining time (tstandard-tleast-j_Tm×k-j_Tr×i)

(14)elsethen

(15)planjm_t={A,B}

(16)endif

(17) planjm_t={A, B, C}

(18)endif

(19)if(tstandard-tleast)≥j_Tr×ithen

(20) getting plan D (RN_m,RN_r+i), the remaining time (tstandard-tleast-j_Tr×i)

(21)if(tstandard-tleast-j_Tr×i)≥j_Tr×kthen

(22) getting plan E (RN_m+k,RN_r+i), the remaining time (tstandard-tleast-j_Tr×i-j_Tm×k)

(23)elsethen

(24) planjr_t={A, D}

(25)endif

(26) planjr_t={A, D, E}

(27)endif

(28)Planj_t={A}∪planjm_t∪planjr_t

(29)endif

(30)endwhile

(31)endfor

2.3 基于最大輪數的作業調度算法—MRNS

在TRN算法確定出Map、Reduce任務最大輪數組合方案和最大標準時間后，MRNS算法結合平臺中現有資源數量，考慮平臺資源利用率，制定出對于當前所有作業的調度策略，以實現收益最大化的目標。MRNS算法如下所示：

算法2：MRNS algorithm

Input：Planj_t: the maximum round numbers plans

tstandard: the maximum standard time for a job

N: the number of jobs

δ: the threshold of the resource utilization

Output： TS: the optimal scheduling strategy based on tasks

F: the maximum profit

(1)forj=0;j

(2) whenj_tstandard=j_deadline/a, calculate f(j_profit)

(3)F=F+j_profit

(4)endfor

(5)F=F+max(f(j_profit))

(6)jobjof the maximum reward is added to scheduling queue

(7)whenj_tstandard=j_deadline/a,Planj_tof the jobjis selected

(8)whilePlanj_tdo

(9) recording the completion timej_completion

(10) calculating the remaining resourcesrwhen the jobjbegins to be scheduled till the end

(11)if(R-r)/R≥δthen

(12)idlingtheremainingresourcesandwaitingforthepreviousjobcompletion

(13)fori=0;i

(14)ifj_completion≥i_deadlinethen

(15) the jobiis added to the end of scheduling queue

(16)elseifi_deadline-j_completion≥i_1tstandard+j_completionthen

(17) the jobigets the reward f(i_profit)

(18)elseifi_2tstandard+j_completion≤i_deadline-j_completionthen

(19) the jobigets the reward 0

(20)elseifi_3tstandard+j_completion≤i_deadline-j_completionthen

(21) the jobigets the punishment f(i_profit)elsethen

(22) the jobigets the maximum punishment f(i_profit)

(23)endfor

(24) according to f(i_profit) for each job, sortingN-1 jobs

(25)if?f(i_profit)≥0then

(26) the jobithat has the maximum reward is added to the scheduling queue

(27)elseif?f(i_profit)<0then

(28) according to sorting results, find the jobimee-ting f(i_profit)>(-i_profit×γ) & the minimum f(i_profit)

(29) the jobiis added to the scheduling queueelsethen

(30) ?f(i+n_profit)<0,f=|-i+1_profit×γ|+|-i+2_profit×γ|+…+|-i+n_profit×γ|

(31)iff(i_profit)≥fthen

(32) the jobiis added to the scheduling queueelsethen

(33) find the jobi+nmeeting f(i+n_pro-fit)=-i+n_profit×γthat is the minimum

(34) the jobi+nis added to the scheduling queue

(35)elsethen

(36) when the scheduled jobjhas the remaining resources, recording the start timetrfor the remaining resources

(37)fori=0;i

(38)iftr≤i_1tstandardthen

(39) the remaining resourcesrmeetsPlani_tthat on thei_1tstandard

(40) Comparingtr+i_1tstandardwithi_tstandard

(41) the jobiget the maximum |f(i_profit)| besides the maximum punishment

(42)ifi_1tstandard≤tr≤i_2tstandardthen

(43) the remaining resourcesrmeetsPlani_tthat on thei_2tstandard

(44) Comparingtr+i_2tstandardwithi_tstandard

(45) the jobiget the maximum |f(i_profit)| besides the maximum punishment

(46)endfor

(47) the jobiis scheduled that has the maximum |f(i_profit)|

(48)F=F+f(i_profit)

(49)endwhile

(50)F=F+f(i_profit)

算法2中首先根據所有作業在j_tstandard=j_deadline/a時的最大獎勵值f(j_profit)，確定被調度的作業并且選擇最大輪數組合方案集(line(1)-line(7))；其次，在前一作業的所有輪數組合方案集中，選擇具有局部最大收益的輪數組合方案進行調度(line(8)-line(49))；最后將所有作業加入到調度隊列后，根據每個作業相應的任務輪數組合方案，計算全局的最大收益值(line(50))。其中，在選擇具有局部最大收益的輪數組合方案時，優先考慮平臺的資源利用率。當資源利用率大于給定閾值時，則空閑資源，直到前一作業執行完成后，在剩余作業中選擇局部收益最大的作業進行串行調度(line(11)-line(34))。在選擇局部收益最大的作業時，通過每個作業的截止時間與前一作業的完成時間差所在范圍，比較每個作業在該時間范圍內可獲得的獎勵或懲罰值，選擇出使得局部收益最大的作業和相應的任務輪數組合方案。當資源利用率小于給定閾值時(line(35)-line(48))，將要調度的作業與前一作業并行執行，充分利用平臺資源，使得資源利用率達到最大。在選擇合適的作業加入到調度隊列之前，記錄前一作業有空余資源時的開始時間點tr，tr與每個作業的最大標準時間tstandard進行比較，當剩余資源能夠滿足最大標準時間范圍內的任務最大輪數要求時，則選擇 |f(j_profit)| 值最大的作業加入到調度隊列，對于懲罰值已經達到最大的作業則放在調度隊列最后進行調度。由于每個被選中的作業都有多種最大輪數組合方案，故在每選擇出一種調度組合方案后則計算相應的局部收益值，最終選擇全局收益值最大的調度隊列與調度策略進行調度。調度策略中包含了已選擇的作業任務輪數與開始時間點，在調度前就預先指定了對于每個作業中任務的資源分配數與分配時間點。

3 實驗評估

3.1 實驗環境與設置

在基于Hadoop框架的計算平臺中對本文提出的調度器進行驗證，使用的是Hadoop 2.7.1版本。平臺中的所有節點都是同構節點，其中包含1個主節點和20個從節點。每個節點的配置信息為CPU 8 cores，16 GB RAM，1 TB hard disk, Red Hat Enterprise Linux6.2 System。資源管理和調度使用Yarn資源框架，設置Container大小為1 core和2G RAM，這樣每個節點上有8個Container，整個平臺中共有160個Container。在實驗中設置塊大小為默認的64 M。

3.2 實驗數據集與實驗指標

為評估算法性能，使用PUMA benchmark suits中的MapReduce標準應用程序，每類程序的數量、輸入數據規模、截止時間以及數據來源見表1。

表1 數據集

WordCount是CPU密集型程序，TeraSort是I/O密集型程序，Inverted-Index既是CPU密集型也是I/O密集型，3種程序均為典型的MapReduce程序。

在評估MPCRS的效率時，將MPCRS分別與FIFO、Fair、EDF調度算法在作業完成時間、服務商收益及平臺資源利用率等方面進行比較。通過以下性能指標評價算法的性能。

(1)作業完成時間：j_completion。

(2)最大收益：Profit。

(3)平臺資源利用率：Utilization。

根據作業的開始時間和執行時間可以得出作業的完成時間，j_start是作業的開始執行時間，j_execute是作業j的全部任務執行完畢所經歷的時間，默認全部的作業都同時到達，作業的完成時間j_completion表示為

j_completion=j_start+j_execute

(7)

j_profit為在SLA協議中規定的截止時間之前完成作業j可獲得的收益，根據服務商可獲得的最大收益，用RP Model中獎懲共存收益函數f(j_profit)衡量

(8)

式(3)中所提到的α，β，γ分別為獎勵比率、懲罰比率以及最大懲罰倍數，設置α=0.3，β=0.5，γ=2，a，b分別為時間限定倍數，設置a=b=1.5。

定義平臺資源利用率Utilization時，使用作業完成后所占資源與時間的乘積和平臺中整體資源與時間的乘積比值表示

(9)

其中，Rtotal是平臺中全部資源數量，Ttotal是全部作業執行完畢后的總時間，Rtaski是作業j中第i輪任務執行所占用的資源數量，Ttaski是第i輪任務所需執行時間。

3.3 實驗結果與分析

3.3.1 作業執行的高效性

如圖2所示，為4個調度器在全部作業的平均完成時間指標上的對比結果。從圖中可以看出，FIFO調度器的作業平均完成時間達到了24.3 min，依次降低是EDF調度器22.3 min、Fair調度器20.8 min以及MPCRS 18 min。FIFO調度器的作業平均完成時間最長是由于當一個作業正在執行時會占用集群中的全部資源，不能使其它作業開始執行，故這樣就會延長大部分作業的完成時間，導致全部作業的平均完成時間延長。MPCRS與FIFO、EDF以及Fair相比，作業的平均完成時間有所減短。

圖2 作業平均完成時間

如圖3所示，為每個作業在不同調度器的調度下執行完畢后的完成時間對比結果。首先可以看出3個類型的作業隨著數據量的增大，作業完成時間在不斷變長；其次在數據量相同時，不同類型的作業完成時間差距很小，說明資源的統一分配對不同類型的作業性能不會造成影響；最后通過比較在4種調度器調度下的作業不同執行性能可以看出，使用MPCRS時的作業完成時間明顯短于使用其它3種調度器時的時間，但Job9使用MPCRS時完成時間高于使用EDF和Fair的時間，這是由于在選擇Job9最大輪數執行方案時，在最大標準時間范圍內，所選的任務輪數最多，這樣會使得作業的整體完成時間較輪數較少時有所延長，而且TRN算法和MRNS算法在計算時具有一定的時間消耗，故在作業執行過程中完成時間有一定延長是不可避免的。這個完成時間的延長范圍在可允許的范圍內，是可以容忍的。

圖3 每個作業完成時間

3.3.2 收益最大化

根據上節作業執行的高效性實驗中可以得出每個作業的執行性能，但本文設計MPCRS的最終目標是使服務商獲取最大收益，故依據式(1)和式(3)，得到如圖4所示的使用不同調度器時的最大化收益。其中Ideal為總收益的理想值

圖4 服務商收益

(10)

從圖4中可以看出使用不同的調度器時，服務商可以獲得的最大收益差值較大。其中使用FIFO獲得的總收益與理想值Ideal差值最大，使用MPCRS可獲得的總收益與理想值Ideal差值最小。由于大部分作業的獎勵值仍為0和有部分延遲作業的存在，故使用MPCRS時服務商可獲得的總收益值與理想值仍有一定差距。

3.3.3 平臺資源利用率的高效性

圖5為在各個統計階段時，使用4個調度器時的平臺資源利用率。從圖中可以看出，使用FIFO時的平臺資源利用率最低，這是由于當一個作業執行時，浪費了其它空閑資源。使用EDF時同樣不能充分利用平臺資源，使得大量空閑資源浪費。使用Fair調度器時的資源利用率較高，但仍然沒有使用MPCRS時的資源利用率高。MPCRS中的MRNS算法考慮了平臺資源利用率的影響，故在各個統計階段中大部分的資源利用率都在90%以上，但在第5階段和第8階段資源利用率較低，這是在統計階段時作業執行情況的影響。

圖5 平臺資源利用率

4 結束語

結合服務商與用戶之間的利益關系，本文提出一種RP Model作為服務商收益最大化的評價標準，并以該評價標準為目標提出了MPCRS，其中具體包括TRN算法和MRNS算法。TRN算法以RP Model中的收益值為標準，根據各個作業的Map和Reduce任務執行時間，確定出作業在不同獎懲階段的Map和Reduce的最大輪數組合以及最大標準時間；MRNS算法以TRN算法得出的結果為輸入，在滿足平臺資源利用率的前提下，選擇出具有局部最大收益的作業和該作業的任務最大輪數方案，從而制定出TS策略。實驗結果表明，本文設計的作業調度器MPCRS所生成的TS策略能夠最大程度的縮短每個作業的完成時間，提高平臺資源利用率，并且在服務商獲得最大收益的同時，用戶能夠得到較為準確的截止時間，真正意義上實現了服務商和用戶的利益共贏。