李葉飛 徐超 許道強(qiáng) 鄒云峰 張曉達(dá) 錢柱中

摘 要：針對Hadoop和Spark等大數(shù)據(jù)分析系統(tǒng)中無先驗(yàn)知識(shí)任務(wù)的高效執(zhí)行問題，設(shè)計(jì)了基于累計(jì)工作量（CRW）的任務(wù)調(diào)度器CRWScheduler。該調(diào)度器根據(jù)CRW將任務(wù)在低權(quán)重隊(duì)列與高權(quán)重隊(duì)列間切換;在為作業(yè)分配資源時(shí)，同時(shí)考慮到作業(yè)所在的隊(duì)列和其瞬時(shí)占用資源量，無需作業(yè)先驗(yàn)知識(shí)即顯著提升系統(tǒng)性能?；贏pache Hadoop YARN實(shí)現(xiàn)了CRWScheduler原型，在28個(gè)節(jié)點(diǎn)的基準(zhǔn)測試集群上的實(shí)驗(yàn)表明，與YARN的公平調(diào)度機(jī)制相比，作業(yè)流時(shí)間（JFT）平均降低21%，其中95百分位的作業(yè)流時(shí)間（JFT）最多降低了35%，并且在與任務(wù)級(jí)調(diào)度程序協(xié)作時(shí)可獲得進(jìn)一步的性能提升。

關(guān)鍵詞：數(shù)據(jù)分析系統(tǒng);作業(yè)流時(shí)間;公平性;饑餓避免

中圖分類號(hào)：?TP316.4

文獻(xiàn)標(biāo)志碼：A

Online task scheduling algorithm for big data analytics based on cumulative running work

LI Yefei1， XU Chao2， XU Daoqiang3， ZOU Yunfeng2， ZHANG Xiaoda1， QIAN Zhuzhong1

1.Department of Computer Science and Technology， Nanjing University， Nanjing Jiangsu 210023， China ;

2.Electric Power Research Institute， State Grid Jiangsu Electric Power Company Limited， Nanjing Jiangsu 210000， China ;

3.State Grid Jiangsu Electric Power Company Limited， Nanjing Jiangsu 210000， China

Abstract：?A Cumulative Running Work （CRW） based task scheduler CRWScheduler was proposed to effectively process tasks without any prior knowledge for big data analytics platform like Hadoop and Spark. The running job was moved from a low-weight queue to a high-weight one based on CRW. When resources were allocated to a job， both the queue of the job and the instantaneous resource utilization of the job were considered， significantly improving the overall system performance without prior knowledge. The prototype of CRWScheduler was implemented based on Apache Hadoop YARN. Experimental results on 28-node benchmark testing cluster show that CRWScheduler reduces average Job Flow Time （JFT） by 21% and decreases JFT of 95th percentile by up to 35% compared with YARN fair scheduler. Further improvements can be obtained when CRWScheduler cooperates with task-level schedulers.

Key words：?data analytics system; Job Flow Time （JFT）; fairness; starvation-free

0 引言

在現(xiàn)代主流分布式大數(shù)據(jù)分析系統(tǒng)（例如Apache Hadoop、Apache Spark、Apache Tez）中，作業(yè)是由一系列任務(wù)構(gòu)成的，可表示為有向無環(huán)圖（Directed Acyclic Graph， DAG），其中每個(gè)頂點(diǎn)表示一個(gè)任務(wù)，邊表示任務(wù)之間輸入與輸出的依賴關(guān)系。為提升資源利用率，作業(yè)之間會(huì)共享集群資源[1-3]。用戶級(jí)公平性確保所有用戶都能夠在共享集群中獲得資源[4-10]。然而，從用戶的角度看，除了可以獲得的資源量，如何高效地利用已分配的資源來快速完成其作業(yè)同樣是一個(gè)重要問題。作業(yè)完成效率一般用作業(yè)流時(shí)間（Job Flow Time， JFT）來刻畫，其值定義為作業(yè)完成時(shí)間減去作業(yè)提交時(shí)間。本文針對分布式大數(shù)據(jù)分析系統(tǒng)的作業(yè)調(diào)度問題，研究任務(wù)調(diào)度算法，在確保用戶之間公平性以及無饑餓的前提下，加速DAG作業(yè)的完成。

即使是在離線場景（offline setting）中，最小化平均作業(yè)流時(shí)間的DAG調(diào)度問題都是NP難的。而Hadoop和Spark中的默認(rèn)集群調(diào)度機(jī)制將資源劃分為固定大小的資源塊（Slots），例如〈2個(gè)CPU內(nèi)核， 1GB內(nèi)存〉，并將資源塊提供給作業(yè)。在此過程中調(diào)度器并未考慮作業(yè)的執(zhí)行進(jìn)展情況，使得作業(yè)完成時(shí)間變長[4，10-11]。為此，Carbyne[12]和Graphene[13]等優(yōu)化的作業(yè)調(diào)度機(jī)制將不同的任務(wù)分類以提高集群整體的吞吐率，同時(shí)根據(jù)初始的作業(yè)量（所有任務(wù)正在處理的時(shí)間總和）以短作業(yè)優(yōu)先（Shortest Job First， SJF）[14]的方式調(diào)度DAG作業(yè)，從而減少了平均作業(yè)流時(shí)間。一方面，短作業(yè)優(yōu)先的策略會(huì)導(dǎo)致長作業(yè)饑餓，損害作業(yè)級(jí)的公平性;另一方面，這種調(diào)度機(jī)制預(yù)先假設(shè)已知作業(yè)的全部先驗(yàn)知識(shí)，如：DAG的結(jié)構(gòu)和任務(wù)處理時(shí)間。對于一些重復(fù)出現(xiàn)的作業(yè)確實(shí)可以在當(dāng)前集群中預(yù)測其特征[15-17]，但在一般的情況下，由于集群的動(dòng)態(tài)性（任務(wù)中斷、推測執(zhí)行）等難以預(yù)先獲取完全的作業(yè)信息，從而限制了這些調(diào)度機(jī)制的實(shí)用性[2-3]。

