MapReduce快速入門系列(11) | MapTask，ReduceTask以及MapReduce運行機制詳解

前面我們講解了MapReduce的Shuffle機制，那么這篇文章博主繼續為大家講解MapTask,ReduceTask和MapReduce運行機制。

目錄

一. MapTask運行機制詳解以及Map任務的并行度

二. ReduceTask 工作機制以及reduceTask的并行度

三. MapReduceshuffle過程

四. MapReduce總體工作機制

一. MapTask運行機制詳解以及Map任務的并行度

整個Map階段流程大體如上圖所示。簡單概述：inputFile通過split被邏輯切分為多個split文件，通過Record按行讀取內容給map（用戶自己實現的）進行處理，數據被map處理結束之后交給OutputCollector收集器，對其結果key進行分區（默認使用hash分區），然后寫入buffer，每個map task都有一個內存緩沖區，存儲著map的輸出結果，當緩沖區快滿的時候需要將緩沖區的數據以一個臨時文件的方式存放到磁盤，當整個map task結束后再對磁盤中這個map task產生的所有臨時文件做合并，生成最終的正式輸出文件，然后等待reduce task來拉數據。

詳細步驟

：

1、首先，

讀取數據組件InputFormat

（默認TextInputFormat）會通過getSplits方法對輸入目錄中文件進行邏輯切片規劃得到splits，有多少個split就對應啟動多少個MapTask。默認情況下split與block的對應關系默認是一對一。

2、將輸入文件切分為splits之后，由

RecordReader對象（默認LineRecordReader）進行讀取

，以\n作為分隔符，讀取一行數據，返回。Key表示每行首字符偏移值，value表示這一行文本內容。

3、讀取split返回，進入用戶自己繼承的Mapper類中，

執行用戶重寫的map函數

。RecordReader讀取一行用戶重寫的map調用一次，并輸出一個。

4、Map輸出的數據會寫入內存，內存中這片區域叫做環形緩沖區，緩沖區的作用是批量收集map結果，減少磁盤IO的影響。key/value對以及Partition的結果都會被寫入緩沖區。當然寫入之前，key與value值都會被序列化成字節數組。

環形緩沖區其實是一個數組，數組中存放著key、value的序列化數據和key、value的元數據信息，包括partition、key的起始位置、value的起始位置以及value的長度。環形結構是一個抽象概念。

緩沖區是有大小限制，默認是100MB。當map task的輸出結果很多時，就可能會撐爆內存，所以需要在一定條件下將緩沖區中的數據臨時寫入磁盤，然后重新利用這塊緩沖區。這個從內存往磁盤寫數據的過程被稱為Spill，中文可譯為溢寫。這個溢寫是由單獨線程來完成，不影響往緩沖區寫map結果的線程。溢寫線程啟動時不應該阻止map的結果輸出，所以整個緩沖區有個溢寫的比例spill.percent。這個比例默認是0.8，也就是當緩沖區的數據已經達到閾值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢寫線程啟動，鎖定這80MB的內存，執行溢寫過程。Map task的輸出結果還可以往剩下的20MB內存中寫，互不影響。

5、

合并溢寫文件

：每次溢寫會在磁盤上生成一個臨時文件（寫之前判斷是否有combiner），如果map的輸出結果真的很大，有多次這樣的溢寫發生，磁盤上相應的就會有多個臨時文件存在。當整個數據處理結束之后開始對磁盤中的臨時文件進行merge合并，因為最終的文件只有一個，寫入磁盤，并且為這個文件提供了一個索引文件，以記錄每個reduce對應數據的偏移量。

至此map整個階段結束。

mapTask的一些基礎設置配置（mapred-site.xml）：

設置一：設置環型緩沖區的內存值大小（默認設置如下）

mapreduce.task.io.sort.mb 100

設置二：設置溢寫百分比（默認設置如下）

mapreduce.map.sort.spill.percent 0.80

設置三：設置溢寫數據目錄（默認設置）

mapreduce.cluster.local.dir ${hadoop.tmp.dir}/mapred/local

設置四：設置一次最多合并多少個溢寫文件（默認設置如下）

mapreduce.task.io.sort.factor 10

二. ReduceTask 工作機制以及reduceTask的并行度

Reduce大致分為copy、sort、reduce三個階段，重點在前兩個階段。copy階段包含一個eventFetcher來獲取已完成的map列表，由Fetcher線程去copy數據，在此過程中會啟動兩個merge線程，分別為inMemoryMerger和onDiskMerger，分別將內存中的數據merge到磁盤和將磁盤中的數據進行merge。待數據copy完成之后，copy階段就完成了，開始進行sort階段，sort階段主要是執行finalMerge操作，純粹的sort階段，完成之后就是reduce階段，調用用戶定義的reduce函數進行處理。

