02．Mapreduce例項——求平均值

實驗目的

1.準確理解Mapreduce求平均值的設計原理

2.熟練掌握Mapreduce求平均值程式的編寫

3.學會編寫Mapreduce求平均值程式程式碼解決問題

實驗原理

求平均數是MapReduce比較常見的演算法，求平均數的演算法也比較簡單，一種思路是Map端讀取資料，在資料輸入到Reduce之前先經過shuffle，將map函式輸出的key值相同的所有的value值形成一個集合value-list，然後將輸入到Reduce端，Reduce端彙總並且統計記錄數，然後作商即可。具體原理如下圖所示：

實驗環境

Linux Ubuntu 14.04

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

實驗內容

現有某電商關於商品點選情況的資料檔案，表名為goods_click，包含兩個欄位（商品分類，商品點選次數），分隔符“\t”，由於資料很大，所以為了方便統計我們只擷取它的一部分資料，內容如下：

1. 商品分類商品點選次數
2. 521275
3. 5212093
4. 5209293
5. 5213238
6. 52006462
7. 5210928
8. 5210943
9. 521320

10. 5213234

11. 521329

12. 5213230

13. 5213245

14. 5213224

15. 520092615

16. 5213225

17. 5209013

18. 521326

19. 521360

20. 5209010

21. 52024347

要求使用mapreduce統計出每類商品的平均點選次數。

結果資料如下：

1. 商品分類商品平均點選次數
2. 52006462
3. 520092615
4. 52024347
5. 5209011
6. 5209293
7. 5210935
8. 5212093
9. 521275

10. 5213223

11. 521360

實驗步驟

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

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

2.在Linux本地新建/data/mapreduce4目錄。

1. mkdir-p/data/mapreduce4

3.在Linux中切換到/data/mapreduce4目錄下，用wget命令從http://192.168.1.100:60000/allfiles/mapreduce4/goods_click網址上下載文字檔案goods_click。

1. cd/data/mapreduce4
2. wgethttp://192.168.1.100:60000/allfiles/mapreduce4/goods_click

然後在當前目錄下用wget命令從http://192.168.1.100:60000/allfiles/mapreduce4/hadoop2lib.tar.gz網址上下載專案用到的依賴包。

1. wgethttp://192.168.1.100:60000/allfiles/mapreduce4/hadoop2lib.tar.gz

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

1. tarzxvfhadoop2lib.tar.gz

4.首先在HDFS上新建/mymapreduce4/in目錄，然後將Linux本地/data/mapreduce4目錄下的goods_click檔案匯入到HDFS的/mymapreduce4/in目錄中。

1. hadoopfs-mkdir-p/mymapreduce4/in
2. hadoopfs-put/data/mapreduce4/goods_click/mymapreduce4/in

5.新建Java Project專案，專案名為mapreduce4。

在mapreduce4專案下新建包，包名為mapreduce。

在mapreduce包下新建類，類名為MyAverage。

6.新增專案所需依賴的jar包，右鍵點選mapreduce4，新建一個資料夾，名為hadoop2lib，用於存放專案所需的jar包。

將/data/mapreduce4目錄下，hadoop2lib目錄中的jar包，拷貝到eclipse中mapreduce4專案的hadoop2lib目錄下。

選中hadoop2lib目錄下所有jar包，並新增到Build Path中。

7.編寫Java程式碼並描述其設計思路。

Mapper程式碼

1. publicstaticclassMapextendsMapper<Object,Text,Text,IntWritable>{
2. privatestaticTextnewKey=newText();
3. //實現map函式
4. publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{
5. //將輸入的純文字檔案的資料轉化成String
6. Stringline=value.toString();
7. System.out.println(line);
8. Stringarr[]=line.split("\t");
9. newKey.set(arr[0]);
10. 10. intclick=Integer.parseInt(arr[1]);
11. 11. context.write(newKey,newIntWritable(click));
12. 12. }
13. 13. }

map端在採用Hadoop的預設輸入方式之後，將輸入的value值通過split()方法截取出來，我們把擷取的商品點選次數字段轉化為IntWritable型別並將其設定為value，把商品分類欄位設定為key,然後直接輸出key/value的值。