LAS （Least-Attained Service）[17]是一種基于作業(yè)提交后已展現(xiàn)的運(yùn)行信息來進(jìn)行調(diào)度的策略，由于它不需要提前獲取完全信息，具有更強(qiáng)的實(shí)用性;并且，當(dāng)作業(yè)的工作量服從重尾分布時(shí)，LAS能近似最優(yōu)策略的性能[3]。但傳統(tǒng)LAS策略僅考慮了計(jì)算量一個(gè)維度，而大數(shù)據(jù)處理作業(yè)對CPU與內(nèi)存的消耗具有同等重要的地位。為此，本文基于LAS思想，提出了支持多類型資源（CPU、內(nèi)存等）的非饑餓集群調(diào)度機(jī)制CRWScheduler。CRWScheduler基于主資源[4]的累計(jì)運(yùn)行工作量（Cumulative Running Work， CRW）來預(yù)測作業(yè)的總工作量，從而無需提前獲取作業(yè)特征，即可減少平均作業(yè)流時(shí)間。

通過使用權(quán)重多隊(duì)列架構(gòu)，CRWScheduler將基于CRW的啟發(fā)式算法（有利于重尾分布）和先進(jìn)先出（First In First Out， FIFO）思想（有利于輕尾分布）組合起來。每個(gè)作業(yè)基于其CRW會(huì)被分配到特定的隊(duì)列中， CRW較小的作業(yè)隊(duì)列被設(shè)置較低的權(quán)重;而在分配資源時(shí)，CRWScheduler首先按照“分?jǐn)?shù)”將隊(duì)列進(jìn)行排序，然后依據(jù)分?jǐn)?shù)遞增的順序分配資源。分?jǐn)?shù)的計(jì)算同時(shí)依賴隊(duì)列中作業(yè)當(dāng)前占用資源的情況和隊(duì)列的權(quán)重。一方面，當(dāng)前不占用資源的隊(duì)列其分?jǐn)?shù)為0，有最高優(yōu)先級(jí)獲得新資源，又由于每個(gè)隊(duì)列中作業(yè)為先進(jìn)先出，進(jìn)而避免了作業(yè)出現(xiàn)饑餓的情況。另一方面，由于CRW較小的隊(duì)列的權(quán)重較小，相比有相同資源的隊(duì)列有更高的優(yōu)先級(jí)獲得新資源。這樣做模擬了短作業(yè)優(yōu)先的策略，降低了平均作業(yè)流時(shí)間。有別于PIAS （Practical Information-Agnostic flow Scheduling）[18]和Aalo[19-20]等其他基于LAS策略的網(wǎng)絡(luò)流調(diào)度機(jī)制，CRWScheduler避免了PIAS等調(diào)度機(jī)制導(dǎo)致作業(yè)饑餓的問題，也克服了Aalo等機(jī)制需要無限分割資源，但并不能實(shí)際滿足DAG作業(yè)的最小資源需求的缺陷。

本文基于Apache Hadoop YARN （Yet Another Resource Negotiator）[21]實(shí)現(xiàn)了CRWScheduler，并將系統(tǒng)部署在28個(gè)節(jié)點(diǎn)的集群中。通過全面的工作負(fù)載測試結(jié)果顯示，與公平調(diào)度相比，CRWScheduler的平均作業(yè)流時(shí)間縮短了10%～35%，而其中95百分位的作業(yè)流時(shí)間縮短了21%～35%。

1 研究現(xiàn)狀

大數(shù)據(jù)分析系統(tǒng)

目前的數(shù)據(jù)分析系統(tǒng)大致可分為資源管理和計(jì)算框架兩類。資源管理平臺(tái)，如Apache YARN[21]和Apache Mesos[7]。從物理機(jī)器抽象出資源，使彈性計(jì)算框架易于構(gòu)建和有效運(yùn)行。研究人員和從業(yè)人員已經(jīng)開發(fā)出了多種易于編程的框架，如Apache Tez[1]、MapReduce[2]、Dryad[3]和Apache Spark[3]。這些框架通常將作業(yè)看成DAG，從而使得它們可以表達(dá)對資源的偏好和約束。吳信東等[22]比較了MapReduce和Spark在不同場景中的性能。

大數(shù)據(jù)分析的任務(wù)調(diào)度

延遲調(diào)度[10]和Quincy[23]通過調(diào)度單個(gè)任務(wù)以更接近于其輸入數(shù)據(jù)，從而提高單個(gè)任務(wù)的數(shù)據(jù)本地性。ShuffleWatcher[24]提高了階段（stage）之間的shuffle局部性;Tetris[15]確保多臺(tái)機(jī)器能更好地打包任務(wù);Corral[16]同時(shí)考慮了結(jié)合數(shù)據(jù)和計(jì)算位置的優(yōu)化。這些工作集中于任務(wù)調(diào)度級(jí)別，同時(shí)CRWScheduler能夠更好地與其協(xié)作。

對于調(diào)度一個(gè)單獨(dú)的作業(yè)，列表調(diào)度和工作竊取調(diào)度[25]是常見的異步優(yōu)化方案，比如對于多個(gè)DAG作業(yè)共享一個(gè)并行多主機(jī)集群，LAPS （Latest Arrival Processor Sharing）[26]有良好的性能，然而并沒有得到應(yīng)用。文獻(xiàn)[13]中證明了短作業(yè)優(yōu)先（SJF）算法與LAPS有相近的性能，并且不涉及參數(shù)。王習(xí)特等[27]提出了基于序列的任務(wù)調(diào)度策略SEQ（SEQuence-based task scheduling strategy）來最大化集群運(yùn)行作業(yè)的收益。本文驗(yàn)證了CRWScheduler能夠在無先驗(yàn)知識(shí)的情況下達(dá)到近似于SJF的性能。

