MapReduce采用的是“分而治之”的思想，把對大規模數據集的操作，分發給一個主節點管理下的各個從節點共同完成，然后通過整合各個節點的中間結果，得到最終結果。簡單來說，MapReduce就是”任務的分解與結果的匯總“。
MapReduce的工作原理
在分布式計算中，MapReduce框架負責處理了並行編程里分布式存儲、工作調度，負載均衡、容錯處理以及網絡通信等復雜問題，現在我們把處理過程高度抽象為Map與Reduce兩個部分來進行闡述，其中Map部分負責把任務分解成多個子任務，Reduce部分負責把分解后多個子任務的處理結果匯總起來，具體設計思路如下。
（1）Map過程需要繼承org.apache.hadoop.mapreduce包中Mapper類，並重寫其map方法。通過在map方法中添加兩句把key值和value值輸出到控制台的代碼，可以發現map方法中輸入的value值存儲的是文本文件中的一行（以回車符為行結束標記），而輸入的key值存儲的是該行的首字母相對於文本文件的首地址的偏移量。然后用StringTokenizer類將每一行拆分成為一個個的字段，把截取出需要的字段（本實驗為買家id字段）設置為key，並將其作為map方法的結果輸出。
（2）Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reducer類，並重寫其reduce方法。Map過程輸出的<key,value>鍵值對先經過shuffle過程把key值相同的所有value值聚集起來形成values，此時values是對應key字段的計數值所組成的列表，然后將<key,values>輸入到reduce方法中，reduce方法只要遍歷values並求和，即可得到某個單詞的總次數。
在main()主函數中新建一個Job對象，由Job對象負責管理和運行MapReduce的一個計算任務，並通過Job的一些方法對任務的參數進行相關的設置。本實驗是設置使用將繼承Mapper的doMapper類完成Map過程中的處理和使用doReducer類完成Reduce過程中的處理。還設置了Map過程和Reduce過程的輸出類型：key的類型為Text，value的類型為IntWritable。任務的輸出和輸入路徑則由字符串指定，並由FileInputFormat和FileOutputFormat分別設定。完成相應任務的參數設定后，即可調用job.waitForCompletion()方法執行任務，其余的工作都交由MapReduce框架處理。
實現內容:現有某電商網站用戶對商品的收藏數據，記錄了用戶收藏的商品id以及收藏日期，名為buyer_favorite1。
buyer_favorite1包含：買家id，商品id，收藏日期這三個字段，數據以“\t”【注：實際代碼中改成了四個空格】分割，樣本數據及格式如下：
買家id 商品id 收藏日期 10181 1000481 2010-04-04 16:54:31 20001 1001597 2010-04-07 15:07:52 20001 1001560 2010-04-07 15:08:27 20042 1001368 2010-04-08 08:20:30 20067 1002061 2010-04-08 16:45:33 20056 1003289 2010-04-12 10:50:55 20056 1003290 2010-04-12 11:57:35 20056 1003292 2010-04-12 12:05:29 20054 1002420 2010-04-14 15:24:12 20055 1001679 2010-04-14 19:46:04 20054 1010675 2010-04-14 15:23:53 20054 1002429 2010-04-14 17:52:45 20076 1002427 2010-04-14 19:35:39 20054 1003326 2010-04-20 12:54:44 20056 1002420 2010-04-15 11:24:49 20064 1002422 2010-04-15 11:35:54 20056 1003066 2010-04-15 11:43:01 20056 1003055 2010-04-15 11:43:06 20056 1010183 2010-04-15 11:45:24 20056 1002422 2010-04-15 11:45:49 20056 1003100 2010-04-15 11:45:54 20056 1003094 2010-04-15 11:45:57 20056 1003064 2010-04-15 11:46:04 20056 1010178 2010-04-15 16:15:20 20076 1003101 2010-04-15 16:37:27 20076 1003103 2010-04-15 16:37:05 20076 1003100 2010-04-15 16:37:18 20076 1003066 2010-04-15 16:37:31 20054 1003103 2010-04-15 16:40:14 20054 1003100 2010-04-15 16:40:16
編寫MapReduce程序，統計每個買家收藏商品數量。統計結果數據如下：
買家id 商品數量 10181 1 20001 2 20042 1 20054 6 20055 1 20056 12 20064 1 20067 1 20076 5
該mapreduce的執行過程:
實現思路：將hdfs上的文本作為輸入，MapReduce通過InputFormat會將文本進行切片處理（按行讀入），並將每行的首字母相對於文本文件的首地址的偏移量作為輸入鍵值對的key，文本內容作為輸入鍵值對的value，經過在map函數處理，輸出中間結果<word,1>的形式，並在reduce函數中完成對每個單詞的詞頻統計。
import java.io.IOException; import java.util.StringTokenizer; 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四個參數：第一個Object表示輸入key的類型；第二個Text表示輸入value的類型；第三個Text表示表示輸出鍵的類型；第四個IntWritable表示輸出值的類型。map這里的輸出是指輸出到reduce public static class doMapper extends Mapper<Object, Text, Text, IntWritable> { public static final IntWritable one = new IntWritable(1);//這里的IntWritable相當於Int類型 public static Text word = new Text();//Text相當於String類型 // map參數<keyIn key,valueIn value,Context context>，將處理后的數據寫入context並傳給reduce protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { //StringTokenizer是Java工具包中的一個類，用於將字符串進行拆分 StringTokenizer tokenizer = new StringTokenizer(value.toString(), " "); //返回當前位置到下一個分隔符之間的字符串 word.set(tokenizer.nextToken()); //將word存到容器中，記一個數 context.write(word, one); } } //參數同Map一樣，依次表示是輸入鍵類型，輸入值類型，輸出鍵類型，輸出值類型。這里的輸入是來源於map,所以類型要與map的輸出類型對應 。 public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; //for循環遍歷，將得到的values值累加 for (IntWritable value : values) { sum += value.get(); } result.set(sum); context.write(key, result);//將結果保存到context中，最終輸出形式為"key" + "result" } } public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { System.out.println("start"); Job job = Job.getInstance(); job.setJobName("wordCount"); Path in = new Path("hdfs://***:9000/user/hadoop/input/buyer_favorite1.txt");//設置這個作業輸入數據的路徑(***部分為自己liunx系統的localhost或者ip地址) Path out = new Path("hdfs://***:9000/user/hadoop/output/wordCount"); //設置這個作業輸出結果的路徑 FileInputFormat.addInputPath(job, in); FileOutputFormat.setOutputPath(job, out); job.setJarByClass(WordCount.class);// 設置運行/處理該作業的類 job.setMapperClass(doMapper.class);//設置實現了Map步的類 job.setReducerClass(doReducer.class);//設置實現了Reduce步的類 job.setOutputKeyClass(Text.class);//設置輸出結果key的類型 job.setOutputValueClass(IntWritable.class);//設置輸出結果value的類型 ////執行作業 System.exit(job.waitForCompletion(true) ? 0 : 1); System.out.println("end"); } }
最后附上林子雨大數據教程中的WordCount例子示意圖以加深理解
