目錄

• MapReduce框架原理
• 一、InputFormat數據輸入
• • 1. 切片與MapTask并行度決定機制
  • 2. FielInputFormat切片機制
  • 3. FileInputFormat切片大小的參數設置
  • 4. TextInputFormat
  • • 1).FileInputFormat實現類
    • 2).TextInputFormat
  • 5. CombineTextInputFormat切片機制
  • • 1). 應用場景
    • 2). 虛擬存儲切片最大值設置
    • 3). 切片機制
• 二、MapReduce工作流程

MapReduce框架原理

一、InputFormat數據輸入

1. 切片與MapTask并行度決定機制

1.問題：
MapTask的并行度決定Map階段的任務處理并發度，進而影響整個job的處理速度。
但是相對于1G數據啟動8個MapTask，可以提高集群的并發處理能力。1k的數據數據啟動8個MapTask不一定會提高集群性能；MapTask并行任務是否越多越好？哪些因素影響了MapTask并行度？

MapTask 并行度由切片個數決定，切片個數由輸入文件和切片規則決定。

2.MapTask并行度決定機制

數據塊：block是HDFS物理上把數據進行分塊(0-128MB)。數據塊是HDFS存儲數據單位；

數據切片：數據切片只是邏輯上對輸入進行分片，并不會在磁盤上將其切分成片進行存儲。數據切片是MapReduce程序計算輸入數據的單位，一個切片會對應啟動一個MapTask。

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

??：

ReduceTask=0，表示沒有Reduce階段，輸出文件個數和Map個數一致；

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

ReduceTask數量并不是任意設置，還要考慮業務邏輯需求，有些情況下，需要計算全 局匯總結果，就只能有1個ReduceTask；

具體多少個ReduceTask，需要根據集群性能而定；

如果分區數不是1，但是ReduceTask為1，是否執行分區過程。答案是:不執行分區過 程。因為在MapTask的源碼中，執行分區的前提是先判斷ReduceNum個數是否大于1。不大于1 肯定不執行。

FielInputFormat切片源碼解析：

2. FielInputFormat切片機制

3. FileInputFormat切片大小的參數設置

4. TextInputFormat

1).FileInputFormat實現類

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

FileInputFormat常見的接口實現類包括：TextInputFormat、KeyValueInputFormat、NLineInputFormat、CombineTextInputFormat和自定義InputFormat等。

2).TextInputFormat

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

5. CombineTextInputFormat切片機制

框架默認的TextInputFormat切片機制是對任務按文件規劃切片，不管文件多小，都會是一個單獨的切片，都會交給一個MapTask，這樣如果有大量小文件，就會產生大量的Maptask，處理效率極其低下。

1). 應用場景

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

2). 虛擬存儲切片最大值設置

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

3). 切片機制

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

二、MapReduce工作流程

切片數量影響Maptask，分區數量影響ReduceTask。

MapReduce詳細工作流程一：

??：MapTask工作機制共分為5個階段：Read階段、Map階段、Collect階段、溢寫階段、Merge階段。

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

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

Collect 收集階段: 在用戶編寫 map()函數中，當數據處理完成后，一般會調用 OutputCollector.collect()輸出結果。在該函數內部，它會將生成的 key/value 分區(調用 Partitioner)，并寫入一個環形內存緩沖區中。

Spill 階段: 即“溢寫”，當環形緩沖區滿后，MapReduce 會將數據寫到本地磁盤上， 生成一個臨時文件。需要注意的是，將數據寫入本地磁盤之前，先要對數據進行一次本地排序，并在必要時對數據進行合并、壓縮等操作。

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

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

tips: 步驟7. 數據往內存中寫到80%的時候，新開了一個線程把內存中的舊數據往磁盤的文件進行溢寫，另一個線程繼續把從MapTask來的數據寫到內存里，因為此時還未到100%，所以新的數據可以正常寫，不需要等所有的數據都溢寫完后再開始。

若是往內存寫數據的線程寫到交界點，它會等溢寫完成后再繼續寫，此處的等待時間會比寫到100%之后溢寫等待的時間短，且此時的等待是為了保證數據干凈必須等待的。

步驟8. 在溢寫之前對數據進行排序，通過對key的索引按照字典順序進行快速排序。

步驟10. 對溢寫之后的數據進行歸并排序。

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

MapReduce詳細工作流程二：

??：ReduceTask工作機制共分為3個階段：Copy階段、Sort階段、Reduce階段。

Copy 階段: ReduceTask 從各個 MapTask 上遠程拷貝一片數據，并針對某一片數 據，如果其大小超過一定閾值，則寫到磁盤上，否則直接放到內存中。

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

Reduce 階段: Reduce()函數將計算結果寫到 HDFS 上。

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。