2 DAG作業(yè)調(diào)度

首先形式化地定義DAG作業(yè)調(diào)度問題以及一些相關(guān)的概念，然后量化分析實(shí)際生產(chǎn)中的工作負(fù)載，最后用一個(gè)例子展示基于CRW的調(diào)度機(jī)制所獲得的性能提升。

2.1 問題定義

本節(jié)定義DAG作業(yè)調(diào)度問題，之后給出問題的計(jì)算復(fù)雜度，最后形式化地定義累計(jì)運(yùn)行工作量（CRW）。

假設(shè)一個(gè)用戶在時(shí)刻0提交了若干個(gè)作業(yè)J，每個(gè)作業(yè)包含若干個(gè)任務(wù)，crij表示第i個(gè)作業(yè)的第j個(gè)任務(wù)對于資源r的需求量。假設(shè)dij表示相應(yīng)任務(wù)的處理時(shí)間，且dij并未考慮持續(xù)時(shí)間內(nèi)位置環(huán)境的影響。Dij表示表示作業(yè)i中任務(wù)j的依賴任務(wù)集合，只有當(dāng)Dij中的任務(wù)全部完成，作業(yè)i中任務(wù)j才能被調(diào)度執(zhí)行。整個(gè)集群包含機(jī)器的集合為P，每臺(tái)機(jī)器s包含資源r的容量為prs。假設(shè)Aij代表作業(yè)i中任務(wù)j分配的機(jī)器，Bij表示任務(wù)分配發(fā)生的時(shí)間，Itij指示在時(shí)刻t，作業(yè)i的任務(wù)j是否在集群上運(yùn)行。

定義1? 作業(yè)i的流時(shí)間為其到達(dá)時(shí)間與其最后一個(gè)任務(wù)完成時(shí)間的間隔時(shí)間，即：max j∈i {t>0 | Itij>0}。

約束條件：

1）在任何時(shí)刻，一臺(tái)機(jī)器上的所有任務(wù)使用的資源總和不得超過當(dāng)前機(jī)器相應(yīng)資源的容量：

r，t，s∈P，∑ Aij=s crij·ltij≤Prs

（1）

2）調(diào)度器不能在一個(gè)任務(wù)還未完成前中斷任務(wù)：

Itij= 1，?? Bij≤t≤Bij+dij0， 其他

（2）

3）一個(gè)任務(wù)只有等到其所依賴的任務(wù)全部執(zhí)行完才能被調(diào)度執(zhí)行：

Bij≥max k∈Dij {Bik+dik}

（3）

目標(biāo)函數(shù)：最小化平均作業(yè)流時(shí)間（即最小化作業(yè)流時(shí)間之和）：

∑ i∈J max j∈i {t>0 | Itij>0}

（4）

定理1? 滿足上述目標(biāo)（4）和約束條件（1）～（3）的DAG作業(yè)調(diào)度問題為NP難問題。

證明? 通過證明該問題的一個(gè)特殊情況是NP難來證明該問題的計(jì)算復(fù)雜性。對于所有作業(yè)中的任務(wù)，假設(shè)Dij為空集且任務(wù)的執(zhí)行時(shí)間為1;假設(shè)在0時(shí)刻共有 | P | 個(gè)箱子。若能夠?qū)⑺腥蝿?wù)打包起來所需的箱子數(shù)小于 | P | ，那么所有作業(yè)的完成時(shí)間為1;否則在1時(shí)刻開啟另外 | P | 個(gè)箱子，來判斷是否能將剩余的任務(wù)打包。因?yàn)榕袛嘧钌俚南渥訑?shù)是NP難的[20]，故DAG作業(yè)調(diào)度問題是NP難的。

注意到即使沒有任務(wù)位置偏好，DAG作業(yè)調(diào)度問題仍然為NP難。鑒于NP難問題的復(fù)雜度[28]，本文在考慮非饑餓的同時(shí)尋找有效的啟發(fā)式方法以最小化平均流時(shí)間。

為了論述方便，預(yù)先定義以下概念。

定義2? 作業(yè)i的總工作量（Original Work）是指對于特定資源，任務(wù)需求與任務(wù)處理時(shí)間的乘積與集群資源容量的最大比率，即：

R（i）=max r? 1 ∑ s∈P prs ∑ j∈i crij·dij

定義3? 作業(yè)i在t時(shí)刻的當(dāng)前運(yùn)行任務(wù)數(shù)為：

CR（i，t）=∑ j∈i Itij

定義4? 作業(yè)i在t時(shí)刻的累計(jì)運(yùn)行工作量（Cumulative Running Work， CRW）是指對于每種資源，該作業(yè)已經(jīng)完成的任務(wù)和正在運(yùn)行任務(wù)中對該資源的需求量和可用量比值的最大值，即：

RW（i，τ）=max r? 1 ∑ s∈P prs ∑ j∈i ∑ t≤τ crij·Irij

2.2 工作負(fù)載分析

如表1所示，Yahoo！采集了在2500個(gè)節(jié)點(diǎn)的集群中的作業(yè)所使用資源（以Slot數(shù)量計(jì)算）的情況。作業(yè)的資源使用情況在實(shí)際生產(chǎn)中滿足重尾分布[29]：即大部分的作業(yè)都屬于輕量級(jí)，它們只占用少量的集群資源;小部分繁重的作業(yè)占據(jù)著大部分的集群資源。

在重尾分布中，一個(gè)作業(yè)已經(jīng)完成的程度將是其實(shí)際工作的一個(gè)好的預(yù)測值[13]。在本文所討論的場景中，CRW很好地近似了作業(yè)總工作量，因此基于CRW的調(diào)度機(jī)制（CRW較小的作業(yè)優(yōu)先）在實(shí)際負(fù)載工作中近似于短作業(yè)優(yōu)先（SJF）。

