碼上快樂

相關內容    簡體    繁體

實驗6：Mapreduce實例——WordCount

本文轉載自   查看原文   2019-10-31 11:00   454 

實驗6：Mapreduce實例——WordCount

 

實驗說明：

1、 本次實驗是第六次上機，屬於驗證性實驗。實驗報告上交截止日期為2018年11月16日上午12點之前。

2、 實驗報告命名為：信1605-1班學號姓名實驗六.doc。

 

實驗目的

1.准確理解Mapreduce的設計原理

2.熟練掌握WordCount程序代碼編寫

3.學會自己編寫WordCount程序進行詞頻統計

實驗原理

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  

實驗步驟

1.切換目錄到/apps/hadoop/sbin下，啟動hadoop。

  1. cd /apps/hadoop/sbin  
  2. ./start-all.sh  

2.在linux上，創建一個目錄/data/mapreduce1。

  1. mkdir -p /data/mapreduce1  

         

3.切換到/data/mapreduce1目錄下，自行建立文本文件buyer_favorite1。

依然在/data/mapreduce1目錄下，使用wget命令，從

網絡下載hadoop2lib.tar.gz，下載項目用到的依賴包。

將hadoop2lib.tar.gz解壓到當前目錄下。

  1. tar -xzvf hadoop2lib.tar.gz  

4.將linux本地/data/mapreduce1/buyer_favorite1，上傳到HDFS上的/mymapreduce1/in目錄下。若HDFS目錄不存在，需提前創建。

  1. hadoop fs -mkdir -p /mymapreduce1/in  
  2. hadoop fs -put /data/mapreduce1/buyer_favorite1 /mymapreduce1/in  

       

 

5.打開Eclipse，新建Java Project項目。

 

並將項目名設置為mapreduce1。

 

6.在項目名mapreduce1下，新建package包。

 

並將包命名為mapreduce 。

 

7.在創建的包mapreduce下，新建類。

 

並將類命名為WordCount。

 

8.添加項目所需依賴的jar包，右鍵單擊項目名，新建一個目錄hadoop2lib，用於存放項目所需的jar包。

將linux上/data/mapreduce1目錄下，hadoop2lib目錄中的jar包，全部拷貝到eclipse中，mapreduce1項目的hadoop2lib目錄下。

選中hadoop2lib目錄下所有的jar包，單擊右鍵，選擇Build Path=>Add to Build Path

 

9.編寫Java代碼，並描述其設計思路。

下圖描述了該mapreduce的執行過程

 

大致思路是將hdfs上的文本作為輸入，MapReduce通過InputFormat會將文本進行切片處理，並將每行的首字母相對於文本文件的首地址的偏移量作為輸入鍵值對的key，文本內容作為輸入鍵值對的value，經過在map函數處理，輸出中間結果<word,1>的形式，並在reduce函數中完成對每個單詞的詞頻統計。整個程序代碼主要包括兩部分：Mapper部分和Reducer部分。

Mapper代碼

  1. public static class doMapper extends Mapper<Object, Text, Text, IntWritable>{  
  2. //第一個Object表示輸入key的類型；第二個Text表示輸入value的類型；第三個Text表示表示輸出鍵的類型；第四個IntWritable表示輸出值的類型  
  3. public static final IntWritable one = new IntWritable(1);  
  4. public static Text word = new Text();  
  5. @Override  
  6. protected void map(Object key, Text value, Context context)  
  7. throws IOException, InterruptedException  
  8. //拋出異常  
  9. {  
  10. StringTokenizer tokenizer = new StringTokenizer(value.toString(),"\t");  
  11. //StringTokenizer是Java工具包中的一個類，用於將字符串進行拆分  
  13. word.set(tokenizer.nextToken());  
  14. //返回當前位置到下一個分隔符之間的字符串  
  15. context.write(word, one);  
  16. //將word存到容器中，記一個數  
  17. }  

在map函數里有三個參數，前面兩個Object key,Text value就是輸入的key和value，第三個參數Context context是可以記錄輸入的key和value。例如context.write(word,one)；此外context還會記錄map運算的狀態。map階段采用Hadoop的默認的作業輸入方式，把輸入的value用StringTokenizer()方法截取出的買家id字段設置為key，設置value為1，然后直接輸出<key,value>。

Reducer代碼

  1. public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable>{  
  2. //參數同Map一樣，依次表示是輸入鍵類型，輸入值類型，輸出鍵類型，輸出值類型  
  3. private IntWritable result = new IntWritable();  
  4. @Override  
  5. protected void reduce(Text key, Iterable<IntWritable> values, Context context)  
  6. throws IOException, InterruptedException {  
  7. int sum = 0;  
  8. for (IntWritable value : values) {  
  9. sum += value.get();  
  10. }  
  11. //for循環遍歷，將得到的values值累加  
  12. result.set(sum);  
  13. context.write(key, result);  
  14. }  
  15. }  

map輸出的<key,value>先要經過shuffle過程把相同key值的所有value聚集起來形成<key,values>后交給reduce端。reduce端接收到<key,values>之后，將輸入的key直接復制給輸出的key,用for循環遍歷values並求和，求和結果就是key值代表的單詞出現的總次，將其設置為value，直接輸出<key,value>。

完整代碼

 

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/mymapreduce1/in/buyer_favorite1");
Path out = new Path("hdfs://localhost:9000/mymapreduce1/out");
FileInputFormat.addInputPath(job,in);
FileOutputFormat.setOutputPath(job,out);
System.exit(job.waitForCompletion(true)?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);
}
}
}

 

 

 

10.在WordCount類文件中，單擊右鍵=>Run As=>Run on Hadoop選項，將MapReduce任務提交到Hadoop中。

 

11.待執行完畢后，打開終端或使用hadoop eclipse插件，查看hdfs上，程序輸出的實驗結果。

  1. hadoop fs -ls /mymapreduce1/out  
  2. hadoop fs -cat /mymapreduce1/out/part-r-00000  

         


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Mapreduce 測試自帶實例 wordcount Python實現MapReduce,wordcount實例，MapReduce實現兩表的Join MapReduce本地運行模式wordcount實例（附：MapReduce原理簡析） hadoop之MapReduce WordCount分析 MapReduce實現WordCount mapreduce(1)--wordcount的實現 MapReduce程序（一）——wordCount 三.hadoop mapreduce之WordCount例子 Kettle實現MapReduce之WordCount Hadoop實例之利用MapReduce實現Wordcount單詞統計 (附源代碼)
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM