Hadoop平臺下公平調(diào)度機(jī)制的源碼研究

Hadoop平臺下公平調(diào)度機(jī)制的源碼研究

MapReduce模型MapReduce是由Google發(fā)明的一種處理大規(guī)模數(shù)據(jù)的分布式編程框架，最初是由Google的工程師設(shè)計并實(shí)現(xiàn)。Hadoop是MapReduee計算模型的一種開源實(shí)現(xiàn)，用于大規(guī)模數(shù)據(jù)集的并行化分析處理。Hadoop中的作業(yè)調(diào)度是JobTracker指派任務(wù)(Tasks)到相應(yīng)TaskTracker上執(zhí)行的過程。

公平調(diào)度機(jī)制公平調(diào)度器按資源池（pool）來組織作業(yè)，并把資源公平的分到這些資源池里。默認(rèn)情況下，每一個用戶擁有一個獨(dú)立的資源池，以使每個用戶都能獲得一份等同的集群資源而不管他們提交了多少作業(yè)。在每一個資源池內(nèi)，會使用公平共享（fairsharing）的方法在運(yùn)行作業(yè)之間共享容量（capacity）。用戶也可以給予資源池相應(yīng)的權(quán)重，以不按比例的方式共享集群。

最小共享量公平調(diào)度器允許賦給資源池保證（guaranteed）最小共享資源，這個用在確保特定用戶、群組或生產(chǎn)應(yīng)用程序總能獲取到足夠的資源時是很有用的。當(dāng)一個資源池包含作業(yè)時，它至少能獲取到它的最小共享資源，但是當(dāng)資源池不完全需要它所擁有的保證共享資源時，額外的部分會在其它資源池間進(jìn)行切分。

JobClient每一個job都會在用戶端通過JobClient類將應(yīng)用程序以及配置參數(shù)打包成jar文件存儲在HDFS，并把路徑提交到JobTracker，然后由JobTracker創(chuàng)建每一個Task并將它們分發(fā)到各個TaskTracker服務(wù)中去執(zhí)行。

JobTrackerJobTracker是一個master服務(wù)，啟動之后JobTracker會接收job，負(fù)責(zé)調(diào)度job的每個子任務(wù)task運(yùn)行于TaskTracker上，并監(jiān)控它們，如果發(fā)現(xiàn)有失敗的task就重新運(yùn)行它。一般情況應(yīng)該把JobTracker部署在單獨(dú)的機(jī)器上。

TaskTrackerTaskTracker是運(yùn)行于多個節(jié)點(diǎn)上的slaver服務(wù)。TaskTracker主動與JobTracker通信，以心跳的形式定期向JobTracker發(fā)送消息或者請求任務(wù)，并負(fù)責(zé)直接執(zhí)行每一個任務(wù)。

JobInProgressJobClient提交job后，JobTracker會創(chuàng)建一個JobInProgress來跟蹤和調(diào)度這個job，并把它添加到j(luò)ob隊列里。JobInProgress會根據(jù)提交的jobjar中定義的輸入數(shù)據(jù)集（已分解成FileSplit）創(chuàng)建對應(yīng)的一批TaskInProgress用于監(jiān)控和調(diào)度Task。

FairSchedulerFairScheduler是hadoop中的作業(yè)公平調(diào)度器，主要是解決當(dāng)TT發(fā)送心跳告訴JT當(dāng)前的空閑slots時，希望JT分配給TT相應(yīng)多個task，讓TT去執(zhí)行這些task。所以JT就需要一個調(diào)度器來對作業(yè)進(jìn)行調(diào)度，選擇出作業(yè)，然后將作業(yè)的task分配TT去執(zhí)行。而hadoop中的task可以分為map,reduce,jobsetup,jobcleanup,taskcleanup這五種task。

任務(wù)調(diào)度流程

HeatBeatTaskTacker調(diào)用JobTracker中的heatBeat()方法請求心跳回應(yīng)。JobTracker每次會對收到的heartbeat進(jìn)行判斷，如果心跳是請求新的task，且沒有Job的stup與clearn任務(wù)。JobTracker將調(diào)用FairScheduler中的assignTasks()分配任務(wù)，調(diào)用該方法將得到分配給該資源的tasklist。

assignTasks先獲取作業(yè)池中各作業(yè)待運(yùn)行的runnabletasks總數(shù)，已經(jīng)在運(yùn)行的任務(wù)runningTasks總數(shù)，獲取集群中的集群狀態(tài)資源，獲取集群中的總資源（slots）數(shù)。獲得發(fā)送該心跳的TaskTracker結(jié)點(diǎn)的狀態(tài)對象（TaskTrackerStatus）來獲得該TaskTracker可執(zhí)行的最大任務(wù)數(shù)。

assignTasksassignMultiple的值表示一次分配一個task還是分配一對task。通過一個循環(huán)體尋找可分配的任務(wù)，知道已達(dá)到分配的要求或者已無任務(wù)分配。退出代碼如下：

if(mapRejected&&reduceRejected||!assignMultiple&&tasks.size()>0){break;}

assignTasks確定任務(wù)分配的taskType，當(dāng)TaskTracker上兩種taskType都可接受時，taskType的值取決于TaskTracker上運(yùn)行的MapTask與ReduceTask的多少，哪種任務(wù)的數(shù)量比較少則分配哪種類型的任務(wù)。

Pool的排序根據(jù)FairScheduler的規(guī)則對各pool進(jìn)行排序的主要指標(biāo)：最小共享資源量（minShare）已運(yùn)行的任務(wù)數(shù)（runningTasks)資源緊缺度（Needy）共享率（minShareRatio）權(quán)值比率（tasksToWeightRatio）

Pool的排序規(guī)則資源首先分配給現(xiàn)在占有資源量少于最小共享資源量的pool如果pool的最小共享量都沒有達(dá)到，也就是Needy都為true，資源就分配給共享率低的pool如果pool的Needy都為false，資源就分配給權(quán)重比率低的pool如果minShareRatio、tasksToWeightRatio兩者有一個相同，則依據(jù)pool的startTime。

Pool中Job排序在FairScheduler.java中的assignTasks中通過對PoolSchedulable進(jìn)行排序之后，只是確定了分配資源給pool的優(yōu)先順序，并沒有確定將資源分配給pool中資源的順序，代碼：assignTask(tts,currentTime,visitedForMap):是調(diào)用了PoolSchedulable中的任務(wù)分配方法，結(jié)果返回一個將獲得該資源的task，如果不能分配返回null。

Pool中Job排序pool中的job排序規(guī)則可以由用戶選擇，目前支持兩種排序規(guī)則：FIFO:優(yōu)先級決定FairScheduler：jobweight的值決定

Pool中Job排序pool中的job排序規(guī)則可以由用戶選擇，目前支持兩種排序規(guī)則：FIFO:優(yōu)先級決定FairScheduler：jobweight的值決定

Job權(quán)重計算在對job進(jìn)行排序時，job的權(quán)重是個非常重要的因素，job的權(quán)重越高獲得的資源也將更多，如果job還未運(yùn)行其權(quán)重默認(rèn)為1.0，已運(yùn)行的job的權(quán)重將與需求量、優(yōu)先級有關(guān)。用戶設(shè)定了weightAdjuster（不知道該起個什么名字），那么job的權(quán)值還與其已經(jīng)提交的時間有關(guān)。設(shè)定有一定的時間，如果提交的時間已經(jīng)超過了這個額度（duration），job的權(quán)重將乘以一個系數(shù)（factor）。

Job權(quán)重計算weight=Math.log1p(runnableTasks)/Math.log(2);weight*=getPriorityFactor(job.getPriority());

if(weightAdjuster!=null){//設(shè)置了權(quán)重改變策略

//Runweightthroughtheuser-suppliedweightAdjusterweight=weightAdjuster.adjustWeight(job,taskType,weight);}

結(jié)束語MapReduce計算模型已經(jīng)被越來越多的企業(yè)的支持和應(yīng)用。由于MapReduce集群的昂