2.3 典型實(shí)例分析

圖1以具體的例子顯示了現(xiàn)有調(diào)度器的缺陷：公平調(diào)度器（fair scheduler）總是將資源分配給當(dāng)前資源占有量最少的作業(yè)[30];容量調(diào)度器（capacity scheduler）根據(jù)作業(yè)提交時(shí)間分配資源[31];SJF[14]能夠?qū)崿F(xiàn)最優(yōu)的調(diào)度，但是它依賴于作業(yè)提交時(shí)完全的先驗(yàn)的作業(yè)特性。

假設(shè)用戶在時(shí)刻0提交作業(yè)A和B，在時(shí)刻1提交作業(yè)C。每個(gè)任務(wù)需要1個(gè)單位的處理時(shí)間，每個(gè)任務(wù)有不同的資源需求。假設(shè)所有資源的總量為1。作業(yè)流時(shí)間定義為從作業(yè)的提交時(shí)間到作業(yè)最后一個(gè)任務(wù)完成的時(shí)間間隔?；诠潭ㄙY 源塊（每個(gè)slot的資源為〈0.5， 0.5〉）的公平調(diào)度器以實(shí)時(shí)最大 最小公平（max-min fairness）的方式為作業(yè)分配資源[30]。公平調(diào)度器作用下的平均作業(yè)流時(shí)間為17/3，而容量調(diào)度器導(dǎo)致平均流時(shí)間也為17/3?？紤]已知DAG結(jié)構(gòu)和細(xì)粒度任務(wù)需求，SJF調(diào)度策略將會(huì)根據(jù)其總工作量逆序調(diào)用作業(yè)，使得平均作業(yè)流時(shí)間為5?；贑RW的調(diào)度器將資源分配給具有最小累計(jì)運(yùn)行工作量的作業(yè)。在時(shí)刻1，基于CRW的調(diào)度器將為作業(yè)C分配2個(gè)任務(wù)，而不是1個(gè)，因?yàn)樽鳂I(yè)A、B、C在時(shí)刻1的CRW分別為0.4、0.5、0?；贑RW的調(diào)度器可使得平均作業(yè)流時(shí)間為16/3。在這個(gè)例子中，基于CRW的調(diào)度器不需要先驗(yàn)的作業(yè)特性，但能夠近似最優(yōu)的調(diào)度器，實(shí)現(xiàn)更低的平均作業(yè)流時(shí)間。

基于此，本文試圖根據(jù)作業(yè)的累計(jì)運(yùn)行工作量設(shè)計(jì)在線的動(dòng)態(tài)調(diào)度器來分配資源以保證無饑餓的同時(shí)減少平均作業(yè)流時(shí)間。

3 基于CRW的作業(yè)調(diào)度機(jī)制

如圖2所示，在分配資源時(shí)，用戶級(jí)調(diào)度機(jī)制根據(jù)用戶在集群中的份額排序。公平調(diào)度機(jī)制為最?。▋?nèi)存）份額的用戶提供資源[10]，然而主資源公平的調(diào)度機(jī)制（Dominate Resource Fairness， DRF）將其擴(kuò)展為考慮多類資源（如CPU、內(nèi)存）[4]。任務(wù)調(diào)度器嘗試提高任務(wù)的數(shù)據(jù)本地性[10]，或者最小化集群資源碎片[15]，或者根據(jù)任務(wù)依賴約束進(jìn)行分類[12]。CRWScheduler位于用戶之間的調(diào)度和任務(wù)調(diào)度之間，將每個(gè)用戶內(nèi)部的作業(yè)作為輸入，試圖縮短平均作業(yè)流時(shí)間。

3.1 權(quán)重多隊(duì)列框架

在多個(gè)DAG作業(yè)共享多臺(tái)機(jī)器的場景中，作業(yè)到達(dá)和離開的時(shí)間都是在線的，難以獲得完整的作業(yè)信息，因此最短作業(yè)優(yōu)先原則實(shí)際上是無法實(shí)施的。因此，依據(jù)2.2節(jié)分析和定義4，累計(jì)工作量是作業(yè)總工作量的良好的預(yù)測值，一個(gè)直觀的啟發(fā)式方法是基于累計(jì)運(yùn)行工作量的思想近似最短作業(yè)優(yōu)先。

在t時(shí)刻，為作業(yè)i賦予分?jǐn)?shù)CR（i， t） * CRW（i， t），按照分?jǐn)?shù)非遞減的方式分配資源。這樣做的目的是在保證饑餓不發(fā)生的前提下減少平均作業(yè)流時(shí)間：因?yàn)轲囸I作業(yè)的分?jǐn)?shù)為0（因?yàn)镃R（i， t）=0），在分配資源時(shí)具有最高的優(yōu)先級(jí);并且，較低CRW的作業(yè)有較高的優(yōu)先級(jí)，又因?yàn)镃RW可以很好地評估一個(gè)作業(yè)的總工作量，所以這個(gè)啟發(fā)式方法有助于最小化平均作業(yè)流時(shí)間。

然而，對于滿足輕尾的作業(yè)分布，上述啟發(fā)式算法實(shí)際上就退化成了作業(yè)之間公平共享資源，這并不能最小化平均作業(yè)流時(shí)間。例如，兩個(gè)相同的作業(yè)等待調(diào)度執(zhí)行，其中一個(gè)作業(yè)被選中執(zhí)行，這個(gè)作業(yè)的分?jǐn)?shù)提高了，將導(dǎo)致另一個(gè)作業(yè)被調(diào)度執(zhí)行，因此，兩個(gè)作業(yè)的完成時(shí)間都變成了最優(yōu)時(shí)間的兩倍。而在這種情況下，F(xiàn)IFO調(diào)度策略能最小化平均完成時(shí)間。因此，本文算法通過權(quán)重多隊(duì)列框架將基于CRW的機(jī)制和FIFO策略組合起來，具體介紹見3.2節(jié)。

