【基于MapReduce的成绩分析系统】——计算每门课程的平均成绩、最高成绩、最低成绩

本次用 MapReduce 计算每门课程的平均成绩、最高成绩、最低成绩是我们《大数据基础》课程的期末大作业的功能需求之一。

临近期末，在这里记录一下自己的学习收获，希望大家在浏览的过程中有所收获。

由于能力有限，博客中难免会存在一些不足，有纰漏之处恳请大佬指正，不胜感激… …????

博客主页：爱跑步的mango ????

一、数据及字段说明

首先准备初始的数据，markinput.txt文件。（把该文件上传到伪分布式HDFS下，处理的输入路径对应为：hdfs://localhost:9000/user/hadoop/markinput）

music,apple,85
music,banana,54
music,cherry,86
music,coconut,85
music,durian,80
music,mango,99
music,grape,85
music,lemon,75
music,olive,48
music,papaya,88
music,watermelon,85
english,peach,57
english,lemon,85
english,olive,48
english,strawberry,85
english,cherry,85
english,apple,96
english,mango,85
english,banana,85
english,grape,75
chinese,lemon,75
chinese,mango,76
chinese,papaya,85
chinese,coconut,85
chinese,olive,42
chinese,watermelon,81
math,strawberry,85
math,mango,76
math,papaya,85
math,coconut,48
math,durian,90
math,banana,54
math,apple,85
math,grape,89
PE,apple,87
PE,banana,34
PE,cherry,96
PE,coconut,87
PE,durian,81
PE,mango,90
PE,grape,99
PE,lemon,65
PE,olive,78
PE,papaya,68
PE,watermelon,89

• 以上是处理的数据，该数据每行有三个字段值，分别是course,name,score
  第一个是课程名称，总共五个课程，包括music、chinese、english、math和PE。
  第二个是学生姓名。
  第三个是是每门课程学生考试的分数。

二、过程分析及解题思路

• 需求: 计算每门课程的平均成绩、最高成绩、最低成绩
  返回结果格式举例：PE max=99 min=34 average79.5

• 解题思路： 要计算每门课程的平均成绩、最高成绩、最低成绩，先按照课程分组，然后对各个课程对应的分数做max,min,average的聚合操作。（这里的分组是直接在mapper阶段把课程名称作为输出的key进行分组的，mapper阶段输出的value是每个学生参考某门课程对应的分数，然后将map方法的输出key-value传递给reduce方法。这样每门课程所有的记录会在同一个reduce方法中进行处理。）

• 关键： mapper阶段和reducer阶段的输入和输出是什么？

• 对于mapper阶段，map方法输出的key-value分别是
  key： 课程名称 course
  value： 分数 score

• 对于reducer阶段，reduce方法接收的参数是
  key： 课程名称 course
  values： 某一门课程对应的所有score的一个迭代器

三、具体代码实现

