Hadoop中MapReducer的工作過程

本篇內(nèi)容介紹了“Hadoop中MapReducer的工作過程”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

1. 從輸入到輸出

一個(gè)MapReducer作業(yè)經(jīng)過了input，map，combine，reduce，output五個(gè)階段，其中combine階段并不一定發(fā)生，map輸出的中間結(jié)果被分到reduce的過程成為shuffle（數(shù)據(jù)清洗）。

在shuffle階段還會(huì)發(fā)生copy（復(fù)制）和sort（排序）。

在MapReduce的過程中，一個(gè)作業(yè)被分成Map和Reducer兩個(gè)計(jì)算階段，它們由一個(gè)或者多個(gè)Map任務(wù)和Reduce任務(wù)組成。如下圖所示，一個(gè)MapReduce作業(yè)從數(shù)據(jù)的流向可以分為Map任務(wù)和Reduce任務(wù)。當(dāng)用戶向Hadoop提交一個(gè)MapReduce作業(yè)時(shí)，JobTracker則會(huì)根據(jù)各個(gè)TaskTracker周期性發(fā)送過來(lái)的心跳信息綜合考慮TaskTracker的資源剩余量，作業(yè)優(yōu)先級(jí)，作業(yè)提交時(shí)間等因素，為TaskTracker分配合適的任務(wù)。Reduce任務(wù)默認(rèn)會(huì)在Map任務(wù)數(shù)量完成5%后才開始啟動(dòng)。

Map任務(wù)的執(zhí)行過程可以概括為：首先通過用戶指定的InputFormat類中的getSplits方法和next方法將輸入文件切片并解析成鍵值對(duì)作為map函數(shù)的輸入。然后map函數(shù)經(jīng)過處理之后將中間結(jié)果交給指定的Partitioner處理，確保中間結(jié)果分發(fā)到指定的Reduce任務(wù)處理，此時(shí)如果用戶指定了Combiner，將執(zhí)行combine操作。最后map函數(shù)將中間結(jié)果保存到本地。

Reduce任務(wù)的執(zhí)行過程可以概括為：首先需要將已經(jīng)完成Map任務(wù)的中間結(jié)果復(fù)制到Reduce任務(wù)所在的節(jié)點(diǎn)，待數(shù)據(jù)復(fù)制完成后，再以key進(jìn)行排序，通過排序，將所有key相同的數(shù)據(jù)交給reduce函數(shù)處理，處理完成后，結(jié)果直接輸出到HDFS上。

2. input

如果使用HDFS上的文件作為MapReduce的輸入，MapReduce計(jì)算框架首先會(huì)用org.apache.hadoop.mapreduce.InputFomat類的子類FileInputFormat類將作為輸入HDFS上的文件切分形成輸入分片(InputSplit)，每個(gè)InputSplit將作為一個(gè)Map任務(wù)的輸入，再將InputSplit解析為鍵值對(duì)。InputSplit的大小和數(shù)量對(duì)于MaoReduce作業(yè)的性能有非常大的影響。

InputSplit只是邏輯上對(duì)輸入數(shù)據(jù)進(jìn)行分片，并不會(huì)將文件在磁盤上分成分片進(jìn)行存儲(chǔ)。InputSplit只是記錄了分片的元數(shù)據(jù)節(jié)點(diǎn)信息，例如起始位置，長(zhǎng)度以及所在的節(jié)點(diǎn)列表等。數(shù)據(jù)切分的算法需要確定InputSplit的個(gè)數(shù)，對(duì)于HDFS上的文件，F(xiàn)ileInputFormat類使用computeSplitSize方法計(jì)算出InputSplit的大小，代碼如下：

1. }

其中 minSize 由mapred-site.xml文件中的配置項(xiàng)mapred.min.split.size決定，默認(rèn)為1；maxSize 由mapred-site.xml文件中的配置項(xiàng)mapred.max.split.size決定，默認(rèn)為9223 372 036 854 775 807；而blockSize是由hdfs-site.xml文件中的配置項(xiàng)dfs.block.size決定，默認(rèn)為67 108 864字節(jié)（64M）。所以InputSplit的大小確定公式為：