3.2 DAG作業(yè)調(diào)度CRWScheduler

本文通過將基于CRW的啟發(fā)式思想和FIFO的啟發(fā)式思想綜合起來考慮，引入權(quán)重多隊(duì)列框架（Weighted Multi-Queue Framework，WMQF）。CRWScheduler基于運(yùn)行中作業(yè)的CRW值將其劃分到相應(yīng)的隊(duì)列Q1，Q2，…，Qn中去。在隊(duì)列Qi中的作業(yè)的CRW值屬于[Qi-1.th，Qi.th]，其中Qi.th是隊(duì)列Qi的閾值同時(shí)滿足Qi+1.th>Qi.th;且 Q0.th=0。CRWScheduler從Q1到Qn以遞減的方式設(shè)置隊(duì)列權(quán)重Qi.weight。一個(gè)作業(yè)的CRW值隨著其任務(wù)在集群中執(zhí)行單調(diào)遞增，一旦其CRW值超出所在隊(duì)列的閾值，當(dāng)前作業(yè)將從當(dāng)前隊(duì)列轉(zhuǎn)移到下一個(gè)隊(duì)列。

CRWScheduler實(shí)時(shí)記錄每個(gè)作業(yè)的CRW值，同時(shí)在必要時(shí)調(diào)整作業(yè)所在的隊(duì)列。當(dāng)一個(gè)作業(yè)周期性地向集群報(bào)告其心跳時(shí)，將調(diào)用UPDATECRW：為作業(yè)分配新的slot（Cn）以執(zhí)行其任務(wù)，同時(shí)返回完成后的slot（Cr）給集群調(diào)度器。當(dāng)作業(yè)更新后的CRW值超出隊(duì)列的閾值時(shí)，該作業(yè)將會(huì)轉(zhuǎn)移到下一個(gè)隊(duì)列（代碼第7）行）。

Qr.score=Qr.weight· 1 ?| Qr | ??∑ i∈Qr CR（i，τ）

（5）

算法1

DAG作業(yè)調(diào)度CRWScheduler。

程序前

procedure UPDATECRW（JOB i， Cn， Cr）

//在作業(yè)向集群報(bào)告其心跳時(shí)調(diào)用

1）

R ←當(dāng)前累計(jì)運(yùn)行資源向量

2）

k←作業(yè)i的隊(duì)列index

3）

在新分配的slot Cn中啟動(dòng)任務(wù)

4）

fo r c∈Cr∪Cn do：

5）

R +=c· r? （time.Now-c.lastUpdateTime）

6）

if? max（ R ）>Qk..th then

7）

將作業(yè)i從隊(duì)列Qk轉(zhuǎn)移到Qk+1

procedure CRWSCHEDULE（NODE n）

//在機(jī)器向集群報(bào)告其心跳時(shí)調(diào)用

8）

wh ile n has free resources do

9）

decide user u to offer resources to

10）

jobList=SCOREDLIST（u）

11）

offer resources to jobList

procedure SCOREDLIST（USER u）

//將作業(yè)排序

12）

n←用戶u所在的隊(duì)列數(shù)

13）

fo r i=1 to n do：

14）

使用式（5）計(jì)算Qi.score

15）

根據(jù)score對隊(duì)列進(jìn)行排序

16）

jlist←連接有序隊(duì)列中的作業(yè)

17）

return jlist

程序后

CRWScheduler維持了用戶級(jí)別的公平性（代碼第9）行），并在每個(gè)用戶內(nèi)部調(diào)度作業(yè)。當(dāng)一臺(tái)機(jī)器周期性地報(bào)告其心跳時(shí)，CRWSchduler被調(diào)用，同時(shí)將重新分配資源。通過設(shè)置權(quán)重隨隊(duì)列索引遞減，CRWScheduler使用式（5）計(jì)算每個(gè)隊(duì)列的分?jǐn)?shù)（代碼第14）行），其中∑ i∈ Q r CR（i，τ）表示的是隊(duì)列Qr中當(dāng)前所有作業(yè)正在運(yùn)行的任務(wù)數(shù)量之和， | Qr | 表示的是隊(duì)列Qr中當(dāng)前的作業(yè)的數(shù)量。也就是說，Qr.score為隊(duì)列中當(dāng)前運(yùn)行任務(wù)的平均值與隊(duì)列權(quán)重的乘積。按照Qr.score非遞減的順序排序可使得CRWSchduler為重尾作業(yè)分布保留了基于CRW的啟發(fā)式特性：因?yàn)橛休^低CRW的作業(yè)所在隊(duì)列的權(quán)重較大，導(dǎo)致其作業(yè)排在jlist前面;與此同時(shí)，也保證了FIFO思想在輕尾作業(yè)分布中的優(yōu)勢，即在同一隊(duì)列中的作業(yè)是FIFO的，從而使CRWScheduler有效地防止了隊(duì)列和作業(yè)中饑餓現(xiàn)象的發(fā)生，同時(shí)縮短了平均作業(yè)流時(shí)間。

3.3 CRWScheduler的實(shí)現(xiàn)

Tez[1]、Hadoop[21]和Spark[32]等經(jīng)典的數(shù)據(jù)分析系統(tǒng)將集群功能劃分為多個(gè)角色，包括集群管理器（resource manager）、節(jié)點(diǎn)管理器（node manager）與作業(yè)管理器（job manager），圖3中帶“+”的部分為CRWScheduler在已有架構(gòu)上的修改與擴(kuò)展。為了保證系統(tǒng)的擴(kuò)展性、性能以及可靠性，類YARN的數(shù)據(jù)分析系統(tǒng)[10]將集群功能劃分為若干個(gè)不同的守護(hù)進(jìn)程。本文通過如下修改將CRWScheduler整合到Apach Hadoop YARN框架（Hadoop 2.6）中。集群管理器為每個(gè)用戶管理多個(gè)隊(duì)列框架，同時(shí)在資源分配中實(shí)現(xiàn)CRWScheduler。節(jié)點(diǎn)管理器進(jìn)行slot分配的同時(shí)追蹤slot的運(yùn)行時(shí)間。作業(yè)管理器負(fù)責(zé)增加自身的CRW值，并在必要時(shí)使用UPDATECRW調(diào)整其所在的隊(duì)列。