package Mapreduce.mark1;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class maxminaverage134 {
	//把业务逻辑相关的信息（哪个是mapper，哪个是reducer，要处理的数据在哪里，输出的结果放哪里……）描述成一个job对象
    //把这个描述好的job提交给集群去运行
	  public static void main(String[] args) throws Exception {
		  if (args.length<2) {
	            System.out.printf("Usage:%s <input> <output>\n");
	        }
		  //创建一个Configuration实体类对象
	        Configuration conf = new Configuration();
	        Job job = Job.getInstance(conf, "course maxminaverage");
	        // 指定我这个job所在的jar包
	        job.setJarByClass(maxminaverage134.class);
	        //指定mapper类和reducer类 等各种其他业务逻辑组件
	        job.setMapperClass(maxminaverage134.averageMapper.class);
	        job.setReducerClass(maxminaverage134.averageReducer.class);
	        //指定reducetask的输出类型
	        job.setOutputKeyClass(Text.class);
	        job.setOutputValueClass(Text.class);
	        
	        Path inputPath = new Path(args[0]);
	        Path outPutPath = new Path(args[1]); 
	        FileSystem fs = FileSystem.get(conf);
	        if (fs.exists(outPutPath)) fs.delete(outPutPath,true);
	        FileInputFormat.setInputPaths(job,inputPath);
	        FileOutputFormat.setOutputPath(job,outPutPath);
	        
	       //指定处理的输入路径
	     //   FileInputFormat.setInputPaths(job, new Path("hdfs://localhost:9000/user/hadoop/markinput"));
			// 指定处理完成之后的结果所保存的位置
		//	FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/user/hadoop/maxminaverageoutput"));
	        
			//最后向yarn集群提交这个job任务
			boolean waitForCompletion = job.waitForCompletion(true);
	        //System.exit(waitForCompletion ? 0 : 1);
			System.out.println(job.waitForCompletion(true)?1:0);
	    }
	  //Mapper组件
	  //输入的key:     输入的value: course,name,score
	  //输出的key: couse     输入的value: score
	  private static class averageMapper extends Mapper<LongWritable, Text, Text, Text> {
		  //map  方法的生命周期:  框架每传一行数据就被调用一次
		    //key :  这一行的起始点在文件中的偏移量
		    //value : 这一行的内容
	        Text keyOut = new Text();
	        Text valueOut = new Text();
	 
	        @Override
	        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
	        	//代码中  key   是行首字母的【偏移量】-----无规律可言,行首字母到所有内容最前端的
	        	//value  是一行真正的数据
	            String[] splits = value.toString().split(",");//.将Text类型的value转换成 string,将这一行用 "," 切分出各个单词
	            String course = splits[0];
	            String score = splits[2];
	 
	            keyOut.set(course);
	            valueOut.set(score);
	 
	            context.write(keyOut, valueOut);////输出数据,context上下文对象
	        }
	    }
	  //Reducer组件：输入的key:      输入的values:
	  //输出的key:     输入的value:
	  private static class averageReducer extends Reducer<Text, Text, Text, Text> {  
	        Text valueOut = new Text(); 
	        List<Integer> scoreList = new ArrayList<>();
	 
	        @Override
	        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
	 
	            scoreList.clear();
	            //遍历
	            for(Text t: values){
	                scoreList.add(Integer.valueOf(t.toString()));
	            }
	            // 求最高成绩和最低成绩
	            Integer maxScore = Collections.max(scoreList);//java.util.Collections类使用求集合最大值
	            Integer minScore = Collections.min(scoreList);//java.util.Collections类使用求集合最小值
	 
	            int sumScore = 0;
	            for(int score: scoreList){
	                sumScore += score;
	            }
	            // 求平均成绩
	            double avgScore = sumScore *1D / scoreList.size()*10;//1D是后缀，是double类型，会自动类型转换
	            valueOut.set("max="+maxScore + "\t" + "min="+minScore + "\t" +"average"+ Math.round(avgScore)/10D);
	            context.write(key, valueOut);//输出结果
	        }
	    }
}

四、程序运行结果

• 程序处理完成之后的结果所保存的位置：hdfs://localhost:9000/user/hadoop/maxminaverageoutput（这是自己定义的位置）。
• 程序的执行结果（课程名称 最大值 最小值 平均分）
  
  （可见，输出的结果就是我们期望得到的。）

---

本次的分享就到这里了，大家对MapReduce了解的基础上也需要配置好相关的环境，以上有任何错误，希望可以得到大佬们的指正，互相学习！✨

???? 基于MapReduce的成绩分析系统如果想看更多功能实现的学习体验，欢迎访问:

【基于MapReduce的成绩分析系统】——计算每门课程学生的平均成绩，并将平均成绩从高到低输出

【基于MapReduce的成绩分析系统】——查找（输入一个学生的姓名，输出该生姓名以及其参加考试的课程和成绩）

【基于MapReduce的成绩分析系统】——求该成绩表每门课程当中出现了相同分数的分数，出现的姓名以及该相同分数的人数

---