下面講解這八個流程
?Inputformat——》map——》（combine）——》partition——》copy&merge——》sort——》reduce——》outputformat
FileSplit:文件的子集--文件分割體
0. 簡介：
這篇文檔描述在hadoop中map和reduce操作是怎樣具體完成的。如果你對Google的MapReduce各式模式不熟悉，請先參閱MapReduce--http://labs.google.com/papers/mapreduce.html

1.MAP

1.1 Map階段

由于Map是并行地對輸入的文件集進行操作，所以它的第一步(FileSplit) 就是把文件集分割成一些子集.如果一個單個的文件大到它已影響到查找效率時，它會被分割成一些小的分割體。要指出的是分割這個一步是不知道輸入文件的內部邏輯結構的，比如，以行為邏輯分割的文本文件會被以任意的字節界限分割，所以這個具體分割要自己去指定也可以用hadoop已經定義的幾個簡單分割。然后每個文件分割體都會對應地有一個新的map任務。
當單個map任務開始時，它會對每個配置過的reduce任務開啟一個新的輸出書寫器(writer).緊接著它(writer)會用從指定的特定InputFormat里得到的RecordReader去讀它的文件分割體。InputFormat類分析輸入文件并產生key-value鍵值對.同時InputFormat必需要處理在以文件分割時邊界處的記錄。比如TextInputFormat會讀取分割邊界的文件分割體有最后一行，如果當讀取的分割體不是第一個時，TextInputFormat會忽略第一行的內容。
InputFormat類并不需要產生一些對有意義的鍵值對。比如TextInputFormat類的默認輸出是以輸入文本的行內容為value,以行偏移量為key--大多數應用只用到而很少用到偏移量。
傳給用戶配置的mapper的鍵值對都是從RecordReader讀入的，用戶提供的Mapper類就可以對鍵值對進行任意的操作再調用OutputCollector.collect方法來重新收集自己定義后的鍵值對。產生的輸出必需用一個Key類和一個value類，這是因為Map的輸出結果要被以SequenceFile的形式寫入磁盤，這種形式包括每個文件的類型信息和所有的記錄都是同類形的(如果你想輸出不同的數據結構你可以繼承個子類出來)。Map的輸入和輸出鍵值對不需要在類型上有聯系.
當Mapper的輸出被收集后，它們會被Partitioner類以指定的方式區分地寫出到輸出文件里。默認是以HashPartitioner類用key類的哈希函數產生的hashcode來區分(因此就要有一個很好的哈希函數，才可以使在各個reduce任務時負載勻衡)。詳細可以查看MapTask類。N個輸入可以產生M個map任務去跑，每個map任務會產生配置的reduce任務數個輸出文件。每個輸出文件都會面向一個特定的reduce任務同時所有從map任務產生的鍵值對都會被送到reduce里。所以在一個特定的reduce任務中對于一個給定的key所有的鍵值對都會被處理。

1.2 Combine（本地reduce）

當map操作輸出它的鍵值對時他們已經在內存中存在了。為了性能和效率的考慮，有時候提供一個擁有reduce功能的合成器是有好處的。如果有合成器，那么map的鍵值對就不會被馬上寫入到輸出里，他們會被收集在list里，一個key值一個list，當寫入一定數量的鍵值對時，這部分緩沖會被送進合成器，每個key都的所有value都會被送進合成器的reduce方法里并且就像原先map輸出的鍵值對一樣。
?比如，hadoop案例中的wordcount程序，它的map操作輸出是(word,1)鍵值對，在輸入中的詞的計數可以用合成器來加速這個操作。一個合成操作會在內存中收集處理lists，一個詞一個list。當一定數量的鍵值對輸出到內存中時，就調用合成操作的reduce方法，每次都以一個唯一的詞為key，values是list的迭代器。然后合成器輸出(word,count-in-this-part-of-the-input)鍵值對。從Reduce操作的觀點來說合成器也擁有Map輸出中相同的信息，但是這樣會比原先遠遠減少硬盤的讀寫。