對Apache Hadoop YARN框架所做的修改并不影響原有調(diào)度器的可擴(kuò)展性，因?yàn)镃RWScheduler在分配資源時(shí)將隊(duì)列排序，而原有的調(diào)度器是直接對作業(yè)進(jìn)行排序。CRWScheduler在作業(yè)更新時(shí)增加了累計(jì)運(yùn)行工作計(jì)算，這是一個(gè)工作量適中的操作，同時(shí)實(shí)驗(yàn)結(jié)果也顯示這一花銷并不會(huì)影響系統(tǒng)的性能。

4 實(shí)驗(yàn)評估

4.1 實(shí)驗(yàn)配置

本文在28個(gè)節(jié)點(diǎn)的集群中運(yùn)用各種類型的工作負(fù)載評估CRWScheduler，并且以仿真的方式在更大的集群中評測CRWScheduler的開銷。

集群中的每個(gè)節(jié)點(diǎn)采用2核CPU，4GB內(nèi)存，1Gb/s NIC網(wǎng)卡，運(yùn)行Ubuntu14.04系統(tǒng)。

對比方法：將CRWScheduler與Hadoop 2.6中實(shí)現(xiàn)的調(diào)度算法進(jìn)行對比。為了與公平調(diào)度器[10]和DRF調(diào)度器[4]相比較，CRWScheduler替換了這些調(diào)度機(jī)制中的作業(yè)調(diào)度組件，同時(shí)保留了用戶之間的調(diào)度組件。同樣將CRWScheduler與容量調(diào)度器[4]相比較。本文利用延遲調(diào)度機(jī)制[10]作為默認(rèn)的任務(wù)調(diào)度機(jī)制。

性能指標(biāo)：主要關(guān)注于提升作業(yè)流時(shí)間（平均作業(yè)流時(shí)間或者絕大多數(shù)作業(yè)流時(shí)間）和作業(yè)吞吐量。

測試負(fù)載：使用Apach Spark來生成運(yùn)行在YARN上的DAG作業(yè)。評估中使用的作業(yè)來源于如下數(shù)據(jù)分析應(yīng)用：WordCount、TPC-H基準(zhǔn)、迭代機(jī)器學(xué)習(xí)以及PageRank。如表2所示，對于每一類工作負(fù)載，數(shù)據(jù)集大小仿照真實(shí)生產(chǎn)環(huán)境的集群按比例縮放（Yahoo！[8]和Facebook[10]）。對于作業(yè)的分布，設(shè)置46%、40%和14%分別為小型、中型、大型的作業(yè)，這與實(shí)際應(yīng)用中作業(yè)的分布相似[10]。

4.2 實(shí)驗(yàn)結(jié)果

本文首先討論CRWScheduler在多用戶環(huán)境下相比現(xiàn)有調(diào)度策略在性能上的提升，同時(shí)集中分析對于單用戶作用吞吐率的提升效果，之后將隔離提交作業(yè)以模擬輕尾工作負(fù)載，最后評估經(jīng)過修改的架構(gòu)與初始架構(gòu)相比所產(chǎn)生的額外開銷。

1）多用戶環(huán)境實(shí)驗(yàn)。假設(shè)有7個(gè)用戶，為每個(gè)用戶生成300個(gè)作業(yè)。圖4顯示了當(dāng)CRWScheduler與公平調(diào)度器協(xié)作（CRWScheduler-fair）以及與DRF調(diào)度器協(xié)作（CRWScheduler-DRF）時(shí)的性能提升，所有的指標(biāo)均按公平調(diào)度標(biāo)準(zhǔn)化。 圖4（a）顯示，與fair scheduler相比，CRWScheduler-fair處理小型、中型、大型作業(yè)的平均JFT分別減少了14、21和10個(gè)百分點(diǎn)，對于所有的作業(yè)而言，JFT平均減少了16個(gè)百分點(diǎn);與DRF調(diào)度器相比，CRWScheduler-DRF處理小型、中型、大型作業(yè)的平均JFT分別減少了10、22以及10個(gè)百分點(diǎn)，對于所有的作業(yè)而言，JFT平均減少15個(gè)百分點(diǎn)。圖4（b）描述了95百分位的作業(yè)流時(shí)間的提升效果。

2）單用戶環(huán)境實(shí)驗(yàn)。為了突出在資源競爭激勵(lì)的環(huán)境下CRWSheduler對性能的提升，設(shè)置的實(shí)驗(yàn)環(huán)境為：單用戶每隔20s提交一次作業(yè)。對每種調(diào)度策略進(jìn)行1h的實(shí)驗(yàn)，并分別統(tǒng)計(jì)其吞吐量。如表3所示，CRW公平調(diào)度策略相于單純的公平調(diào)度策略吞吐量提升了32%，CRWSheduler-DRF調(diào)度策略相比DRF吞吐量提升了45%。

3）輕尾作業(yè)分布。實(shí)驗(yàn)為每個(gè)尺寸的作業(yè)生成100個(gè)作業(yè)，并且分開提交到服務(wù)器集群。容量調(diào)度程序在這些輕尾作業(yè)分布中進(jìn)行最優(yōu)調(diào)度。圖5顯示，CRWScheduler的調(diào)度效果近似于容量調(diào)度器，因?yàn)镃RWSchduler能保持作業(yè)在相同的隊(duì)列中進(jìn)行FIFO調(diào)度，所以優(yōu)于公平調(diào)度。

