平均成绩

"平均成绩"主要目的还是在重温经典"WordCount"例子，可以说是在基础上的微变化版对输入文件中数据进行就算学生平均成绩。输入文件中的每行内容均为一个学生的姓名和他相应的成绩，如果有多门学科，则每门学科为一个文件。要求在输出中每行有两个间隔的数据，其中，第一个代表学生的姓名，第二个代表其平均成绩。

数据准备

amath

张三 88

李四 99

王五 66

赵六 77

achinese

张三 78

李四 89

王五 96

赵六 67

aenglish

张三 80

李四 82

王五 84

赵六 86

设计思路

　　计算学生平均成绩是一个仿"WordCount"例子，用来重温一下开发MapReduce程序的流程。程序包括两部分的内容：Map部分和Reduce部分，分别实现了map和reduce的功能。

　　Map处理的是一个纯文本文件，文件中存放的数据时每一行表示一个学生的姓名和他相应一科成绩。Mapper处理的数据是由InputFormat分解过的数据集，其中 InputFormat的作用是将数据集切割成小数据集InputSplit，每一个InputSlit将由一个Mapper负责处理。此外，InputFormat中还提供了一个RecordReader的实现，并将一个InputSplit解析成<key,value>对提供给了map函数。InputFormat的默认值是TextInputFormat，它针对文本文件，按行将文本切割成InputSlit，并用 LineRecordReader将InputSplit解析成<key,value>对，key是行在文本中的位置，value是文件中的一行。

　　Map的结果会通过partion分发到Reducer，Reducer做完Reduce操作后，将通过以格式OutputFormat输出。

　　Mapper最终处理的结果对<key,value>，会送到Reducer中进行合并，合并的时候，有相同key的键/值对则送到同一个 Reducer上。Reducer是所有用户定制Reducer类地基础，它的输入是key和这个key对应的所有value的一个迭代器，同时还有 Reducer的上下文。Reduce的结果由Reducer.Context的write方法输出到文件中。

import org.apache.hadoop.conf.Configuration;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class AvgScore extends Configured implements Tool {

	@Override
	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub
		Configuration conf = getConf();
		Job job = new Job(conf, "avgScore");
		job.setJarByClass(AvgScore.class);
		job.setMapperClass(AvgMap.class);
		job.setReducerClass(AvgReducer.class);
		// 不设置时,默认输入输出格式,
		/*
		 * job.setInputFormatClass(TextInputFormat.class);
		 * job.setOutputFormatClass(TextOutputFormat.class);
		 */
		// 设置map的key/value的输出格式
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		// 设置reducer的key/value的输出格式
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path("/value/a*"));
		FileOutputFormat.setOutputPath(job, new Path("/outvalue/avgScore"));
		job.submit();
		return job.isSuccessful() ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		ToolRunner.run(new Configuration(), new AvgScore(), null);
	}
}

class AvgMap extends Mapper<LongWritable, Text, Text, IntWritable> {
	protected void map(LongWritable key, Text value, Context context)
			throws java.io.IOException, InterruptedException {
		// 将输入的纯文本文件的数据转化成String
		String line = value.toString();
		// 每行按空格划分
		String[] lineSplit = line.split("\\s++");
		// 学生姓名
		String name = lineSplit[0];
		// 学生成绩
		int score = Integer.parseInt(lineSplit[1]);
		context.write(new Text(name), new IntWritable(score));
	}
}

class AvgReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	private IntWritable avg = new IntWritable();

	protected void reduce(Text key, java.lang.Iterable<IntWritable> values,
			Context context) throws java.io.IOException, InterruptedException {
		int sum = 0;
		int count = 0;
		for (IntWritable score : values) {
			sum += score.get();
			count++;
		}
		int avgScore = sum / count;
		avg.set(avgScore);
		context.write(key, avg);
	}
}

运行结果

张三 82
李四 90
王五 82
赵六 76