2. Reduce

?當一個reduce任務開始時，它的輸入是分散在各個節點上的map的輸出文件里。如果在分布式的模式下，他們需要先在拷貝步驟里拷貝到本地文件系統上。詳細可以查看ReduceTaskRunner類
一旦所有的數據都在本地有效時，它會在添加步驟里加到一個文件里。然后這個文件會被合并分類這樣相同的key的鍵值對就可以排在一起(分類步驟)。這樣可以使真正的reduce操作變得簡單，這個文件會被順序地讀入，值(values)會從輸入文件里用一個迭代器傳給reduce方法-直到下一個key。詳細可以查看ReduceTask類。
?最后，輸出由每個reduce任務的輸出文件組成。面他們的格式可以由JobConf.setOutputFormat類指定，如果用到JobConf.setOutputFormat類，那么輸出的key類和value類都要同時指定。

3. 舉例子：

我們以wordcount為例，假設有個6400M的文件，100臺hadoop機器（準確地說應該是tasktracker機），默認block大小為64M，這樣每臺執行map的文件剛好是一個64M的block文件（假設這個分發過程已經完成，同時忽略備份數之類的細節），并且我們使用10個reduce任務來歸并文件。Hadoop的mapreducer的執行過程如下：
這100臺機器上面的map都是并發、獨立的執行，以wordcount為例，步驟如下：
1、 每個map任務使用默認的textinputformat類的LineRecordReader方法按行讀取文件，這個讀取的行數據就被交給map函數去執行，wordcount的map做的就是提取里面的單詞，并以單詞為key，1為value作為輸出，格式為：<wordinteger（1）>。
2、 如果有combine，先對第一步的輸出結果就行combine操作。Combine就是個小reduce操作，作用就是對某個map自己的輸出結果先進行一次歸并，把相同word的計數累加，這樣假設某個map輸出結果做如果有50%的重復word，那combine后的中間結果大小可以減少一半，可減少后續的patition、copy、sort等的開銷，提高性能。
3、 每個map對自己的輸出文件進行patition操作。上面提到有10個reducer任務，那默認的patition操作就是對map的輸出kay進行hash，并對10求余（hash(key)），并提供10個文件（內存足夠的話可以是鏈表等內存數據結構），假設是r1、r2….r10這10個文件，把不同key的放到不同的文件，這次操作就可以把相同key聚合到同一個文件。由于算法一樣，保證了每個map的輸出結果經過這個操作后，相同key的肯定在同一個聚合文件里，比如某個單詞word肯定都在r1文件里。
4、 接下來就是copy文件的過程了，10個reducer任務各自從所有map機器上取到屬于自己的文件，比如reducer1會從100臺map機器上取到所有r1文件，reducer2取所有r2的文件，這樣同一類word已經到了同一臺reducer機器上了。
5、 每個reducer合并（meger）自己取到的文件，reducer1就是合并100個r1文件（實際過程是在上面第4步操作中會邊copy邊meger，在內存中）。
6、 合并好后進行下sort（排序）操作，再次把不同小文件中的同一個單詞聚合在一起。作為提供給reduce操作的數據。
7、 進行reduce操作，對同一個單詞的value列表再次進行累加，最終得到某個單詞的詞頻數。
8、 Outputformat操作，把reduce結果寫到磁盤。
所以，總的流程應該是這樣的：
* ?Inputformat——》map——》（combine）——》partition——》copy&merge——》sort——》reduce——》outputformat
由此我們也可以看出，執行reduce的代價還是有些的，所以如果我們的應用只使用map就能搞定的話，那就盡量不要再有reduce操作在其中。

總結

以上是生活随笔為你收集整理的map-reduce的八个流程的全部內容，希望文章能夠幫你解決所遇到的問題。

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