4）額外開銷：使用300節(jié)點(diǎn)集群進(jìn)行仿真實(shí)驗(yàn)，每個(gè)機(jī)架有30臺(tái)服務(wù)器，同時(shí)生成10000個(gè)作業(yè)，計(jì)算從節(jié)點(diǎn)中作業(yè)處理和更新的平均時(shí)間。表4顯示，對于每個(gè)作業(yè)心跳的額外開銷在0.05ms以內(nèi)，這對于集群管理者來說可以忽略不計(jì)。在節(jié)點(diǎn)心跳中，公平調(diào)度器直接對運(yùn)行中的作業(yè)排序，而CRWScheduler首先對隊(duì)列進(jìn)行排序，之后通過合并有序的隊(duì)列得到有序的作業(yè)列表，從而使得CRWScheduler在資源分配時(shí)能達(dá)到更短的響應(yīng)時(shí)間。

表格中的“事件平均處理時(shí)間”是算法的開銷，而“更短的響應(yīng)時(shí)間”是指之前實(shí)驗(yàn)中作業(yè)流時(shí)間這個(gè)指標(biāo)，確認(rèn)表述正確

5 結(jié)語

目前的DAG作業(yè)調(diào)度機(jī)制都需要預(yù)先獲得完整的作業(yè)信息，但在大量實(shí)際應(yīng)用中難以實(shí)現(xiàn)。為此，本文提出了累計(jì)運(yùn)行工作量（CRW），以估算作業(yè)總工作量，并設(shè)計(jì)了CRWScheduler調(diào)度器。它基于作業(yè)的CRW值將當(dāng)前運(yùn)行的作業(yè)劃分成多個(gè)隊(duì)列，同時(shí)考慮作業(yè)瞬時(shí)資源占用率以及不同的隊(duì)列權(quán)重，不僅能縮短平均作業(yè)流時(shí)間，也抑制了“饑餓”現(xiàn)象。而通過保持作業(yè)在每個(gè)隊(duì)列中的FIFO順序，CRWSheduler還提高了輕尾作業(yè)分布的性能。本文基于Apache Hadoop YARN實(shí)現(xiàn)了CRWSchduler原型系統(tǒng)，通過部署在28個(gè)節(jié)點(diǎn)的集群中進(jìn)行多類型數(shù)據(jù)分析應(yīng)用的測試。實(shí)驗(yàn)結(jié)果顯示，相比YARN的公平調(diào)度機(jī)制，CRWScheduler將平均作業(yè)流時(shí)間縮短21%，而95百分位的作業(yè)流時(shí)間縮短35%。然而，當(dāng)前任務(wù)執(zhí)行時(shí)的資源使用是隧道化的，同一時(shí)刻會(huì)同時(shí)使用多種資源。多任務(wù)同時(shí)運(yùn)行在同一物理機(jī)上時(shí)會(huì)造成不可預(yù)知的資源競爭，從而難以對任務(wù)執(zhí)行時(shí)間建模。元任務(wù)（monotask）是將任務(wù)劃分成更小的粒度，每個(gè)元任務(wù)只消耗一種資源（磁盤、網(wǎng)絡(luò)和內(nèi)存等）。通過將每個(gè)任務(wù)劃分成元任務(wù)，能夠更好地對任務(wù)和作業(yè)的執(zhí)行時(shí)間建模。下一步擬研究針對元任務(wù)的調(diào)度機(jī)制，以進(jìn)一步提高集群的資源利用率和作業(yè)性能。

參考文獻(xiàn)

[1]?Apache Software Foundation. Apache Tez [EB/OL]. [2017-12-21]. http：//tez.apache.org/.

[2]?DEAN J， GHEMAWAT S. MapReduce： simplified data processing on large clusters [C]// Proceedings of the 6th Conference on Symposium on Operating Systems Design & Implementation. Berkeley， CA： USENIX Association， 2004， 6： 10-10.

[3]?ISARD M， BUDIU M， YU Y， et al. Dryad： distributed data-parallel programs from sequential building blocks [C]// Proceedings of the 2nd ACM Special Interest Groups in Operating Systems （SIGOPS）/European Conference on Computer Systems. New York： ACM， 2007： 59-72.

[4]?ZAHARIA M， CHOWDHURY M， DAS T， et al. Resilient distributed datasets： a fault-tolerant abstraction for in-memory cluster computing [C]// Proceedings of the 9th USENIX Symposium on Networked Systems Design and Implementation. Berkeley， CA： USENIX Association， 2012： 15-28.

[5]?GHODSI A， ZAHARIA M， HINDMAN B， et al. Dominant resource fairness： Fair allocation of multiple resource types [C]// Proceedings of the 8th USENIX Symposium on Networked Systems Design and Implementation. Berkeley， CA： USENIX Association， 2011： 323-336.

[6]?GHODSI A， ZAHARIA M， SHENKER S， et al. Choosy： max-min fair sharing for datacenter jobs with constraints [C]// Proceedings of the 8th ACM European Conference on Computer Systems. New York： ACM， 2013： 365-378.

[7]?HINDMAN B， KONWINSKI A， ZAHARIA M， et al. Mesos： A platform for fine-grained resource sharing in the data center [C]// Proceedings of the 8th USENIX Symposium on Networked Systems Design and Implementation. Berkeley， CA： USENIX Association， 2011： 295-308.

[8] ?VAVILAPALLI V K， MURTHY A C， DOUGLAS C， et al. Apache hadoop YARN： yet another resource negotiator [C]// Proceedings of the 4th Annual Symposium on Cloud Computing. New York： ACM， 2013： 1-16.

[9]?WANG W， LIANG B， LI B. Multi-resource fair allocation in heterogeneous cloud computing systems [J]. IEEE Transactions on Parallel and Distributed Systems， 2015， 26（10）： 2822-2835.

