MapReduce實例：編寫MapReduce程序，統計每個買家收藏商品數量

實驗原理

MapReduce采用的是“分而治之”的思想，把對大規模數據集的操作，分發給一個主節點管理下的各個從節點共同完成，然后通過整合各個節點的中間結果，得到最終結果。簡單來說，MapReduce就是”任務的分解與結果的匯總“。

1.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框架處理。

2.MapReduce框架的作業運行流程

 

（1）ResourceManager：是YARN資源控制框架的中心模塊，負責集群中所有資源的統一管理和分配。它接收來自NM(NodeManager)的匯報，建立AM，並將資源派送給AM(ApplicationMaster)。

（2）NodeManager：簡稱NM，NodeManager是ResourceManager在每台機器上的代理，負責容器管理，並監控他們的資源使用情況（cpu、內存、磁盤及網絡等），以及向ResourceManager提供這些資源使用報告。

（3）ApplicationMaster：以下簡稱AM。YARN中每個應用都會啟動一個AM，負責向RM申請資源，請求NM啟動Container，並告訴Container做什么事情。

（4）Container：資源容器。YARN中所有的應用都是在Container之上運行的。AM也是在Container上運行的，不過AM的Container是RM申請的。Container是YARN中資源的抽象，它封裝了某個節點上一定量的資源（CPU和內存兩類資源）。Container由ApplicationMaster向ResourceManager申請的，由ResouceManager中的資源調度器異步分配給ApplicationMaster。Container的運行是由ApplicationMaster向資源所在的NodeManager發起的，Container運行時需提供內部執行的任務命令（可以是任何命令，比如java、Python、C++進程啟動命令均可）以及該命令執行所需的環境變量和外部資源（比如詞典文件、可執行文件、jar包等）。

另外，一個應用程序所需的Container分為兩大類，如下：

①運行ApplicationMaster的Container：這是由ResourceManager（向內部的資源調度器）申請和啟動的，用戶提交應用程序時，可指定唯一的ApplicationMaster所需的資源。

②運行各類任務的Container：這是由ApplicationMaster向ResourceManager申請的，並為了ApplicationMaster與NodeManager通信以啟動的。

以上兩類Container可能在任意節點上，它們的位置通常而言是隨機的，即ApplicationMaster可能與它管理的任務運行在一個節點上。

實驗環境

Linux Ubuntu 14.0

jdk-7u75-linux-x64

hadoop-2.6.0-cdh5.4.5

hadoop-2.6.0-eclipse-cdh5.4.5.jar

eclipse-java-juno-SR2-linux-gtk-x86_64

實驗內容

現有某電商網站用戶對商品的收藏數據，記錄了用戶收藏的商品id以及收藏日期，名為buyer_favorite1。

buyer_favorite1包含：買家id，商品id，收藏日期這三個字段，數據以“\t”分割，樣本數據及格式如下：

 

1. 買家id   商品id    收藏日期  
2. 10181   1000481   2010-04-04 16:54:31  
3. 20001   1001597   2010-04-07 15:07:52  
4. 20001   1001560   2010-04-07 15:08:27  
5. 20042   1001368   2010-04-08 08:20:30  
6. 20067   1002061   2010-04-08 16:45:33  
7. 20056   1003289   2010-04-12 10:50:55  
8. 20056   1003290   2010-04-12 11:57:35  
9. 20056   1003292   2010-04-12 12:05:29  
10. 20054   1002420   2010-04-14 15:24:12  
11. 20055   1001679   2010-04-14 19:46:04  
12. 20054   1010675   2010-04-14 15:23:53  
13. 20054   1002429   2010-04-14 17:52:45  
14. 20076   1002427   2010-04-14 19:35:39  
15. 20054   1003326   2010-04-20 12:54:44  
16. 20056   1002420   2010-04-15 11:24:49  
17. 20064   1002422   2010-04-15 11:35:54  
18. 20056   1003066   2010-04-15 11:43:01  
19. 20056   1003055   2010-04-15 11:43:06  
20. 20056   1010183   2010-04-15 11:45:24  
21. 20056   1002422   2010-04-15 11:45:49  
22. 20056   1003100   2010-04-15 11:45:54  
23. 20056   1003094   2010-04-15 11:45:57  
24. 20056   1003064   2010-04-15 11:46:04  
25. 20056   1010178   2010-04-15 16:15:20  
26. 20076   1003101   2010-04-15 16:37:27  
27. 20076   1003103   2010-04-15 16:37:05  
28. 20076   1003100   2010-04-15 16:37:18  
29. 20076   1003066   2010-04-15 16:37:31  
30. 20054   1003103   2010-04-15 16:40:14  
31. 20054   1003100   2010-04-15 16:40:16  

要求編寫MapReduce程序，統計每個買家收藏商品數量。

統計結果數據如下：

1. 買家id 商品數量  
2. 10181   1  
3. 20001   2  
4. 20042   1  
5. 20054   6  
6. 20055   1  
7. 20056   12  
8. 20064   1  
9. 20067   1  
10. 20076   5  

要求編寫MapReduce程序，統計每個買家收藏商品數量。

 實驗源碼如下：

package mapreduce1;  
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 {  
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {  
        Job job = Job.getInstance();  
        job.setJobName("WordCount");  
        job.setJarByClass(WordCount.class);  
        job.setMapperClass(doMapper.class);  
        job.setReducerClass(doReducer.class);  
        job.setOutputKeyClass(Text.class);  
        job.setOutputValueClass(IntWritable.class);  
        Path in = new Path("hdfs://192.168.57.128:9000/mymapreduce1/in/buyer_favorite1.txt");  
        Path out = new Path("hdfs://192.168.57.128:9000/mymapreduce1/out");  
        FileInputFormat.addInputPath(job, in);  
        FileOutputFormat.setOutputPath(job, out);  
        boolean flag = job.waitForCompletion(true);
        System.out.println(flag);
        System.exit( flag? 0 : 1);  
        
    }  
    public static class doMapper extends Mapper<Object, Text, Text, IntWritable>{  
        public static final IntWritable one = new IntWritable(1);  
        public static Text word = new Text();  
        @Override  
        protected void map(Object key, Text value, Context context)  
                    throws IOException, InterruptedException {  
            StringTokenizer tokenizer = new StringTokenizer(value.toString(), "   ");  
                word.set(tokenizer.nextToken());  
                context.write(word, one);  
        }  
    }  
    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 (IntWritable value : values) {  
        sum += value.get();  
        }  
        result.set(sum);  
        context.write(key, result);  
        }  
    }  
}

 實驗結果如下：