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

MapReduce模型MapReduce是由Google發(fā)明的一種處理大規(guī)模數(shù)據(jù)的分布式編程框架，最初是由Google的工程師設(shè)計(jì)并實(shí)現(xiàn)。Hadoop是MapReduee計(jì)算模型的一種開源實(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)情況下，每一個(gè)用戶擁有一個(gè)獨(dú)立的資源池，以使每個(gè)用戶都能獲得一份等同的集群資源而不管他們提交了多少作業(yè)。在每一個(gè)資源池內(nèi)，會(huì)使用公平共享（fairsharing）的方法在運(yùn)行作業(yè)之間共享容量（capacity）。用戶也可以給予資源池相應(yīng)的權(quán)重，以不按比例的方式共享集群。

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

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

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

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

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

FairSchedulerFairScheduler是hadoop中的作業(yè)公平調(diào)度器，主要是解決當(dāng)TT發(fā)送心跳告訴JT當(dāng)前的空閑slots時(shí)，希望JT分配給TT相應(yīng)多個(gè)task，讓TT去執(zhí)行這些task。所以JT就需要一個(gè)調(diào)度器來對(duì)作業(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()方法請(qǐng)求心跳回應(yīng)。JobTracker每次會(huì)對(duì)收到的heartbeat進(jìn)行判斷，如果心跳是請(qǐng)求新的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)對(duì)象（TaskTrackerStatus）來獲得該TaskTracker可執(zhí)行的最大任務(wù)數(shù)。

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

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

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

Pool的排序根據(jù)FairScheduler的規(guī)則對(duì)各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兩者有一個(gè)相同，則依據(jù)pool的startTime。

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

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

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

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

Job權(quán)重計(jì)算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ì)算模型已經(jīng)被越來越多的企業(yè)的支持和應(yīng)用。由于MapReduce集群的昂