現有某電商網站用戶對商品的收藏數據，記錄了用戶收藏的商品id以及收藏日期，名為buyer_favorite1。 buyer_favorite1包含：買家id，商品id，收藏日期這三個字段，數據以“\t”分割

實驗內容（mapReduce安裝請按照林子雨教程http://dblab.xmu.edu.cn/blog/631-2/）

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

buyer_favorite1包含：買家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程序，統計每個買家收藏商品數量。

實驗原理

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可能與它管理的任務運行在一個節點上。

 

 新建map Reduce項目：（自動導入包）

將buyer_favorite1.txt放入hdfs中

 

 

package mapReduce;

           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://localhost:9000/mapReduce/mymapreduce1/buyer_favorite1"); 
          
  Path out = new Path("hdfs://localhost:9000/mapReduce/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> {

// 第一個 Object 表示輸入 key 的類型；第二個 Text 表示輸入 value 的類型；第三個 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(), " ");//以空格分割 
         

//StringTokenizer 是 Java 工具包中的一個類，用於將字符串進行拆分 

   word.set(tokenizer.nextToken());

// 返回當前位置到下一個分隔符之間的字符串

　　context.write(word, one);

// 將 word 存到容器中，記一個數

             } 
          
 } 
         

            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); 
          
  } 
          
 } 
          
} 
         

 

運行結果：