Mapreduce实例之WORDCOUNT

实验内容

现有某电商网站用户对商品的收藏数据，记录了用户收藏的商品id以及收藏日期，名为buyer_favorite1。

buyer_favorite1包含：买家id，商品id，收藏日期这三个字段，数据以“ ”分割，样本数据及格式如下：

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()," ");  
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>。

完整代码

1. package mapreduce;  
2. import java.io.IOException;  
3. import java.util.StringTokenizer;  
4. import org.apache.hadoop.fs.Path;  
5. import org.apache.hadoop.io.IntWritable;  
6. import org.apache.hadoop.io.Text;  
7. import org.apache.hadoop.mapreduce.Job;  
8. import org.apache.hadoop.mapreduce.Mapper;  
9. import org.apache.hadoop.mapreduce.Reducer;  
10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
12. public class WordCount {  
13. public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {  
14. Job job = Job.getInstance();  
15. job.setJobName("WordCount");  
16. job.setJarByClass(WordCount.class);  
17. job.setMapperClass(doMapper.class);  
18. job.setReducerClass(doReducer.class);  
19. job.setOutputKeyClass(Text.class);  
20. job.setOutputValueClass(IntWritable.class);  
21. Path in = new Path("hdfs://localhost:9000/mymapreduce1/in/buyer_favorite1");  
22. Path out = new Path("hdfs://localhost:9000/mymapreduce1/out");  
23. FileInputFormat.addInputPath(job, in);  
24. FileOutputFormat.setOutputPath(job, out);  
25. System.exit(job.waitForCompletion(true) ? 0 : 1);  
26. }  
27. public static class doMapper extends Mapper<Object, Text, Text, IntWritable>{  
28. public static final IntWritable one = new IntWritable(1);  
29. public static Text word = new Text();  
30. @Override  
31. protected void map(Object key, Text value, Context context)  
32. throws IOException, InterruptedException {  
33. StringTokenizer tokenizer = new StringTokenizer(value.toString(), " ");  
34. word.set(tokenizer.nextToken());  
35. context.write(word, one);  
36. }  
37. }  
38. public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable>{  
39. private IntWritable result = new IntWritable();  
40. @Override  
41. protected void reduce(Text key, Iterable<IntWritable> values, Context context)  
42. throws IOException, InterruptedException {  
43. int sum = 0;  
44. for (IntWritable value : values) {  
45. sum += value.get();  
46. }  
47. result.set(sum);  
48. context.write(key, result);  
49. }  
50. }  
51. }  

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  ：