自己实现 一个MapReduce 示例

有这样一个实际的问题需要要通过hadoop的来解决一下。
有一个学生成绩表，有学生姓名 和成绩格式如下
[html] view plain copy

1. zs 89  
2. zs 100  
3. ls 98  
4. ls 100  
5. zs 20  
6. ww 89  
7. ww 67  
8. ls 30  
9. ww 20  

一个学生 有多个科目，有不同的成绩。
需要对每个同学的成绩求平均值。
同时，把这个student.txt 上传到 hadoop的 file System 中。

./bin/hadoop fs -put ~/file/student.txt

代码如下：

[java] view plain copy

1. package com.picc.test;  
2.   
3. import java.io.IOException;  
4. import java.util.Iterator;  
5. import java.util.StringTokenizer;  
6.   
7. import org.apache.hadoop.conf.Configuration;  
8. import org.apache.hadoop.fs.Path;  
9. import org.apache.hadoop.io.IntWritable;  
10. import org.apache.hadoop.io.LongWritable;  
11. import org.apache.hadoop.io.Text;  
12. import org.apache.hadoop.mapreduce.Job;  
13. import org.apache.hadoop.mapreduce.Mapper;  
14. import org.apache.hadoop.mapreduce.Reducer;  
15. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
16. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
17. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
18. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
19. import org.apache.hadoop.util.Tool;  
20. import org.apache.hadoop.util.ToolRunner;  
21. import org.slf4j.Logger;  
22. import org.slf4j.LoggerFactory;  
23.   
24. import com.picc.mapreducetest.MyMapReduceTest;  
25. /*** 
26. * 定义一个AvgScore 求学生的平均值 要实现一个Tool 工具类，是为了初始化一个hadoop配置实例 
27. */  
28. public class AvgScore implements Tool{  
29.     public static final Logger log=LoggerFactory.getLogger(AvgScore.class);  
30.     Configuration configuration;  
31.     // 是版本 0.20.2的实现  
32.     public static class MyMap extends Mapper<Object, Text, Text, IntWritable>{  
33.   
34.         @Override  
35.         protected void map(Object key, Text value, Context context)  throws IOException, InterruptedException {  
36.             String stuInfo = value.toString();//将输入的纯文本的数据转换成String  
37.             System.out.println("studentInfo:"+stuInfo);  
38.             log.info("MapSudentInfo:"+stuInfo);  
39.             //将输入的数据先按行进行分割  
40.             StringTokenizer tokenizerArticle = new StringTokenizer(stuInfo, "\n");  
41.             //分别对每一行进行处理  
42.             while(tokenizerArticle.hasMoreTokens()){  
43.             // 每行按空格划分   
44.                 StringTokenizer tokenizer = new StringTokenizer(tokenizerArticle.nextToken());  
45.                 String name = tokenizer.nextToken();//学生姓名  
46.                 String score = tokenizer.nextToken();//学生成绩  
47.                 Text stu = new Text(name);  
48.                 int intscore = Integer.parseInt(score);  
49.                 log.info("MapStu:"+stu.toString()+" "+intscore);  
50.                 context.write(stu,new IntWritable(intscore));//输出学生姓名和成绩    
51.             }  
52.         }  
53.           
54.     }  
55.     public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{  
56.   
57.         @Override  
58.         protected void reduce(Text key, Iterable<IntWritable> values,Context context)  
59.                 throws IOException, InterruptedException {  
60.             int sum=0;  
61.             int count=0;  
62.             Iterator<IntWritable> iterator=    values.iterator();  
63.             while(iterator.hasNext()){  
64.                 sum+=iterator.next().get();//计算总分  
65.                 count++;//统计总科目  
66.             }  
67.             int avg= (int)sum/count;  
68.             context.write(key,new  IntWritable(avg));//输出学生姓名和平均值  
69.         }  
70.           
71.     }  
72.     public  int run(String [] args) throws Exception{  
73.           
74.           Job job = new Job(getConf());  
75.           job.setJarByClass(AvgScore.class);  
76.           job.setJobName("avgscore");  
77.           job.setOutputKeyClass(Text.class);  
78.           job.setOutputValueClass(IntWritable.class);  
79.           job.setMapperClass(MyMap.class);  
80.           job.setCombinerClass(MyReduce.class);  
81.           job.setReducerClass(MyReduce.class);  
82.           job.setInputFormatClass(TextInputFormat.class);  
83.           job.setOutputFormatClass(TextOutputFormat.class);  
84.           FileInputFormat.addInputPath(job, new Path(args[0]));//设置输入文件路径   
85.           FileOutputFormat.setOutputPath(job, new Path(args[1]));//设置输出文件路径  
86.           boolean success=  job.waitForCompletion(true);  
87.             
88.           return success ? 0 : 1;  
89.             
90.     }  
91.    public static void main(String[] args) throws Exception {  
92.     //在eclipse 工具上配置输入和输出参数  
93.     int ret = ToolRunner.run(new AvgScore(), args);  
94.     System.exit(ret);  
95.    }  
96. @Override  
97. public Configuration getConf() {  
98.     return configuration;  
99. }  
100. @Override  
101. public void setConf(Configuration conf) {  
102.        conf = new Configuration();  
103.        configuration=conf;  
104. }  
105. }  