1. 
   

2. 
   

3. 壓縮格式	工具	算法	文件擴(kuò)展名	是否包含多個(gè)文件	是否可切分
   DEFLATE*	N/A	DEFLATE	.deflate	否	否
   Gzip	gzip	DEFLATE	.gz	否	否
   bzip2	bzip2	bzip2	.bz2	否	是
   LZO	Lzop	LZO	.lzo	否	否

   map輸出的中間結(jié)果存儲(chǔ)的格式為IFile，IFile是一種支持航壓縮的存儲(chǔ)格式，支持上述壓縮算法。

   Reducer通過Http方式得到輸出文件的分區(qū)。將map輸出的中間結(jié)果發(fā)送到Reducer的工作線程的數(shù)量由mapred-site.xml文件的tasktracker.http.threds配置項(xiàng)決定，此配置針對(duì)每個(gè)節(jié)點(diǎn)，而不是每個(gè)Map任務(wù)，默認(rèn)是40，可以根據(jù)作業(yè)大小，集群規(guī)模以及節(jié)點(diǎn)的計(jì)算能力而增大。

   4. shuffle

   shuffle，也叫數(shù)據(jù)清洗。在某些語(yǔ)境下，代表map函數(shù)產(chǎn)生輸出到reduce的消化輸入的整個(gè)過程。

   4.1 copy階段

   Map任務(wù)輸出的結(jié)果位于Map任務(wù)的TaskTracker所在的節(jié)點(diǎn)的本地磁盤上。TaskTracker需要為這些分區(qū)文件（map輸出）運(yùn)行Reduce任務(wù)。但是，Reduce任務(wù)可能需要多個(gè)Map任務(wù)的輸出作為其特殊的分區(qū)文件。每個(gè)Map任務(wù)的完成時(shí)間可能不同，當(dāng)只要有一個(gè)任務(wù)完成，Reduce任務(wù)就開始復(fù)制其輸出。這就是shuffle的copy階段。如下圖所示，Reduce任務(wù)有少量復(fù)制線程，可以并行取得Map任務(wù)的輸出，默認(rèn)值為5個(gè)線程，該值可以通過設(shè)置mapred-site.xml的mapred.reduce.parallel.copies的配置項(xiàng)來(lái)改變。

   

   如果map輸出相當(dāng)小，則會(huì)被復(fù)制到Reduce所在TaskTracker的內(nèi)存的緩沖區(qū)中，緩沖區(qū)的大小由mapred-site.xml文件中的mapred.job.shuffle.input.buffer.percent配置項(xiàng)指定。否則，map輸出將會(huì)被復(fù)制到磁盤。一旦內(nèi)存緩沖區(qū)達(dá)到閾值大小（由mapred-site.xml文件mapred.job.shuffle.merge.percent配置項(xiàng)決定）或緩沖區(qū)的文件數(shù)達(dá)到閾值大?。ㄓ蒻apred-site.xml文件mapred.inmem.merge.threshold配置項(xiàng)決定），則合并后溢寫到磁盤中。

   4.2 sort階段

   隨著溢寫到磁盤的文件增多，shuffle進(jìn)行sort階段。這個(gè)階段將合并map的輸出文件，并維持其順序排序，其實(shí)做的是歸并排序。排序的過程是循環(huán)進(jìn)行，如果有50個(gè)map的輸出文件，而合并因子（由mapred-site.xml文件的io.sort.factor配置項(xiàng)決定，默認(rèn)為10）為10，合并操作將進(jìn)行5次，每次將10個(gè)文件合并成一個(gè)文件，最后有5個(gè)文件，這5個(gè)文件由于不滿足合并條件（文件數(shù)小于合并因子），則不會(huì)進(jìn)行合并，將會(huì)直接把5個(gè)文件交給Reduce函數(shù)處理。到此shuffle階段完成。

   從shuffle的過程可以看出，Map任務(wù)處理的是一個(gè)InputSplit，而Reduce任務(wù)處理的是所有Map任務(wù)同一個(gè)分區(qū)的中間結(jié)果。

   5. reduce及最后結(jié)果的輸出

   reduce階段操作的實(shí)質(zhì)就是對(duì)經(jīng)過shuffle處理后的文件調(diào)用reduce函數(shù)處理。由于經(jīng)過了shuffle的處理，文件都是按鍵分區(qū)且有序，對(duì)相同分區(qū)的文件調(diào)用一次reduce函數(shù)處理。

   與map的中間結(jié)果不同的是，reduce的輸出一般為HDFS。

   6. sort

   排序貫穿于Map任務(wù)和Reduce任務(wù)，排序操作屬于MapReduce計(jì)算框架的默認(rèn)行為，不管流程是否需要，都會(huì)進(jìn)行排序。在MapReduce計(jì)算框架中，主要用到了兩種排序算法：快速排序和歸并排序。

   在Map任務(wù)和Reduce任務(wù)的過程中，一共發(fā)生了3次排序操作。

   （1）當(dāng)map函數(shù)產(chǎn)生輸出時(shí)，會(huì)首先寫入內(nèi)存的環(huán)形緩沖區(qū)，當(dāng)達(dá)到設(shè)定的閾值，在刷寫磁盤之前，后臺(tái)線程會(huì)將緩沖區(qū)的數(shù)據(jù)劃分相應(yīng)的分區(qū)。在每個(gè)分區(qū)中，后臺(tái)線程按鍵進(jìn)行內(nèi)排序。如下圖所示。

   （2）在Map任務(wù)完成之前，磁盤上存在多個(gè)已經(jīng)分好區(qū)，并排好序，大小和緩沖區(qū)一樣的溢寫文件，這時(shí)溢寫文件將被合并成一個(gè)已分區(qū)且已排序的輸出文件。由于溢寫文件已經(jīng)經(jīng)過一次排序，所以合并文件時(shí)只需再做一次排序就可使輸出文件整體有序。如下圖所示。

   

   （3）在shuffle階段，需要將多個(gè)Map任務(wù)的輸出文件合并，由于經(jīng)過第二次排序，所以合并文件時(shí)只需在做一次排序就可以使輸出文件整體有序。

