目錄

• MapReduce框架原理
• 一、InputFormat數(shù)據(jù)輸入
• • 1. 切片與MapTask并行度決定機(jī)制
  • 2. FielInputFormat切片機(jī)制
  • 3. FileInputFormat切片大小的參數(shù)設(shè)置
  • 4. TextInputFormat
  • • 1).FileInputFormat實(shí)現(xiàn)類
    • 2).TextInputFormat
  • 5. CombineTextInputFormat切片機(jī)制
  • • 1). 應(yīng)用場(chǎng)景
    • 2). 虛擬存儲(chǔ)切片最大值設(shè)置
    • 3). 切片機(jī)制
• 二、MapReduce工作流程

MapReduce框架原理

一、InputFormat數(shù)據(jù)輸入

1. 切片與MapTask并行度決定機(jī)制

1.問題：
MapTask的并行度決定Map階段的任務(wù)處理并發(fā)度，進(jìn)而影響整個(gè)job的處理速度。
但是相對(duì)于1G數(shù)據(jù)啟動(dòng)8個(gè)MapTask，可以提高集群的并發(fā)處理能力。1k的數(shù)據(jù)數(shù)據(jù)啟動(dòng)8個(gè)MapTask不一定會(huì)提高集群性能；MapTask并行任務(wù)是否越多越好？哪些因素影響了MapTask并行度？

MapTask 并行度由切片個(gè)數(shù)決定，切片個(gè)數(shù)由輸入文件和切片規(guī)則決定。

2.MapTask并行度決定機(jī)制

數(shù)據(jù)塊：block是HDFS物理上把數(shù)據(jù)進(jìn)行分塊(0-128MB)。數(shù)據(jù)塊是HDFS存儲(chǔ)數(shù)據(jù)單位；

數(shù)據(jù)切片：數(shù)據(jù)切片只是邏輯上對(duì)輸入進(jìn)行分片，并不會(huì)在磁盤上將其切分成片進(jìn)行存儲(chǔ)。數(shù)據(jù)切片是MapReduce程序計(jì)算輸入數(shù)據(jù)的單位，一個(gè)切片會(huì)對(duì)應(yīng)啟動(dòng)一個(gè)MapTask。

tips: 切片大小最好與block大小一致，即設(shè)置默認(rèn)128MB，處理更加有效率。

??：

ReduceTask=0，表示沒有Reduce階段，輸出文件個(gè)數(shù)和Map個(gè)數(shù)一致；

ReduceTask默認(rèn)值就是1，所以輸出文件個(gè)數(shù)為一個(gè)。 (3)如果數(shù)據(jù)分布不均勻，就有可能在Reduce階段產(chǎn)生數(shù)據(jù)傾斜；

ReduceTask數(shù)量并不是任意設(shè)置，還要考慮業(yè)務(wù)邏輯需求，有些情況下，需要計(jì)算全 局匯總結(jié)果，就只能有1個(gè)ReduceTask；

具體多少個(gè)ReduceTask，需要根據(jù)集群性能而定；

如果分區(qū)數(shù)不是1，但是ReduceTask為1，是否執(zhí)行分區(qū)過程。答案是:不執(zhí)行分區(qū)過 程。因?yàn)樵贛apTask的源碼中，執(zhí)行分區(qū)的前提是先判斷ReduceNum個(gè)數(shù)是否大于1。不大于1 肯定不執(zhí)行。

FielInputFormat切片源碼解析：

2. FielInputFormat切片機(jī)制

3. FileInputFormat切片大小的參數(shù)設(shè)置

4. TextInputFormat

1).FileInputFormat實(shí)現(xiàn)類

在運(yùn)行MapReduce程序時(shí)，輸入的文件格式包括：基于行的日志文件、二進(jìn)制格式文件、數(shù)據(jù)庫表等。那么，針對(duì)不同的數(shù)據(jù)類型，MapReduce時(shí)如何讀取這些數(shù)據(jù)的呢？

FileInputFormat常見的接口實(shí)現(xiàn)類包括：TextInputFormat、KeyValueInputFormat、NLineInputFormat、CombineTextInputFormat和自定義InputFormat等。

2).TextInputFormat

