一个hadoop map/reduce例子

小fish

　　资源文件math
　　张三 99
　　李四 90
　　王五 90
　　赵六 60
　　资源文件china
　　张三 79
　　李四 75
　　王五 80
　　赵六 90
　　资源文件english
　　张三 89
　　李四 75
　　王五 70
　　赵六 90
　　分析：
　　map 阶段将将学生姓名作为key 成绩作为value.这样Reduce阶段得到的数据就是
　　key:张三 value:{99,79,89}
　　……
　　在Reduce中将学生的成绩球平均值。
　　实现：
　　package com.bwzy.Hadoop;
　　import java.io.IOException;
　　import java.util.StringTokenizer;
　　import org.apache.hadoop.conf.Configured;
　　import org.apache.hadoop.fs.Path;
　　import org.apache.hadoop.io.IntWritable;
　　import org.apache.hadoop.io.LongWritable;
　　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.Reducer.Context;
　　import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
　　import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
　　import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
　　import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
　　import org.apache.hadoop.util.Tool;
　　import org.apache.hadoop.util.ToolRunner;
　　import com.bwzy.hadoop.HeBing.Map;
　　import com.bwzy.hadoop.HeBing.Reduce;

　　public>
　　public static>　　public void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {
　　String line = value.toString();
　　StringTokenizer tokenizer = new StringTokenizer(line);
　　while(tokenizer.hasMoreElements()){
　　String strName = tokenizer.nextToken();
　　String strSorce = tokenizer.nextToken();
　　context.write(new Text(strName), new IntWritable(Integer.parseInt(strSorce)));
　　}
　　}
　　}

　　public static>　　public void reduce(Text key, Iterable values, Context context)
　　throws IOException, InterruptedException {
　　int sum = 0;
　　int num = 0;
　　for (IntWritable sorce : values) {
　　sum+=sorce.get();
　　num++;
　　}
　　context.write(key, new IntWritable((int)(sum/num)));
　　}
　　}
　　@Override
　　public int run(String[] arg0) throws Exception {
　　Job job = new Job(getConf());
　　job.setJobName("AvgSorce");
　　job.setOutputKeyClass(Text.class);
　　job.setOutputValueClass(IntWritable.class);
　　job.setMapperClass(Map.class);
　　//        job.setCombinerClass(Reduce.class);
　　job.setReducerClass(Reduce.class);
　　job.setInputFormatClass(TextInputFormat.class);
　　job.setOutputFormatClass(TextOutputFormat.class);
　　FileInputFormat.setInputPaths(job, new Path(arg0[0]));
　　FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
　　boolean success = job.waitForCompletion(true);
　　return success ? 0 : 1;
　　}
　　public static void main(String[] args) throws Exception {
　　int ret = ToolRunner.run(new AvgSorce(), args);
　　System.exit(ret);
　　}
　　}
　　运行：
　　1：将程序打包
　　选中打包的类-->右击-->Export-->java-->JAR file--填入保存路径-->完成
　　2：将jar包拷贝到hadoop的目录下。（因为程序中用到来hadoop的jar包）
　　3：将资源文件上传到定义的hdfs目录下
　　创建hdfs目录命令（在hadoop已经成功启动的前提下）：hadoop fs -mkdir /自定义/自定义/input
　　上传本地资源文件到hdfs上：hadop fs -put -copyFromLocal /home/user/Document/math /自定义/自定义/input
　　……
　　4：运行MapReduce程序：
　　hadoop jar /home/user/hadoop-1.0.4/AvgSorce.jar com.bwzy.hadoop.AvgSorce /自定义/自定义/input /自定义/自定义/output
　　说明：hadoop运行后会自动创建/自定义/自定义/output目录，在该目录下会有两个文件，其中一个文件中存放来MapReduce运行的结果。如果重新运行该程序，需要将/自定义/自定义/output目录删除，否则系统认为该结果已经存在了。
　　5：运行的结果为
　　张三 89
　　李四 80
　　王五 80
　　赵六 80
　　推荐阅读：
　　Hadoop 新 MapReduce 框架 Yarn 详解 http://www.linuxidc.com/Linux/2013-09/90090.htm
　　Hadoop中HDFS和MapReduce节点基本简介 http://www.linuxidc.com/Linux/2013-09/89653.htm
　　MapReduce的自制Writable分组输出及组内排序 http://www.linuxidc.com/Linux/2013-09/89652.htm
　　MapReduce的一对多连接操作 http://www.linuxidc.com/Linux/2013-09/89647.htm
　　Hadoop--两个简单的MapReduce程序 http://www.linuxidc.com/Linux/2013-08/88631.htm
　　Hadoop 中利用 MapReduce 读写 MySQL 数据 http://www.linuxidc.com/Linux/2013-07/88117.htm