mapreduce實現學生平均成績

思路：

　　首先從文本讀入一行數據，按空格對字符串進行切割，切割后包含學生姓名和某一科的成績，map輸出key->學生姓名    value->某一個成績

　　然后在reduce里面對成績進行遍歷求和，求平均數，然后輸出key->學生姓名    value->平均成績

 

　　源數據：

　　　chines.txt　

zhangsan    78
lisi    89
wangwu    96
zhaoliu    67

　　english.txt

zhangsan    80
lisi    82
wangwu    84
zhaoliu    86

　　math.txt

zhangsan    88
lisi    99
wangwu    66
zhaoliu    77

　　源代碼：

package com.duking.hadoop;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
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.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.GenericOptionsParser;

public class Score {

	public static class Map extends

	Mapper<Object, Text, Text, IntWritable> {

		// 實現map函數

		public void map(Object key, Text value, Context context)

		throws IOException, InterruptedException {

			// 將輸入的純文本文件的數據轉化成String

			String line = value.toString();

			// 將輸入的數據首先按行進行分割

			StringTokenizer tokenizerArticle = new StringTokenizer(line);  //以空格分隔字符串

			// 分別對每一行進行處理

			while (tokenizerArticle.hasMoreElements()) {

				String strName= tokenizerArticle.nextToken();  // 學生姓名部分
				
				String strScore = tokenizerArticle.nextToken();// 成績部分
				
                Text name = new Text(strName);

                int scoreInt = Integer.parseInt(strScore);
				// 輸出姓名和成績

				context.write(name, new IntWritable(scoreInt));

			}

		}

	}

	public static class Reduce extends

	Reducer<Text, IntWritable, Text, IntWritable> {

		// 實現reduce函數

		public void reduce(Text key, Iterable<IntWritable> values,

		Context context) throws IOException, InterruptedException {

			int sum = 0;

			int count = 0;

			Iterator<IntWritable> iterator = values.iterator();  //循環遍歷成績

			while (iterator.hasNext()) {

				sum += iterator.next().get();// 計算總分

				count++;// 統計總的科目數

			}

			int average = (int) sum / count;// 計算平均成績

			context.write(key, new IntWritable(average));

		}

	}

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();

		conf.set("mapred.job.tracker", "192.168.60.129:9000");

		// 指定帶運行參數的目錄為輸入輸出目錄
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();

		/*
		 * 指定工程下的input2為文件輸入目錄 output2為文件輸出目錄 String[] ioArgs = new String[] {
		 * "input2", "output2" };
		 * 
		 * String[] otherArgs = new GenericOptionsParser(conf, ioArgs)
		 * .getRemainingArgs();
		 */

		if (otherArgs.length != 2) { // 判斷路徑參數是否為2個

			System.err.println("Usage: Data Deduplication <in> <out>");

			System.exit(2);

		}

		// set maprduce job name
		Job job = new Job(conf, "Score Average");

		job.setJarByClass(Score.class);

		// 設置Map、Combine和Reduce處理類

		job.setMapperClass(Map.class);

		job.setCombinerClass(Reduce.class);

		job.setReducerClass(Reduce.class);

		// 設置輸出類型

		job.setOutputKeyClass(Text.class);

		job.setOutputValueClass(IntWritable.class);

		// 設置輸入和輸出目錄

		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}