1. 在這3次排序中第一次是在內(nèi)存緩沖區(qū)做的內(nèi)排序，使用的算法是快速排序；第二次排序和第三次排序都是在文件合并階段發(fā)生的，使用的是歸并排序。

   7. 作業(yè)的進(jìn)度組成

   一個(gè)MapReduce作業(yè)在Hadoop上運(yùn)行時(shí)，客戶端的屏幕通常會(huì)打印作業(yè)日志，如下：

   

   對(duì)于一個(gè)大型的MapReduce作業(yè)來(lái)說，執(zhí)行時(shí)間可能會(huì)比較比較長(zhǎng)，通過日志了解作業(yè)的運(yùn)行狀態(tài)和作業(yè)進(jìn)度是非常重要的。對(duì)于Map來(lái)說，進(jìn)度代表實(shí)際處理輸入所占比例，例如 map 60% reduce 0% 表示Map任務(wù)已經(jīng)處理了作業(yè)輸入文件的60%，而Reduce任務(wù)還沒有開始。而對(duì)于Reduce的進(jìn)度來(lái)說，情況比較復(fù)雜，從前面得知，reduce階段分為copy，sort 和 reduce，這三個(gè)步驟共同組成了reduce的進(jìn)度，各占1/3。如果reduce已經(jīng)處理了2/3的輸入，那么整個(gè)reduce的進(jìn)度應(yīng)該為1/3 + 1/3 + 1/3 * (2/3) = 5/9 ，因?yàn)閞educe開始處理時(shí)，copy和sort已經(jīng)完成。

“Hadoop中MapReducer的工作過程”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！