[[186685]]

MapReduce處理過程總覽

對于MP的處理過程我想大部分人都已經知道了其原理，思路不難，這是肯定的，但是整個過程中需要考慮的細枝末節的點還挺多的，MP的輸入輸出格式就是其中的一點，那本文就帶領大家看看MP中的格式設置問題。

map函數，起到了如下的轉換作用：map：(k1,v1)—>list(k2,v2)

reduce 函數，則起到了這樣的格式轉換作用：reduce：(k2,list(v2))—>list(k3,v3)

怎么，你說你看不懂?那看來你還是沒有對mapreduce的過程有所理解，看看這幅圖，不需要解釋，你就明白上面的格式轉化是什么意思了：

上面這幅圖出自我的另外一篇博文：編寫自己的***個Hadoop實例，如果你設置了combiner函數，那么中間的格式轉化將會是這個樣子：

map：(k1,v1)—>list(k2,v2)  
combiner：(k2,list(v2))—>list(k2,v2)  
reduce：(k2,list(v2))—>list(k3,v3) 

是的，沒錯，combiner從功能上來講就是一個reducer，它的存在大大減小了reducer的壓力。

partition函數對中間結果(k2,v2)進行處理，返回一個索引值，即分區號

partition：(k2,v2)—>integer 

在前面的文章中，也就是我的博客：MapReduce輸入分片詳解中，我提到了分片是與map函數數量相等，同時它不是咱們想的那樣是一個實物分片，在程序上輸入分片在java中表現為InputSplit接口

public abstract class InputSpilt{ 
       public abstract long getLength(); 
       public abstract String[] getLoacations(); 
} 

存儲位置供mapreduce使用，以便使map任務盡量在分片附近。分片大小是用來對分片進行排序，以便優先處理***的***分片，從而最小化時間。InputSplit不需要MR開發人直接處理，而是由InputFormat創建。

客戶端通過調用InputFormat的getSplits()計算分片，然后將他們送到application master(或jobtracker)，am使用存儲位置信息調度map任務在tasktracker上處理這些分片數據。map任務把輸入分片傳遞給InputFormat的getRecordReader()方法來獲得這個分片的RecordReader。RecordReader類似迭代器，對map任務進行迭代，來生成鍵/值對，然后傳遞給map函數。也就是說InputFormat不僅僅可以計算分片，進行數據分割，還可以對分片進行迭代，也就是說獲得分片的迭代器，所有有關分片的操作都由InputFormat來支持，可見其強大性。

輸入格式

那既然InputFormat這么牛逼，那我們就來看看這個接口到底包含了什么，先來看看下面這張圖：

FileInputFormat類

FileInputFormat是所有文件作為數據源的InputFormat的實現類，主要有兩個功能：指定輸入文件位置和輸入文件生成分片的實現代碼段。換句話說，它并不生成分片，只是返回文件位置，并且實現了分片算法。

FileInputFormat指定輸入路徑

addInputPath(Job job,Path path);  
addInputPaths(Job job,String paths);  
setInputPaths(Job job,Path ...inputPaths); 

可以添加一個路徑或者多個路徑，其中setInputPaths是以此設定完成的路徑列表。其中路徑可以是一個文件、一個目錄、或者一個glob(通配，通過通配符來獲取路徑)，當路徑是一個目錄的時候表示包含目錄下的所有文件。當目錄中包含目錄的時候，這個目錄也會被解釋稱文件，所以會報錯。可以通過使用一個文件glob或者一個過濾器根據命名模式限定選擇目錄中的文件。還可以通過設置屬性mapred.input.dir.recursive為true強制對目錄進行遞歸讀取。如果需要排除目錄中的個別文件，可以通過setInputPathFileter()設置一個過濾器來進行過濾，如果不設置過濾器，也會有默認的過濾器排除隱藏文件(以.和_開頭的)。路徑和過濾器業可以使用配置文件進行配置：mapred.input.dir和mapred.input.path.Fileter.class

小文件處理

(小文件是指比HDFS塊小很多)在Hadoop中使用小文件的弊端：

(1)、增加map開銷，因為每個分片都要執行一次map任務，map操作會造成額外的開銷

(2)、MapReduce處理數據的***速度就是和集群中的傳輸速度相同，而處理小文件將增加作業的尋址次數

(3)、浪費namenode的內存

解決方法：

使用SequenceFile將這些小文件合并成一個大文件或多個大文件：將文件名作為鍵，文本內容作為值。

但是如果HDFS中已經存在的大批小文件，可以使用CombinerFileInputFormat。

CombinerFileInputFormat把多個文件打包成一個文件以便每個mapper能夠處理更過的數據

避免切分

