Hadoop 學習系列（四）之 MapReduce 原理講解
原文連結： ofollow,noindex">pengtuo.tech/大資料研發/2018/…
本篇文章將會介紹 Hadoop 重要的計算框架 MapReduce 。
完整的 MapReduce 框架包含兩部分：
- 演算法邏輯層面，即
map、shuffle以及reduce三個重要演算法組成部分，本篇文章將會介紹這個層面； - 實際執行層面，即演算法邏輯作業在分散式主機中是以什麼形式和什麼流程執行的，因為自
MapReduce version2以後，作業都是提交給YARN進行管理，所以本文將不會介紹此部分。
系列其他文章有：
一、What is MapReduce?
MapReduce 是一個基於 java 的並行分散式計算框架，使用它來編寫的資料處理應用可以執行在大型的商用硬體叢集上來處理大型資料集中的可並行化問題，資料處理可以發生在儲存在檔案系統（非結構化）或資料庫（結構化）中的資料上。 MapReduce 可以利用資料的位置，在儲存的位置附近處理資料，以最大限度地減少通訊開銷。
MapReduce 框架通過編組分散式伺服器，並行執行各種任務，管理系統各部分之間的所有通訊和資料傳輸；其還能自動完成計算任務的並行化處理，自動劃分計算資料和計算任務，在叢集節點上自動分配和執行任務以及收集計算結果，將資料分佈儲存、資料通訊、容錯處理等平行計算涉及到的很多系統底層的複雜細節交由系統負責處理，減少開發人員的負擔。
MapReduce 還是一個並行程式設計模型與方法（Programming Model & Methodology）。它藉助於函式式程式設計語言Lisp的設計思想，提供了一種簡便的並行程式設計方法，將複雜的、運行於大規模叢集上的平行計算過程高度地抽象到了兩個函式：Map和Reduce，用Map和Reduce兩個函式程式設計實現基本的平行計算任務，提供了抽象的操作和並行程式設計介面，以簡單方便地完成大規模資料的程式設計和計算處理。
二、The Algorithm
MapReduce框架通常由三個操作（或步驟）組成：
-
Map：每個工作節點將map函式應用於本地資料，並將輸出寫入臨時儲存。主節點確保僅處理冗餘輸入資料的一個副本。 -
Shuffle：工作節點根據輸出鍵（由map函式生成）重新分配資料，對資料對映排序、分組、拷貝，目的是屬於一個鍵的所有資料都位於同一個工作節點上。 -
Reduce：工作節點現在並行處理每個鍵的每組輸出資料。
MapReduce 流程圖：
MapReduce 允許分散式執行 Map 操作，只要每個 Map 操作獨立於其他 Map 操作就可以並行執行。
另一種更詳細的，將 MapReduce 分為5個步驟的理解是：
- Prepare the Map() input ：
MapReduce框架先指定Map處理器，然後給其分配將要處理的輸入資料 -- 鍵值對K1，併為該處理器提供與該鍵值相關的所有輸入資料； - Run the user-provided Map() code ：
Map()在K1鍵值對上執行一次，生成由K2指定的鍵值對的輸出； - Shuffle the Map output to the Reduce processors ：將先前生成的
K2鍵值對，根據『鍵』是否相同移至相同的工作節點； - Run the user-provided Reduce() code ：對於每個工作節點上的
K2鍵值對進行Reduce()操作； - Produce the final output ：
MapReduce框架收集所有Reduce輸出，並按K2對其進行排序以產生最終結果進行輸出。
實際生產環境中，資料很有可能是分散在各個伺服器上，對於原先的大資料處理方法，則是將資料傳送至程式碼所在的地方進行處理，這樣非常低效且佔用了大量的頻寬，為應對這種情況， MapReduce 框架的處理方法是，將 Map() 操作或者 Reduce() 傳送至資料所在的伺服器上，以『移動計算替代移動資料』，來加速整個框架的執行速度，大多數計算都發生在具有本地磁碟上資料的節點上，從而減少了網路流量。
Mapper
一個 Map 函式就是對一些獨立元素組成的概念上的列表的每一個元素進行指定的操作，所以每個元素都是被獨立操作的，而原始列表沒有被更改，因為這裡建立了一個新的列表來儲存新的答案。這就是說， Map 操作是可以高度並行的
MapReduce 框架的 Map 和 Reduce 函式都是根據 (key, value) 形式的資料結構定義的。 Map 在一個數據域（Data Domain）中獲取一個鍵值對，然後返回一個鍵值對的列表：
Map(k1,v1) → list(k2,v2) 複製程式碼
Map 函式會被並行呼叫，應用於輸入資料集中的每個鍵值對（keyed by K1）。然後每個呼叫返回一個鍵值對（keyed by K2）列表。之後， MapReduce 框架從所有列表中收集具有相同 key （這裡是 k2）的所有鍵值對，並將它們組合在一起，為每個 key 建立一個組。
Reducer
而 Reduce 是對一個列表的元素進行適當的合併。雖然不如 Map 函式那麼並行，但是因為化簡總是有一個簡單的答案，大規模的運算相對獨立，所以化簡函式在高度並行環境下也很有用。 Reduce 函式並行應用於每個組，從而在同一個資料域中生成一組值：
Reduce(k2, list (v2)) → list(v3) 複製程式碼
Reduce 端接收到不同任務傳來的有序資料組。此時 Reduce() 會根據程式猿編寫的程式碼邏輯進行相應的 reduce 操作，例如根據同一個鍵值對進行計數加和等。如果 Reduce 端接受的資料量相當小，則直接儲存在記憶體中，如果資料量超過了該緩衝區大小的一定比例，則對資料合併後溢寫到磁碟中。
Partitioner
前面提到過， Map 階段有一個分割成組的操作，這個劃分資料的過程就是 Partition ，而負責分割槽的 java 類就是 Partitioner 。
Partitioner 元件可以讓 Map 對 Key 進行分割槽，從而將不同分割槽的 Key 交由不同的 Reduce 處理，由此， Partitioner 數量等同於 Reducer 的數量，一個 Partitioner 對應一個 Reduce 作業，可認為其就是 Reduce 的輸入分片，可根據實際業務情況程式設計控制，提高 Reduce 效率或進行負載均衡。 MapReduce 的內建分割槽是 HashPartition 。
具有多個分割總是有好處的，因為與處理整個輸入所花費的時間相比，處理分割所花費的時間很短。當分割較小時，可以更好的處理負載平衡，但是分割也不宜太小，如果過小，則會使得管理拆分和任務載入的時間在總執行時間中佔過高的比重。
下圖是 map 任務和 reduce 任務的示意圖：
三、WordCount Example
這裡給出一個統計詞頻案例的 Java 程式碼：
import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WordCount { // 繼承 Mapper 類，實現自己的 map 功能 public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); // map 功能必須實現的函式 public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } // 繼承 Reducer 類，實現自己的 reduce 功能 public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { // 初始化Configuration，讀取mapreduce系統配置資訊 Configuration conf = new Configuration(); // 構建 Job 並且載入計算程式 WordCount.class Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); //指定 Mapper、Combiner、Reducer，也就是我們自己繼承實現的類 job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); // 設定輸入輸出資料 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } } 複製程式碼
上述程式碼會發現在指定 Mapper 以及 Reducer 時，還指定了 Combiner 類， Combiner 是一個本地化的 reduce 操作（因此我們看見 WordCount 類裡是用 reduce 進行載入的），它是 map 運算的後續操作，與 map 在同一個主機上進行，主要是在 map 計算出中間檔案前做一個簡單的合併重複key值的操作，減少中間檔案的大小，這樣在後續進行到 Shuffle 時，可以降低網路傳輸成本，提高網路傳輸效率。
提交 MR 作業的命令：
hadoop jar {程式的 jar 包} {任務名稱} {資料輸入路徑} {資料輸出路徑}
複製程式碼
例如：
hadoop jar hadoop-mapreduce-wordcount.jar WordCount /sample/input /sample/output 複製程式碼
上述程式碼示意圖：
Map -> Shuffle -> Reduce 的中間結果，包括最後的輸出都是儲存在本地磁碟上。
四、Advantage & Shortcoming of MapReduce
MapReduce 的兩大優勢是：
1 ) 並行處理：
在 MapReduce 中，我們將作業劃分為多個節點，每個節點同時處理作業的一部分。因此， MapReduce 基於Divide and Conquer範例，它幫助我們使用不同的機器處理資料。由於資料由多臺機器而不是單臺機器並行處理，因此處理資料所需的時間會減少很多。
2 ) 資料位置：
我們將計算移動到 MapReduce 框架中的資料，而不是將資料移動到計算部分。資料分佈在多個節點中，其中每個節點處理駐留在其上的資料部分。
這使得具有以下優勢：
- 將處理單元移動到資料所在位置可以降低網路成本；
- 由於所有節點並行處理其部分資料，因此處理時間縮短；
- 每個節點都會獲取要處理的資料的一部分，因此節點不會出現負擔過重的可能性。
但是，MapReduce 也有其限制：
- 不能進行流式計算和實時計算，只能計算離線資料；
- 中間結果儲存在磁碟上，加大了磁碟的 I/O 負載，且讀取速度比較慢；
- 開發麻煩，例如
wordcount功能就需要很多的設定和程式碼量，而Spark將會非常簡單。