Hbase副本數(shù)與hdfs副本數(shù)
hbase的副本與hdfs的副本之間沒有任何關(guān)系。
hbase的副本是對歷史數(shù)據(jù)的備份,是新舊數(shù)據(jù),而非相同數(shù)據(jù)的copy;hdfs的副本是同一數(shù)據(jù)的copy。
hbase的數(shù)據(jù)文件存放在hdfs上,但是在hdfs上具體如何存儲對hbase是透明的。hbase的hfile
文件是存儲在hdfs的普通文件,這個文件同hdfs上的其他文件一樣,會自動備份成多份,并由namenode管理,而hbase只是通過namenode去訪問hfile,具體訪問的文件位置對hbase是透明的。
所以你的問題的答案是:只跟hdfs的副本數(shù)有關(guān),跟hbase的副本數(shù)無關(guān)。即3
hadoop有哪些優(yōu)缺點
一、HDFS缺點:
1、不能做到低延遲:由于hadoop針對高數(shù)據(jù)吞吐量做了優(yōu)化,犧牲了獲取數(shù)據(jù)的延遲,所以對于低延遲數(shù)據(jù)訪問,不適合hadoop,對于低延遲的訪問需求,HBase是更好的選擇,
2、不適合大量的小文件存儲:由于namenode將文件系統(tǒng)的元數(shù)據(jù)存儲在內(nèi)存中,因此該文件系統(tǒng)所能存儲的文件總數(shù)受限于namenode的內(nèi)存容量,根據(jù)經(jīng)驗,每個文件、目錄和數(shù)據(jù)塊的存儲信息大約占150字節(jié)。
3、不適合多用戶寫入文件,修改文件:Hadoop2.0雖然支持文件的追加功能,但是還是不建議對HDFS上的文件進(jìn)行修改,因為效率低。
4、對于上傳到HDFS上的文件,不支持修改文件,HDFS適合一次寫入,多次讀取的場景。
5、HDFS不支持多用戶同時執(zhí)行寫操作,即同一時間,只能有一個用戶執(zhí)行寫操作。
二、HDFS優(yōu)點:
1、高可靠性。Hadoop按位存儲和處理數(shù)據(jù)的能力值得人們信賴。
2、高擴(kuò)展性。Hadoop是在可用的計算機(jī)集簇間分配數(shù)據(jù)并完成計算任務(wù)的,這些集簇可以方便地擴(kuò)展到數(shù)以千計的節(jié)點中。
3、高效性。Hadoop能夠在節(jié)點之間動態(tài)地移動數(shù)據(jù),并保證各個節(jié)點的動態(tài)平衡,因此處理速度非常快。
4、高容錯性。Hadoop能夠自動保存數(shù)據(jù)的多個副本,并且能夠自動將失敗的任務(wù)重新分配。
5、低成本。與一體機(jī)、商用數(shù)據(jù)倉庫以及QlikView、Yonghong Z-Suite等數(shù)據(jù)集市相比,hadoop是開源的,項目的軟件成本因此會大大降低。
6、Hadoop帶有用Java語言編寫的框架,因此運(yùn)行在 Linux生產(chǎn)平臺上是非常理想的。Hadoop上的應(yīng)用程序也可以使用其他語言編寫,比如 C++。
hdfs詳解之塊、小文件和副本數(shù)
1、block:block是物理切塊,在文件上傳到HDFS文件系統(tǒng)后,對大文件將以每128MB的大小切分若干,存放在不同的DataNode上。例如一個文件130M,那么他會存被切分成2個塊,一個塊128M,另一個塊2M.
1、HDFS適應(yīng)場景:大文件存儲,小文件是致命的
2、如果小文件很多的,則有可能將NN(4G=42億字節(jié))撐爆。例如:1個小文件(閾值<=30M),那么NN節(jié)點維護(hù)的字節(jié)大約250字節(jié)。一億個小文件則是250b* 1億=250億.將會把NN節(jié)點撐爆。如果一億個小文件合并成100萬個大文件:250b* 1百萬=2億字節(jié)。
3、在生產(chǎn)上一般會:
1)調(diào)整小文件閾值
2)合并小文件:
a.數(shù)據(jù)未落地到hdfs之前合并
b.數(shù)據(jù)已經(jīng)落到hdfs,調(diào)用spark service服務(wù)。每天調(diào)度去合并(-15天業(yè)務(wù)周期)
3)小文件的危害:
a.撐爆NN。
b.影響hive、spark的計算。占用集群計算資源
1、如果是偽分布式,那么副本數(shù)只能為一。
2、生成上副本數(shù)一般也是官方默認(rèn)參數(shù): 3份
如果一個文件130M,副本數(shù)為3。那么第一個block128M,有三份。另外一個block2M,也有三份。
題目:
blockSize128M,副本數(shù)3份,那么一個文件260M,請問多少塊,多少實際存儲?
260%128=2....4M 3個塊 3個副本=9塊
260M 3=780M
hadoop中存儲文件系統(tǒng)hdfs的冗余機(jī)制是怎么進(jìn)行的有什么特點
可以只用一行代碼來運(yùn)行MapReduce作業(yè):JobClient.runJon(conf),Job作業(yè)運(yùn)行時參與的四個實體:
1.JobClient寫代碼,配置作業(yè),提交作業(yè)。
2.JobTracker:初始化作業(yè),分配作業(yè),協(xié)調(diào)作業(yè)運(yùn)行。這是一個java程序,主類是JobTracker。
3.TaskTracker:運(yùn)行作業(yè)劃分后的任務(wù),即分配數(shù)據(jù)分配上執(zhí)行Map或Reduce任務(wù)。
4.HDFS:保存作業(yè)數(shù)據(jù)、配置信息等,保存作業(yè)結(jié)果。
Map/Reduce作業(yè)總體執(zhí)行流程:
代碼編寫---->作業(yè)配置---->作業(yè)提交---->Map任務(wù)分配和執(zhí)行---->處理中間結(jié)果----> Reduce任務(wù)分配與執(zhí)行---->輸出結(jié)果
而對于每個作業(yè)的執(zhí)行,又包含:
輸入準(zhǔn)備---->任務(wù)執(zhí)行---->輸出結(jié)果
作業(yè)提交JobClient:
JobClient的runJob方法產(chǎn)生一個Jobclient實例并調(diào)用其submitJob方法,然后runJob開始循環(huán)嗎,并在循環(huán)中調(diào)用getTaskCompetionEvents方法,獲得TaskCompletionEvent實例,每秒輪詢作業(yè)進(jìn)度(后面有介紹進(jìn)度和狀態(tài)更新),把進(jìn)度寫到控制臺,作業(yè)完成后顯示作業(yè)計數(shù)器,若失敗,則把錯誤記錄到控制臺。
submitJob方法作業(yè)提交的過程:
1.向JobTracker請求一個新的JobId。
2.檢查作業(yè)相關(guān)路徑,如果路徑不正確就會返回錯誤。
3.計算作業(yè)輸入分片及其劃分信息。
4.將作業(yè)運(yùn)行需要的資源(jar文件、配置文件等)復(fù)制到Shared HDFS,并
復(fù)制多個副本(參數(shù)控制,默認(rèn)值為10)供tasktracker訪問,也會將計算的分片復(fù)制到HDFS。
5.調(diào)用JobTracker對象的submitJob()方法來真正提交作業(yè),告訴JobTracker作業(yè)準(zhǔn)備執(zhí)行。
作業(yè)的初始化JobTracker:
JobTracker收到submitJob方法調(diào)用后,會把調(diào)用放入到一個內(nèi)部隊列,由作業(yè)調(diào)度器(Job scheduler)進(jìn)行調(diào)度并對其初始化。Job初始化即創(chuàng)建一個作業(yè)對象。
當(dāng)作業(yè)被調(diào)度后,JobTracker會創(chuàng)建一個代表這個作業(yè)的JobInProgress對象,并將任務(wù)和記錄信息封裝在這個對象中,以便跟蹤任務(wù)狀態(tài)和進(jìn)程。
初始化過程就是JobInProgress對象的initTasks方法進(jìn)行初始化的。
初始化步驟:
1.從HDFS中讀取作業(yè)對應(yīng)的job.split信息,為后面的初始化做好準(zhǔn)備。
2.創(chuàng)建并初始化map和reduce任務(wù)。根據(jù)數(shù)據(jù)分片信息中的個數(shù)確定map task的個數(shù),然后為每個map task生成一個TaskInProgress對象來處理數(shù)據(jù)分片,先將其放入nonRunningMapCache,以便JobTracker分配任務(wù)的時候使用。接下來根據(jù)JobConf中的mapred.reduce.tasks屬性利用setNumReduceTasks()方法設(shè)置reduce task的數(shù)量,然后同map task創(chuàng)建方式。
3.最后就是創(chuàng)建兩個初始化task,進(jìn)行map和reduce的初始化。
任務(wù)的分配JobTracker:
消息傳遞HeartBeat: tasktracker運(yùn)行一個簡單循環(huán)定期發(fā)送心跳(heartbeat)給JobTracker。由心跳告知JobTracker自己是否存活,同時作為消息通道傳遞其它信息(請求新task)。作為心跳的一部分,tasktracker會指明自己是否已準(zhǔn)備好運(yùn)行新的任務(wù),如果是,jobtracker會分配它一個任務(wù)。
分配任務(wù)所屬于的作業(yè):在Jobtracker分配任務(wù)前需先確定任務(wù)所在的作業(yè)。后面會介紹到各種作業(yè)調(diào)度算法,默認(rèn)是一個FIFO的作業(yè)調(diào)度。
分配Map和Reduce任務(wù):tasktracker有固定數(shù)量的任務(wù)槽,一個tasktracker可以同時運(yùn)行多個Map和Reduce任務(wù),但其準(zhǔn)確的數(shù)量由tasktracker的核的數(shù)量和內(nèi)存大小決定。默認(rèn)調(diào)度器會先填滿Map任務(wù)槽,再填Reduce任務(wù)槽。jobtracker會選擇距離離分片文件最近的tasktracker,最理想情況下,任務(wù)是數(shù)據(jù)本地化(data-local)的,當(dāng)然也可以是機(jī)架本地化(rack-local),如果不是本地化的,那么他們就需要從其他機(jī)架上檢索數(shù)據(jù)。Reduce任務(wù)分配很簡單,jobtracker會簡單的從待運(yùn)行的reduce任務(wù)列表中選取下一個來執(zhí)行,不用考慮數(shù)據(jù)本地化。
任務(wù)的執(zhí)行TaskTracker:
TaskTracker收到新任務(wù)后,就要在本地運(yùn)行任務(wù)了,運(yùn)行任務(wù)的第一步就是通過localizedJob將任務(wù)本地化所需要的注入配置、數(shù)據(jù)、程序等信息進(jìn)行本地化。
1.本地化數(shù)據(jù):從共享文件系統(tǒng)將job.split、job.jar(在分布式緩存中)復(fù)制本地,將job配置信息寫入job.xml。
2.新建本地工作目錄:tasktracker會加壓job.jar文件到本工作目錄。
3.調(diào)用launchTaskForJob方法發(fā)布任務(wù)(其中會新建TaskRunner實例運(yùn)行任務(wù)),如果是Map任務(wù)就啟用MapTaskRunner,對于Reduce就是ReduceTaskRunner。
在這之后,TaskRunner會啟用一個新的JVM來運(yùn)行每個Map/Reduce任務(wù),防止程序原因而導(dǎo)致tasktracker崩潰,但不同任務(wù)間重用JVM還是可以的,后續(xù)會講到任務(wù)JVM重用。
對于單個Map,任務(wù)執(zhí)行的簡單流程是:
1.分配任務(wù)執(zhí)行參數(shù)
2.在Child臨時文件中添加map任務(wù)信息(Child是運(yùn)行Map和Reduce任務(wù)的主進(jìn)程)
3.配置log文件夾,配置map任務(wù)的通信和輸出參數(shù)
4.讀取input split,生成RecordReader讀取數(shù)據(jù)
5.為Map生成MapRunnable,依次從RecordReader中接收數(shù)據(jù),并調(diào)用Map函數(shù)進(jìn)行處理。
6.最后將map函數(shù)的輸出調(diào)用collect收集到MapOutputBuffer(參數(shù)控制其大小)中。
Streaming和Pipes:
Streaming和Pipes都運(yùn)行特殊的Map和Reduce任務(wù),目的是運(yùn)行用戶提供的可執(zhí)行程序并與之通信。
Streaming:使用標(biāo)準(zhǔn)輸入輸出Streaming與進(jìn)程進(jìn)行通信。
Pipes:用來監(jiān)聽套接字,會發(fā)送一個端口號給C++程序,兩者便可建立鏈接。
進(jìn)度和狀態(tài)更新:
一個作業(yè)和它的任務(wù)都有狀態(tài)(status),其中包括:運(yùn)行成功失敗狀態(tài)、Map/Reduce進(jìn)度、作業(yè)計數(shù)器值、狀態(tài)消息。
狀態(tài)消息與客戶端的通信:
1.對于Map任務(wù)Progress的追蹤:progress是已經(jīng)處理完的輸入所占的比例。
2.對于Reduce:稍復(fù)雜,reduce任務(wù)分三個階段(每個階段占1/3),復(fù)制、排序和Reduce處理,若reduce已執(zhí)行一半的輸入的話,那么任務(wù)進(jìn)度便是1/3+1/3+1/6=5/6。
3.任務(wù)計數(shù)器:任務(wù)有一組計數(shù)器,負(fù)責(zé)對任務(wù)運(yùn)行各個事件進(jìn)行計數(shù)。
4.任務(wù)進(jìn)度報告:如果任務(wù)報告了進(jìn)度,便會設(shè)置一個標(biāo)記以表明狀態(tài)將被發(fā)送到tasktracker。有一個獨立線程每隔三秒檢查一次此標(biāo)記,如果已設(shè)置,則告知tasktracker當(dāng)前狀態(tài)。
5.tasktracker進(jìn)度報告:tasktracker會每隔5秒(這個心跳是由集群大小決定,集群越大時間會越長)發(fā)送heartbeat到j(luò)obtracker,并且tasktracker運(yùn)行的所有狀態(tài)都會在調(diào)用中被發(fā)送到j(luò)obtracker。
6.jobtracker合并各任務(wù)報告:產(chǎn)生一個表明所有運(yùn)行作業(yè)機(jī)器所含任務(wù)狀態(tài)的全局視圖。
前面提到的JobClient就是通過每秒查詢JobTracker來接收最新狀態(tài),而且客戶端JobClient的getJob方法可以得到一個RunningJob的實例,其包含了作業(yè)的所以狀態(tài)信息。
作業(yè)的完成:
當(dāng)jobtracker收到作業(yè)最后一個任務(wù)已完成的通知后,便把作業(yè)狀態(tài)設(shè)置成成功。JobClient查詢狀態(tài)時,便知道任務(wù)已成功完成,于是JobClient打印一條消息告知用戶,然后從runJob方法返回。
如果jobtracker有相應(yīng)設(shè)置,也會發(fā)送一個Http作業(yè)通知給客戶端,希望收到回調(diào)指令的客戶端可以通過job.end.notification.url屬性來進(jìn)行設(shè)置。
jobtracker情況作業(yè)的工作狀態(tài),指示tasktracker也清空作業(yè)的工作狀態(tài),如刪除中間輸出。
失敗
實際情況下,用戶的代碼存在軟件錯誤進(jìn)程會崩潰,機(jī)器也會產(chǎn)生故障,但Hadoop能很好的應(yīng)對這些故障并完成作業(yè)。
1.任務(wù)失敗
子任務(wù)異常:如Map/Reduce任務(wù)中的用戶代碼拋出異常,子任務(wù)JVM進(jìn)程會在退出前向父進(jìn)程tasktracker發(fā)送錯誤報告,錯誤被記錄用戶日志。tasktracker會將此次task attempt標(biāo)記為tailed,并釋放這個任務(wù)槽運(yùn)行另外一個任務(wù)。
子進(jìn)程JVM突然退出:可能由于JVM bug導(dǎo)致用戶代碼造成的某些特殊原因?qū)е翵VM退出,這種情況下,tasktracker會注意到進(jìn)程已經(jīng)退出,并將此次嘗試標(biāo)記為failed。
任務(wù)掛起:一旦tasktracker注意一段時間沒有收到進(jìn)度更新,便會將任務(wù)標(biāo)記為failed,JVM子進(jìn)程將被自動殺死。任務(wù)失敗間隔時間通常為10分鐘,可以以作業(yè)或者集群為基礎(chǔ)設(shè)置過期時間,參數(shù)為mapred.task.timeout。注意:如果參數(shù)值設(shè)置為0,則掛起的任務(wù)永遠(yuǎn)不會釋放掉它的任務(wù)槽,隨著時間的推移會降低整個集群的效率。
任務(wù)失敗嘗試次數(shù):jobtracker得知一個tasktracker失敗后,它會重新調(diào)度該任務(wù)執(zhí)行,jobtracker會嘗試避免重新調(diào)度失敗過的tasktracker任務(wù)。如果一個任務(wù)嘗試次數(shù)超過4次,它將不再被重試。這個值是可以設(shè)置的,對于Map任務(wù),參數(shù)是mapred.map.max.attempts,對于reduce任務(wù),則由mapred.reduce.max.attempts屬性控制。如果次數(shù)超過限制,整個作業(yè)都會失敗。有時我們不希望少數(shù)幾個任務(wù)失敗就終止運(yùn)行的整個作業(yè),因為即使有些任務(wù)失敗,作業(yè)的一些結(jié)果可能還是有用的,這種情況下,可以為作業(yè)設(shè)置在不觸發(fā)作業(yè)失敗情況下的允許任務(wù)失敗的最大百分比,Map任務(wù)和Reduce任務(wù)可以獨立控制,參數(shù)為mapred.max.map.failures.percent和mapred.max.reduce.failures.percent。
任務(wù)嘗試中止(kill):任務(wù)終止和任務(wù)失敗不同,task attempt可以中止是因為他是一個推測副本或因為它所處的tasktracker失敗,導(dǎo)致jobtracker將它上面的所有task attempt標(biāo)記為killed。被終止的task attempt不會被計入任務(wù)運(yùn)行嘗試次數(shù),因為嘗試中止并不是任務(wù)的錯。
2.tasktracker失敗
tasktracker由于崩潰或者運(yùn)行過慢而失敗,他將停止向jobtracker發(fā)送心跳(或很少發(fā)送心跳)。jobtracker注意已停止發(fā)送心跳的tasktracker(過期時間由參數(shù)mapred.tasktracker.expiry.interval設(shè)置,單位毫秒),并將它從等待調(diào)度的tasktracker池中移除。如果是未完成的作業(yè),jobtracker會安排次tasktracker上已經(jīng)運(yùn)行成功的Map任務(wù)重新運(yùn)行,因為此時reduce任務(wù)已無法訪問(中間輸出存放在失敗的tasktracker的本地文件系統(tǒng)上)。
即使tasktracker沒有失敗,也有可能被jobtracker列入黑名單。如果tasktracker上面的失敗任務(wù)數(shù)量遠(yuǎn)遠(yuǎn)高于集群的平均失敗任務(wù)次數(shù),他就會被列入黑名單,被列入黑名單的tasktracker可以通過重啟從jobtracker黑名單中移除。
3.jobtracker失敗
老版本的JobTracker失敗屬于單點故障,這種情況下作業(yè)注定失敗。
作業(yè)調(diào)度:
早期作業(yè)調(diào)度FIFO:按作業(yè)提交順序先進(jìn)先出。可以設(shè)置優(yōu)先級,通過設(shè)置mapred.job.priority屬性或者JobClient的setJobPriority()方法制定優(yōu)先級(優(yōu)先級別:VERY_HIGH,HIGH,NORMAL,LOW,VERY_LOW)。注意FIFO調(diào)度算法不支持搶占(preemption),所以高優(yōu)先級作業(yè)仍然會被那些已經(jīng)開始的長時間運(yùn)行的低優(yōu)先級作業(yè)所阻塞。
Fair Scheduler:目標(biāo)是讓每個用戶公平地共享集群能力。當(dāng)集群存在很多作業(yè)時,空閑的任務(wù)槽會以”讓每個用戶共享集群“的方式進(jìn)行分配。默認(rèn)每個用戶都有自己的作業(yè)池。FairScheduler支持搶占,如果一個池在特定的一段時間未得到公平地資源共享,它會終止池中得到過多的資源任務(wù),以便把任務(wù)槽讓給資源不足的池。FairScheduler是一個后續(xù)模塊,使用它需要將其jar文件放在Hadoop的類路徑下。可以通過參數(shù)map.red.jobtracker.taskScheduler屬性配置(值為org.apache.hadoop.mapred.FairScheduler)
Capacity Scheduler:
集群由很多隊列組成,每個隊列都有一個分配能力,這一點與FairScheduler類似,只不過在每個隊列內(nèi)部,作業(yè)根據(jù)FIFO方式進(jìn)行調(diào)度。本質(zhì)上說,Capacity Scheduler允許用戶或組織為每個用戶模擬一個獨立使用FIFO的集群。
shuffle和排序:
MapReduce確保每個Reducer的輸入都是按鍵排序的。系統(tǒng)執(zhí)行排序的過程-將map輸出作為輸入傳給reducer的過程稱為shuffle。shuffle屬于不斷被優(yōu)化和改進(jìn)的代碼庫的一部分,從許多方面來看,shuffle是MapReduce的心臟。
整個shuffle的流程應(yīng)該是這樣:
map結(jié)果劃分partition排序sort分割spill合并同一劃分合并同一劃分合并結(jié)果排序 reduce處理輸出
Map端:
寫入緩沖區(qū):Map函數(shù)的輸出,是由collector處理的,它并不是簡單的將結(jié)果寫到磁盤。它利用緩沖的方式寫到內(nèi)存,并處于效率的考慮進(jìn)行預(yù)排序。每個map都有一個環(huán)形的內(nèi)存緩沖區(qū),用于任務(wù)輸出,默認(rèn)緩沖區(qū)大小為100MB(由參數(shù)io.sort.mb調(diào)整),一旦緩沖區(qū)內(nèi)容達(dá)到閾值(默認(rèn)0.8),后臺進(jìn)程邊開始把內(nèi)容寫到磁盤(spill),在寫磁盤過程中,map輸出繼續(xù)被寫到緩沖區(qū),但如果緩沖區(qū)被填滿,map會阻塞知道寫磁盤過程完成。寫磁盤將按照輪詢方式寫到mapred.local.dir屬性制定的作業(yè)特定子目錄中。
寫出緩沖區(qū):collect將緩沖區(qū)的內(nèi)容寫出時,會調(diào)用sortAndSpill函數(shù),這個函數(shù)作用主要是創(chuàng)建spill文件,按照key值對數(shù)據(jù)進(jìn)行排序,按照劃分將數(shù)據(jù)寫入文件,如果配置了combiner類,會先調(diào)用combineAndSpill函數(shù)再寫文件。sortAndSpill每被調(diào)用一次,就會寫一個spill文件。
合并所有Map的spill文件:TaskTracker會在每個map任務(wù)結(jié)束后對所有map產(chǎn)生的spill文件進(jìn)行merge,merge規(guī)則是根據(jù)分區(qū)將各個spill文件中數(shù)據(jù)同一分區(qū)中的數(shù)據(jù)合并在一起,并寫入到一個已分區(qū)且排序的map輸出文件中。待唯一的已分區(qū)且已排序的map輸出文件寫入最后一條記錄后,map端的shuffle階段就結(jié)束了。
在寫磁盤前,線程首先根據(jù)數(shù)據(jù)最終要傳遞到的reducer把數(shù)據(jù)劃分成響應(yīng)的分區(qū)(partition),在每個分區(qū)中,后臺線程按鍵進(jìn)行內(nèi)排序,如果有一個combiner,它會在排序后的輸出上運(yùn)行。
內(nèi)存達(dá)到溢出寫的閾值時,就會新建一個溢出寫文件,因為map任務(wù)完成其最后一個輸出記錄之后,會有幾個溢出寫文件。在任務(wù)完成前,溢出寫文件會被合并成一個已分區(qū)且已排序的輸出文件。配置屬性io.sort.facor控制一次最多能合并多少流,默認(rèn)值是10。
如果已經(jīng)指定combiner,并且寫次數(shù)至少為3(通過min.mum.spills.for.combine設(shè)置)時,則combiner就會在輸出文件寫到磁盤之前運(yùn)行。運(yùn)行combiner的意義在于使map輸出更緊湊,舍得寫到本地磁盤和傳給reducer的數(shù)據(jù)更少。
寫磁盤時壓縮:寫磁盤時壓縮會讓寫的速度更快,節(jié)約磁盤空間,并且減少傳給reducer的數(shù)據(jù)量。默認(rèn)情況下,輸出是不壓縮的,但可以通過設(shè)置mapred.compress.map.output值為true,就可以啟用壓縮。使用的壓縮庫是由mapred.map.output.compression.codec制定。
reducer獲得文件分區(qū)的工作線程:reducer通過http方式得到輸出文件的分區(qū),用于文件分區(qū)的工作線程數(shù)量由tracker.http.threads屬性指定,此設(shè)置針對的是每個tasktracker,而不是每個map任務(wù)槽。默認(rèn)值為40,在大型集群上此值可以根據(jù)需要而增加。
Reduce端:
復(fù)制階段:reduce會定期向JobTracker獲取map的輸出位置,一旦拿到輸出位置,reduce就會從對應(yīng)的TaskTracker上復(fù)制map輸出到本地(如果map輸出很小,則會被復(fù)制到TaskTracker節(jié)點的內(nèi)存中,否則會被讓如磁盤),而不會等到所有map任務(wù)結(jié)束(當(dāng)然這個也有參數(shù)控制)。
合并階段:從各個TaskTracker上復(fù)制的map輸出文件(無論在磁盤還是內(nèi)存)進(jìn)行整合,并維持?jǐn)?shù)據(jù)原來的順序。
Reduce階段:從合并的文件中順序拿出一條數(shù)據(jù)進(jìn)行reduce函數(shù)處理,然后將結(jié)果輸出到本地HDFS。
Map的輸出文件位于運(yùn)行map任務(wù)的tasktracker的本地磁盤,tasktracker要為分區(qū)文件運(yùn)行reduce任務(wù)。每個任務(wù)完成時間可能不同,但是只要有一個任務(wù)完成,reduce任務(wù)就開始復(fù)制其輸出,這就是reduce任務(wù)的復(fù)制階段(copy phase)。reduce任務(wù)有少量復(fù)制線程,因此能夠并行取得map輸出。默認(rèn)值是5個線程,可以通過mapred.reduce.parallel.copies屬性設(shè)置。
Reducer如何得知從哪個tasktracker獲得map輸出:map任務(wù)完成后會通知其父tasktracker狀態(tài)已更新,tasktracker進(jìn)而通知(通過heart beat)jobtracker。JobTracker就知道m(xù)ap輸出和tasktracker之間的映射關(guān)系,reducer中的一個線程定期詢問jobtracker以便獲知map輸出位置。由于reducer有可能失敗,因此tasktracker并沒有在第一個reducer檢索到map輸出時就立即從磁盤上刪除它們,相反他會等待jobtracker告示它可以刪除map輸出時才刪除,這是作業(yè)完成后最后執(zhí)行的。
如果map輸出很小,則會被直接復(fù)制到reduce tasktracker的內(nèi)存緩沖區(qū)(大小由mapred.job.shuffle.input.buffer.percent控制,占堆空間的百分比),否則,map輸出被復(fù)制到磁盤。一旦內(nèi)存緩沖區(qū)達(dá)到閾值大小(由mapred.iob.shuffle.merge.percent)
或達(dá)到map輸出閾值大小(mapred.inmem.threadhold),則合并后溢出寫到磁盤中。
隨著磁盤上副本增多,后臺線程會將他們合并為更大的、排好序的文件。注意:為了合并,壓縮的map輸出必須在內(nèi)存中被解壓縮。
排序階段:復(fù)制階段完成后,reduce任務(wù)會進(jìn)入排序階段,更確切的說是合并階段,這個階段將合并map輸出,維持其順序排列。合并是循環(huán)進(jìn)行的,由合并因子決定每次合并的輸出文件數(shù)量。但讓有可能會產(chǎn)生中間文件。
reduce階段:在最后reduce階段,會直接把排序好的文件輸入reduce函數(shù),不會對中間文件進(jìn)行再合并,最后的合并即可來自內(nèi)存,也可來自磁盤。此階段的輸出會直接寫到文件系統(tǒng),一般為hdfs。
細(xì)節(jié):這里合并是并非平均合并,比如有40個文件,合并因子為10,我們并不是每趟合并10個,合并四趟。而是第一趟合并4個,后三趟合并10,在最后一趟中4個已合并的文件和余下6個未合并會直接并入reduce。