TextInputFormat是默認(rèn)的FileInputFormat實(shí)現(xiàn)類。按行讀取每條記錄。鍵時(shí)存儲(chǔ)該行在整個(gè)文件中的起始字節(jié)偏移量，LongWritable類型。值是這行的內(nèi)容，不包括任何行終止符(換行符和回車符)，Text類型。

5. CombineTextInputFormat切片機(jī)制

框架默認(rèn)的TextInputFormat切片機(jī)制是對(duì)任務(wù)按文件規(guī)劃切片，不管文件多小，都會(huì)是一個(gè)單獨(dú)的切片，都會(huì)交給一個(gè)MapTask，這樣如果有大量小文件，就會(huì)產(chǎn)生大量的Maptask，處理效率極其低下。

1). 應(yīng)用場(chǎng)景

CombineTextInputFormat用于小文件過多的場(chǎng)景，它可以將多個(gè)小文件從邏輯上規(guī)劃到一個(gè)切片中，這樣，多個(gè)小文件就可以交給一個(gè)MapTask處理。

2). 虛擬存儲(chǔ)切片最大值設(shè)置

CombineTextInputFormat.setmaxInputSplitSize(job,4194304); // 4M
tips: 虛擬存儲(chǔ)切片最大值設(shè)置最好根據(jù)實(shí)際的小文件大小情況來設(shè)置具體的值。

3). 切片機(jī)制

生成切片過程包括：虛擬存儲(chǔ)過程和切片過程兩部分。

二、MapReduce工作流程

切片數(shù)量影響Maptask，分區(qū)數(shù)量影響ReduceTask。

MapReduce詳細(xì)工作流程一：

??：MapTask工作機(jī)制共分為5個(gè)階段：Read階段、Map階段、Collect階段、溢寫階段、Merge階段。

Read 階段: MapTask 通過 InputFormat 獲得的 RecordReader，從輸入 InputSplit 中 解析出一個(gè)個(gè) key/value。

Map 階段: 該節(jié)點(diǎn)主要是將解析出的 key/value 交給用戶編寫 map()函數(shù)處理，并 產(chǎn)生一系列新的 key/value。

Collect 收集階段: 在用戶編寫 map()函數(shù)中，當(dāng)數(shù)據(jù)處理完成后，一般會(huì)調(diào)用 OutputCollector.collect()輸出結(jié)果。在該函數(shù)內(nèi)部，它會(huì)將生成的 key/value 分區(qū)(調(diào)用 Partitioner)，并寫入一個(gè)環(huán)形內(nèi)存緩沖區(qū)中。

Spill 階段: 即“溢寫”，當(dāng)環(huán)形緩沖區(qū)滿后，MapReduce 會(huì)將數(shù)據(jù)寫到本地磁盤上， 生成一個(gè)臨時(shí)文件。需要注意的是，將數(shù)據(jù)寫入本地磁盤之前，先要對(duì)數(shù)據(jù)進(jìn)行一次本地排序，并在必要時(shí)對(duì)數(shù)據(jù)進(jìn)行合并、壓縮等操作。

溢寫階段:
.
a. 利用快速排序算法對(duì)緩存區(qū)內(nèi)的數(shù)據(jù)進(jìn)行排序，排序方式是，先按照分區(qū)編號(hào) Partition 進(jìn)行排序，然后按照 key 進(jìn)行排序。這樣，經(jīng)過排序后，數(shù)據(jù)以分區(qū)為單位聚集在 一起，且同一分區(qū)內(nèi)所有數(shù)據(jù)按照 key 有序。
.
b. 按照分區(qū)編號(hào)由小到大依次將每個(gè)分區(qū)中的數(shù)據(jù)寫入任務(wù)工作目錄下的臨時(shí)文 件 output/spillN.out(N 表示當(dāng)前溢寫次數(shù))中。如果用戶設(shè)置了 Combiner，則寫入文件之 前，對(duì)每個(gè)分區(qū)中的數(shù)據(jù)進(jìn)行一次聚集操作。
.
c. 將分區(qū)數(shù)據(jù)的元信息寫到內(nèi)存索引數(shù)據(jù)結(jié)構(gòu) SpillRecord 中，其中每個(gè)分區(qū)的元信息包括在臨時(shí)文件中的偏移量、壓縮前數(shù)據(jù)大小和壓縮后數(shù)據(jù)大小。如果當(dāng)前內(nèi)存索引大 小超過 1MB，則將內(nèi)存索引寫到文件 output/spillN.out.index 中。