Reducer程式碼

1. publicstaticclassReduceextendsReducer<Text,IntWritable,Text,IntWritable>{
2. //實現reduce函式
3. publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{
4. intnum=0;
5. intcount=0;
6. for(IntWritableval:values){
7. num+=val.get();//每個元素求和num
8. count++;//統計元素的次數count
9. }
10. 10. intavg=num/count;//計算平均數
11. 11.
12. 12. context.write(key,newIntWritable(avg));
13. 13. }
14. 14. }

map的輸出<key,value>經過shuffle過程整合<key,values>鍵值對，然後將<key,values>鍵值對交給reduce。reduce端接收到values之後，將輸入的key直接複製給輸出的key，將values通過for迴圈把裡面的每個元素求和num並統計元素的次數count，然後用num除以count 得到平均值avg，將avg設定為value，最後直接輸出<key,value>就可以了。

完整程式碼

1. packagemapreduce;
2. importjava.io.IOException;
3. importorg.apache.hadoop.conf.Configuration;
4. importorg.apache.hadoop.fs.Path;
5. importorg.apache.hadoop.io.IntWritable;
6. importorg.apache.hadoop.io.NullWritable;
7. importorg.apache.hadoop.io.Text;
8. importorg.apache.hadoop.mapreduce.Job;
9. importorg.apache.hadoop.mapreduce.Mapper;

10. importorg.apache.hadoop.mapreduce.Reducer;

11. importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;

12. importorg.apache.hadoop.mapreduce.lib.input.TextInputFormat;

13. importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

14. importorg.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

15. publicclassMyAverage{

1. 16. publicstaticclassMapextendsMapper<Object,Text,Text,IntWritable>{
2. 17. privatestaticTextnewKey=newText();
3. 18. publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{
4. 19. Stringline=value.toString();
5. 20. System.out.println(line);
6. 21. Stringarr[]=line.split("\t");
7. 22. newKey.set(arr[0]);
8. 23. intclick=Integer.parseInt(arr[1]);
9. 24. context.write(newKey,newIntWritable(click));
10. 25. }
11. 26. }
12. 27. publicstaticclassReduceextendsReducer<Text,IntWritable,Text,IntWritable>{
13. 28. publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{
14. 29. intnum=0;
15. 30. intcount=0;
16. 31. for(IntWritableval:values){
17. 32. num+=val.get();
18. 33. count++;
19. 34. }
20. 35. intavg=num/count;
21. 36. context.write(key,newIntWritable(avg));
22. 37. }
23. 38. }
24. 39. publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{
25. 40. Configurationconf=newConfiguration();
26. 41. System.out.println("start");
27. 42. Jobjob=newJob(conf,"MyAverage");
28. 43. job.setJarByClass(MyAverage.class);
29. 44. job.setMapperClass(Map.class);
30. 45. job.setReducerClass(Reduce.class);
31. 46. job.setOutputKeyClass(Text.class);
32. 47. job.setOutputValueClass(IntWritable.class);
33. 48. job.setInputFormatClass(TextInputFormat.class);
34. 49. job.setOutputFormatClass(TextOutputFormat.class);
35. 50. Pathin=newPath("hdfs://localhost:9000/mymapreduce4/in/goods_click");
36. 51. Pathout=newPath("hdfs://localhost:9000/mymapreduce4/out");
37. 52. FileInputFormat.addInputPath(job,in);
38. 53. FileOutputFormat.setOutputPath(job,out);
39. 54. System.exit(job.waitForCompletion(true)?0:1);
40. 55.
41. 56. }
42. 57. }

8.在MyAverage類檔案中，右鍵並點選=>Run As=>Run on Hadoop選項，將MapReduce任務提交到Hadoop中。

9.待執行完畢後，進入命令模式下，在HDFS上/mymapreduce4/out中檢視實驗結果。

1. hadoopfs-ls/mymapreduce4/out
2. hadoopfs-cat/mymapreduce4/out/part-r-00000