[10]?ZAHARIA M， BORTHAKUR D， SARMA J S， et al. Delay scheduling： a simple technique for achieving locality and fairness in cluster scheduling [C]// Proceedings of the 5th ACM European Conference on Computer Systems. New York： ACM， 2013： 265-278.

[11]?Apache Software Foundation. Hadoop MapReduce Next Generation — Fair Scheduler [EB/OL]. [2018-10-21]. http：//tinyurl.com/j9vzsl9.

[12]??GRANDL R， CHOWDHURY M， AKELLA A， et al. Altruistic? scheduling in multi-resource clusters [C]// Proceedings of the 12th USENIX Symposium on Operating Systems Design and Implementation. Berkeley， CA： USENIX Association， 2016： 65-80.

[13]?GRANDL R， KANDULA S， RAO S， et al. Graphene： packing and dependency-aware scheduling for data-parallel clusters [C]// Proceedings of the 12th USENIX Symposium on Operating Systems Design and Implementation. Berkeley， CA： USENIX Association， 2016： 81-97.

[14]?AGRAWAL K， LI J， LU K， et al. Scheduling parallel DAG jobs online to minimize average flow time [C]// Proceedings of the 27th annual ACM-SIAM Symposium on Discrete algorithms. Philadelphia， PA： Society for Industrial and Applied Mathematics， 2016： 176-189.

[15]?FERGUSON A D， BODIK P， KANDULA S， et al. Jockey： guaranteed job latency in data parallel clusters [C]// Proceedings of the 7th ACM European Conference on Computer Systems. New York： ACM， 2012： 99-112.

[16]?GRANDL R， ANANTHANARAYANAN G， KANDULA S， et al. Multi-resource packing for cluster schedulers [C]// Proceedings of the 2014 ACM Conference on SIGCOMM. New York： ACM， 2014： 455-466.

[17]??JALAPARTI V， BODIK P， MENACHE I， et al. Network-aware ?scheduling for data-parallel jobs： Plan when you can [C]// Proceedings of the 2015 ACM Conference on SIGCOMM. New York： ACM， 2015： 407-420.

[18]?RAI I A， URVOY-KELLER G， BIERSACK E W. Analysis of LAS scheduling for job size distributions with high variance [C]// Proceedings of the 2003 ACM SIGMETRICS International Conference on Measurement and Modeling of Computer Systems. New York： ACM， 2003： 218-228.

[19]?BAI W， CHEN L， CHEN K， et al. Information-agnostic flow scheduling for commodity data centers [C]// Proceedings of the 12th USENIX Symposium on Networked Systems Design and Implementation. Berkeley， CA： USENIX Association， 2015： 455-468.

[20]?CHOWDHURY M， STOICA I. Efficient coflow scheduling without prior knowledge[C]// Proceedings of the 2015 ACM Conference on Special Interest Group on Data Communication. New York： ACM， 2015： 393-406.

[21]?Apache Software Foundation. Apache Hadoop NextGen MapReduce （YARN） [EB/OL]. [2017-12-21]. http：//tinyurl.com/zyy8kbc.

[22]?吳信東，嵇圣硙.MapReduce與Spark用于大數(shù)據(jù)分析值比較[J].軟件學(xué)報(bào)，2018，29（6）：1770-1791. （WU X D， JI S W. Comparive study on MapReduce and Spark for bid data analytics [J]. Journal of Software， 2018， 29（6）： 1770-1791.）

[23]?ISARD M， PRABHAKARAN V， CURREY J， et al. Quincy： fair scheduling for distributed computing clusters [C]// Proceedings of the ACM SIGOPS 22nd Symposium on Operating Systems Principles. New York： ACM， 2009： 261-276.

[24]?AHMAD F， CHAKRADHAR S， RAGHUNATHAN A， et al. Shufflewatcher： Shuffle-aware scheduling in multi-tenant mapreduce clusters [C]// USENIX Proceedings of 2014 USENIX Annual Technical Conference. Berkeley， CA： USENIX Association， 2014： 1-12.

[25]??BLUMOFE R D， LEISERSON C E. Scheduling multithreaded computations by work stealing [J]. Journal of the ACM. 1999， 46（5）： 720-748.

[26]?EDMONDS J， PRUHS K. Scalably scheduling processes with arbitrary speedup curves [J]. ACM Transactions on Algorithms， 2012， 8（3）： 256-265. [C]// Proceedings of the twentieth annual ACM-SIAM symposium on Discrete algorithms. New York： ACM， 2009： 685-692.

[27]?王習(xí)特，申德榮，于戈，等.MapReduce集群中最大收益問題的研究[J].計(jì)算機(jī)學(xué)報(bào)，2015，38（1）：109-121. （WANG X T， SHEN D R， YU G， et al. Research on maximum benefit problem in a MapReduce cluster [J]. Chinese Journal of Computers， 2015， 38（1）： 109-121.）

[28]?VAZIRANI V V. Approximation Algorithms [M]. Berlin： Springer， 2003： 74-78.

https：//www.logobook.ru/prod_show.php？object_uid=11146016

[29]?NAIR J， WIERMAN A， ZWART B. The fundamentals of heavy-tails： properties， emergence， and identification [C]// Proceedings of the ACM SIGMETRICS/International Conference on Measurement and Modeling of Computer Systems. New York： ACM， 2013： 387-388.

[30]?WIKIPEDIA. Max-min fairness [EB/OL]. [2017-12-21]. http：//tinyurl.com/krkdmho.

[31]?Apache Software Foundation. Hadoop MapReduce Next Generation — Capacity Scheduler [EB/OL]. [2018-12-01]. http：//tinyurl.com/j739ojm.

[32]?Apache Software Foundation. Apache Spark [EB/OL]. [2018-11-07]. http：//spark.apache.org/.