Merge 階段: 當(dāng)所有數(shù)據(jù)處理完成后，MapTask 對(duì)所有臨時(shí)文件進(jìn)行一次合并， 以確保最終只會(huì)生成一個(gè)數(shù)據(jù)文件。
當(dāng)所有數(shù)據(jù)處理完后，MapTask 會(huì)將所有臨時(shí)文件合并成一個(gè)大文件，并保存到文件 output/file.out 中，同時(shí)生成相應(yīng)的索引文件 output/file.out.index。
在進(jìn)行文件合并過程中，MapTask 以分區(qū)為單位進(jìn)行合并。對(duì)于某個(gè)分區(qū)，它將采用多 輪遞歸合并的方式。每輪合并 mapreduce.task.io.sort.factor(默認(rèn) 10)個(gè)文件，并將產(chǎn)生的文 件重新加入待合并列表中，對(duì)文件排序后，重復(fù)以上過程，直到最終得到一個(gè)大文件。
讓每個(gè) MapTask 最終只生成一個(gè)數(shù)據(jù)文件，可避免同時(shí)打開大量文件和同時(shí)讀取大量 小文件產(chǎn)生的隨機(jī)讀取帶來的開銷。

tips: 步驟7. 數(shù)據(jù)往內(nèi)存中寫到80%的時(shí)候，新開了一個(gè)線程把內(nèi)存中的舊數(shù)據(jù)往磁盤的文件進(jìn)行溢寫，另一個(gè)線程繼續(xù)把從MapTask來的數(shù)據(jù)寫到內(nèi)存里，因?yàn)榇藭r(shí)還未到100%，所以新的數(shù)據(jù)可以正常寫，不需要等所有的數(shù)據(jù)都溢寫完后再開始。

若是往內(nèi)存寫數(shù)據(jù)的線程寫到交界點(diǎn)，它會(huì)等溢寫完成后再繼續(xù)寫，此處的等待時(shí)間會(huì)比寫到100%之后溢寫等待的時(shí)間短，且此時(shí)的等待是為了保證數(shù)據(jù)干凈必須等待的。

步驟8. 在溢寫之前對(duì)數(shù)據(jù)進(jìn)行排序，通過對(duì)key的索引按照字典順序進(jìn)行快速排序。

步驟10. 對(duì)溢寫之后的數(shù)據(jù)進(jìn)行歸并排序。

combiner在聚合操作的場(chǎng)景下，使得傳到Reduce的數(shù)據(jù)量變小( <a,1><a,1> ==> <a,2> )，從而提高效率。

MapReduce詳細(xì)工作流程二：

??：ReduceTask工作機(jī)制共分為3個(gè)階段：Copy階段、Sort階段、Reduce階段。

Copy 階段: ReduceTask 從各個(gè) MapTask 上遠(yuǎn)程拷貝一片數(shù)據(jù)，并針對(duì)某一片數(shù) 據(jù)，如果其大小超過一定閾值，則寫到磁盤上，否則直接放到內(nèi)存中。

Sort 階段: 在遠(yuǎn)程拷貝數(shù)據(jù)的同時(shí)，ReduceTask 啟動(dòng)了兩個(gè)后臺(tái)線程對(duì)內(nèi)存和磁 盤上的文件進(jìn)行合并，以防止內(nèi)存使用過多或磁盤上文件過多。按照 MapReduce 語義，用 戶編寫 reduce()函數(shù)輸入數(shù)據(jù)是按 key 進(jìn)行聚集的一組數(shù)據(jù)。為了將 key 相同的數(shù)據(jù)聚在一 起，Hadoop 采用了基于排序的策略。由于各個(gè) MapTask 已經(jīng)實(shí)現(xiàn)對(duì)自己的處理結(jié)果進(jìn)行了 局部排序，因此，ReduceTask 只需對(duì)所有數(shù)據(jù)進(jìn)行一次歸并排序即可。

Reduce 階段: Reduce()函數(shù)將計(jì)算結(jié)果寫到 HDFS 上。

tips: 步驟13. ReduceTask主動(dòng)從MapTask分區(qū)拉取數(shù)據(jù)(copy階段)，而不是等待MapTask傳遞數(shù)據(jù)給它。

總結(jié)

以上是生活随笔為你收集整理的hadoop--MapReduce框架原理的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。