有時候不需要將文件進行切分，mapper完整處理每個輸入文件。例如檢查一個文件的所有記錄是否有序。

可以通過設置最小分片大小大于要處理的文件。第二種就是使用FileInputFormat的具體子類，并且重載isSplitable()方法，把返回值設置為false。

mapper中的信息

通過調用Mapper中的Context的getInputSolit()返回一個InputSplit，如果使用的是FileInputFormat，則可以強轉為FileSplit，然后用此訪問正在輸入文件的路徑getPath(),分片開始處的字節偏移量，getLength()分片的長度。

TextInputFormat

文本輸入是默認的InputFormat，每條記錄是一行輸入，鍵是LongWritable類型，存儲該記錄在整個文件的字節偏移量。值是該行的內容，不包括終止符(回車、換行等)，它被打包成Text對象。

KeyValueTextInputFormat

當文件中的每一行是一個鍵/值對，使用某個分界符進行分割，如制表符。可以通過mapreduce.input.keyvaluelinerecordreader.key.value.seperator屬性來指定分隔符。默認是一個制表符。其中這個鍵是分隔符前的文本，值是分隔符后的文本，其類型都是Text類型。如：

line1:this is line1 text  
line2:this is line2 text 

則被分為兩條記錄，分別是:

(line1,this is line1 text) 
(line2,this is line2 text) 

NLineInputFormat

在TextInputFormat和KeyValueTextInputFormat中，每個mapper收到的輸入行數并不確定，行數取決于輸入分片的大小和行的長度。如果希望mapper收到固定行數的輸入，可以使用NLineInputFormat作為InputFormat。與TextInputFormat奕揚，鍵是文件中行的字節偏移量，值是行的內容。

N是每個mapper收到的輸入行數，默認是1。可以通過mapreduce.input.lineinputformat.linespermap屬性設置。如：

On the top of the Crumetty Tree

The Quangle Wangle sat,

But his face you could not see,

On account of his Beaver Hat.

當N為2的時候，每個輸入分片包含兩行。

(0,On the top of the Crumetty Tree)

(33,The Quangle Wangle sat,)

另一個mapper則收到后兩行

(57,But his face you could not see,)

(89,On account of his Beaver Hat.)

StreamInputFormat

當解析XMl文件的時候可以使用StreamInputFormat，將stream.recordreader.class屬性設置為org.apache.Hadoop.Streaming.StreamXmlRecordReader使用StreamXmlRecordReader類。具體實現(沒用過)可以查看該類官方文檔

SequenceFileInputFormat

Hadoop順序文件格式存儲二進制的鍵/值對的序列。當需要使用順序文件作為MapReduce的輸入時，應該使用SequenceFileInputFormat。鍵和值由順序文件指定，只需要保證map輸入的類型匹配。

SequenceFileAsTextInputFormat

SequenceFileAsTextInputFormat是SequenceFileInputFormat的變體，將順序文件的鍵和值轉化為Text對象。

SequenceFileAsBinaryInputFormat

SequenceFileAsBinaryInputFormat是SequenceFileInputFormat的一種變體，獲取順序文件的鍵和值作為二進制對象。

MutipleInputs

一個MapReduce作業可能由多個輸入文件，但所有文件都由同一個InputFormat和同一個mapper來處理。但是數據格式卻有所不同，需要對不同的數據集進行連接操作。可以使用MutipleInputs類處理

DBInputFormat

DBInputFormat用于使用JDBC從關系數據庫中讀取數據。需要注意在數據庫中運行太多mapper讀取數據，可能會使數據庫受不了，所以一般使用DBInputFormat加載少量數據。可以現將數據導入到HDFS中，一般將關系性數據庫中數據導入到HDFS中可以使用Sqoop

輸出格式

TextOutputFormat

默認的輸出模式TextOutputFormat，每條記錄寫為一行。鍵和值是任意的，因為TextOutputFormat都要將其toString()轉換為字符串。鍵值默認使用制表符分割，可以使用mapreduce.output.textoutputformat.separator屬性改變分割符

SequenceFileOutputFormat

將輸出寫為一個順序文件，當輸出需要作為后續的MapReduce輸入的時候，這種輸出非常合適，因為它格式緊湊，容易被壓縮。

SequenceFileAsBinaryOutputFormat

SequenceFileAsBinaryOutputForamt與SequenceFileAsBinaryInputFormat對應，將輸出的鍵和值作為二進制格式寫到SequenceFile容器中。

MultipleOutputFormat

有時候可能需要將每個reduce輸出多個文件，可以使用MutltipleOutputFormat。

LazyOutputFormat

延遲輸出，他是封裝輸出格式，可以保證指定分區***條記錄輸出時才真正創建文件。