我在eclipse 上配置参数。会报异常。所以，我把以上代码导出成 avgscore.jar
把这个avgscore.jar 放到hadoop 0.20.2/目录下。
输入命令 ./bin/hadoop  jar avgscore.jar com/picc/test/AvgScore  input/student.txt out1

结果 图：

和计算的结果 没有错。

以下是对 以上算法的一个分析：

[java] view plain copy

1.  package com.picc.test;  
2.   
3. import java.io.IOException;  
4. import java.util.Iterator;  
5. import java.util.StringTokenizer;  
6.   
7. import org.apache.hadoop.conf.Configuration;  
8. import org.apache.hadoop.fs.Path;  
9. import org.apache.hadoop.hbase.HBaseConfiguration;  
10. import org.apache.hadoop.hbase.client.HTable;  
11. import org.apache.hadoop.hbase.client.Put;  
12. import org.apache.hadoop.hbase.util.Bytes;  
13. import org.apache.hadoop.io.IntWritable;  
14. import org.apache.hadoop.io.LongWritable;  
15. import org.apache.hadoop.io.Text;  
16. import org.apache.hadoop.mapreduce.Job;  
17. import org.apache.hadoop.mapreduce.Mapper;  
18. import org.apache.hadoop.mapreduce.Reducer;  
19. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
20. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
22. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
23. import org.apache.hadoop.util.Tool;  
24. import org.apache.hadoop.util.ToolRunner;  
25. import org.slf4j.Logger;  
26. import org.slf4j.LoggerFactory;  
27.   
28. import com.picc.mapreducetest.MyMapReduceTest;  
29.   
30. public class AvgScore implements Tool{  
31.     public static final Logger log=LoggerFactory.getLogger(AvgScore.class);  
32.     Configuration configuration;  
33.       
34.     public static class MyMap extends Mapper<Object, Text, Text, IntWritable>{  
35.         Configuration config = HBaseConfiguration.create();//获取hbase 的操作上下文  
36.         private static IntWritable linenum = new IntWritable(1);//初始化一个变量值  
37.         @Override  
38.         protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {  
39.             String stuInfo = value.toString();  
40.             System.out.println("studentInfo:"+stuInfo);  
41.             log.info("MapSudentInfo:"+stuInfo);  
42.             StringTokenizer tokenizerArticle = new StringTokenizer(stuInfo, "\n");  
43.             while(tokenizerArticle.hasMoreTokens()){  
44.                 StringTokenizer tokenizer = new StringTokenizer(tokenizerArticle.nextToken());  
45.                 String name = tokenizer.nextToken();  
46.                 String score = tokenizer.nextToken();  
47.                 Text stu = new Text(name);  
48.                 int intscore = Integer.parseInt(score);  
49.                 log.info("MapStu:"+stu.toString()+" "+intscore);  
50.                 context.write(stu,new IntWritable(intscore));  //zs 90  
51.                 //create 'stu','name','score'  
52.                 HTable table=new HTable(config,"stu");  
53.                 byte[] row1 = Bytes.toBytes("name"+linenum);  
54.                 Put p1=new Put(row1);  
55.                 byte[] databytes = Bytes.toBytes("name");  
56.                 p1.add(databytes, Bytes.toBytes("1"), Bytes.toBytes(name));  
57.                 table.put(p1);//put 'stu','name','name:1','zs'  
58.                 table.flushCommits();  
59.                   
60.                 byte [] row2 = Bytes.toBytes("score"+linenum);  
61.                 Put p2 = new Put(row2);  
62.                 byte [] databytes2 = Bytes.toBytes("score");  
63.                 p2.add(databytes2, Bytes.toBytes("1"), Bytes.toBytes(score));  
64.                 table.put(p2);//put 'stu','score','score:1','90'  
65.                 table.flushCommits();  
66.                 linenum= new IntWritable(linenum.get()+1);//对变量值进行变值处理  
67.             }  
68.         }  
69.           
70.     }  
71.     public static class MyReduce extends Reducer<Text, IntWritable, Text, IntWritable>{  
72.   
73.         @Override  
74.         protected void reduce(Text key, Iterable<IntWritable> values,Context context)  
75.                 throws IOException, InterruptedException {  
76.             int sum=0;  
77.             int count=0;  
78.             Iterator<IntWritable> iterator=    values.iterator();  
79.             while(iterator.hasNext()){  
80.                 sum+=iterator.next().get();  
81.                 count++;  
82.             }  
83.             int avg= (int)sum/count;  
84.             context.write(key,new  IntWritable(avg));  
85.         }  
86.           
87.     }  
88.     public  int run(String [] args) throws Exception{  
89.           
90.           Job job = new Job(getConf());  
91.           job.setJarByClass(AvgScore.class);  
92.           job.setJobName("avgscore");  
93.           job.setOutputKeyClass(Text.class);  
94.           job.setOutputValueClass(IntWritable.class);  
95.           job.setMapperClass(MyMap.class);  
96.           job.setCombinerClass(MyReduce.class);  
97.           job.setReducerClass(MyReduce.class);  
98.           job.setInputFormatClass(TextInputFormat.class);  
99.           job.setOutputFormatClass(TextOutputFormat.class);  
100.           FileInputFormat.addInputPath(job, new Path(args[0]));  
101.           FileOutputFormat.setOutputPath(job, new Path(args[1]));  
102.           boolean success=  job.waitForCompletion(true);  
103.             
104.           return success ? 0 : 1;  
105.             
106.     }  
107.    public static void main(String[] args) throws Exception {  
108.       
109.     int ret = ToolRunner.run(new AvgScore(), args);  
110.     System.exit(ret);  
111.    }  
112. @Override  
113. public Configuration getConf() {  
114.     return configuration;  
115. }  
116. @Override  
117. public void setConf(Configuration conf) {  
118.        conf = new Configuration();  
119.        configuration=conf;  
120. }  
121. }  

这个代码是对上一个代码的调试分析处理后的代码，

把map 处理的过程放到的数据库中，在MapReduce 中处理 hbase数据时，需要 把hbase 的数据包放到hadoop的lib 包下。

处理的结果，见视图：

注意，在hbase数据库中 row中的Key是不能相同的，否则会 后一条会覆盖前一条值。需要保让其唯一性。

name1 和score1 是一条数据，这两列表是一个学生的成绩，和关系型数据库不同，以列值存储，思想需要转换一下。