詳細步驟

：

1、

Copy階段

，簡單地拉取數據。Reduce進程啟動一些數據copy線程(Fetcher)，通過HTTP方式請求maptask獲取屬于自己的文件。

2、

Merge階段

。這里的merge如map端的merge動作，只是數組中存放的是不同map端copy來的數值。Copy過來的數據會先放入內存緩沖區中，這里的緩沖區大小要比map端的更為靈活。merge有三種形式：內存到內存；內存到磁盤；磁盤到磁盤。默認情況下第一種形式不啟用。當內存中的數據量到達一定閾值，就啟動內存到磁盤的merge。與map 端類似，這也是溢寫的過程，這個過程中如果你設置有Combiner，也是會啟用的，然后在磁盤中生成了眾多的溢寫文件。第二種merge方式一直在運行，直到沒有map端的數據時才結束，然后啟動第三種磁盤到磁盤的merge方式生成最終的文件。

3、

合并排序

。把分散的數據合并成一個大的數據后，還會再對合并后的數據排序。

4、

對排序后的鍵值對調用reduce方法

，鍵相等的鍵值對調用一次reduce方法，每次調用會產生零個或者多個鍵值對，最后把這些輸出的鍵值對寫入到HDFS文件中。

三. MapReduceshuffle過程

map階段處理的數據如何傳遞給reduce階段，是MapReduce框架中最關鍵的一個流程，這個流程就叫shuffle。

shuffle: 洗牌、發牌——（核心機制：數據分區，排序，分組，規約，合并等過程）。

shuffle是Mapreduce的核心，它分布在Mapreduce的map階段和reduce階段。一般把從Map產生輸出開始到Reduce取得數據作為輸入之前的過程稱作shuffle。

1、

Collect階段

：將MapTask的結果輸出到默認大小為100M的環形緩沖區，保存的是key/value，Partition分區信息等。

2、

Spill階段

：當內存中的數據量達到一定的閥值的時候，就會將數據寫入本地磁盤，在將數據寫入磁盤之前需要對數據進行一次排序的操作，如果配置了combiner，還會將有相同分區號和key的數據進行排序。

3、

Merge階段

：把所有溢出的臨時文件進行一次合并操作，以確保一個MapTask最終只產生一個中間數據文件。

4、

Copy階段

：ReduceTask啟動Fetcher線程到已經完成MapTask的節點上復制一份屬于自己的數據，這些數據默認會保存在內存的緩沖區中，當內存的緩沖區達到一定的閥值的時候，就會將數據寫到磁盤之上。

5、

Merge階段

：在ReduceTask遠程復制數據的同時，會在后臺開啟兩個線程對內存到本地的數據文件進行合并操作。

6、

Sort階段

：在對數據進行合并的同時，會進行排序操作，由于MapTask階段已經對數據進行了局部的排序，ReduceTask只需保證Copy的數據的最終整體有效性即可。

Shuffle中的緩沖區大小會影響到mapreduce程序的執行效率，原則上說，緩沖區越大，磁盤io的次數越少，執行速度就越快

緩沖區的大小可以通過參數調整, 參數：mapreduce.task.io.sort.mb 默認100M

四. MapReduce總體工作機制

1. Map到Reduce內存角度宏觀流程

2. Map到reduce處理流程角度宏觀步驟

1、map邏輯完之后，將map的每條結果通過context.write進行collect數據收集。在collect中，會先對其進行分區處理，默認使用HashPartitioner。

2.MapReduce提供Partitioner接口，它的作用就是根據key或value及reduce的數量來決定當前的這對輸出數據最終應該交由哪個reduce task處理。默認對key hash后再以reduce task數量取模。默認的取模方式只是為了平均reduce的處理能力，如果用戶自己對Partitioner有需求，可以訂制并設置到job上。

3.當溢寫線程啟動后，需要對這80MB空間內的key做排序(Sort)。排序是MapReduce模型默認的行為，這里的排序也是對序列化的字節做的排序。

4.如果job設置過Combiner，那么現在就是使用Combiner的時候了。將有相同key的key/value對的value加起來，減少溢寫到磁盤的數據量。Combiner會優化MapReduce的中間結果，所以它在整個模型中會多次使用。

5.哪些場景才能使用Combiner呢？從這里分析，Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計算結果。Combiner只應該用于那種Reduce的輸入key/value與輸出key/value類型完全一致，且不影響最終結果的場景。比如累加，最大值等（求平均值絕不能用Combiner）。Combiner的使用一定得慎重，如果用好，它對job執行效率有幫助，反之會影響reduce的最終結果。

本次的分享就到這里了,不知道小伙伴們理解多少了呢~如果對您所幫助的話，請會評論！ 謝謝

MapReduce

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。