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

本次用 MapReduce 查找（输入一个学生的姓名，输出该生姓名以及其参加考试的课程和成绩）是我们《大数据基础》课程的期末大作业的功能需求之一。

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

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

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

一、数据及字段说明

这个数据与另一篇博客的输入数据是一样的，点击下方即可查看：
???? 【基于MapReduce的成绩分析系统】——计算每门课程的平均成绩、最高成绩、最低成绩 ????

二、过程分析及解题思路

• 需求: 求该成绩表每门课程当中出现了相同分数的分数，出现的次数，以及该相同分数的人数
  返回结果格式举例：PE 87 coconut,apple 共有2人
• 解题思路： 求该成绩表每门课程当中出现了相同分数的分数，出现的次数，以及该相同分数的人数，mapper阶段先按照课程和分数分组，然后map方法的输出key-value传递给reduce方法，reduce方法对每门课程出现了相同分数对应的所有学生姓名相关提取操作再输出。
• 关键： mapper阶段和reducer阶段的输入和输出是什么？
• 对于mapper阶段，map方法输出的key-value分别是
  key：课程名称 course 和 分数score
  value：学生姓名 name
• 对于reducer阶段，reduce方法输出的key-value分别是
  key：课程名称 course 和 分数score
  values： 每门课程当中相同分数的学生姓名names 和 人数number

三、具体代码实现

package Mapreduce.mark7;

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;

public class same7 {
	public static void main(String[] args) throws Exception {
		if (args.length<2) {
            System.out.printf("Usage:%s <input> <output>\n");
        }
		  //初始化一个Job对象
	        Configuration conf = new Configuration();
	        Job job = Job.getInstance(conf, "course same");
	        // 指定我这个job所在的jar包
	        job.setJarByClass(same7.class);
	        //指定mapper类和reducer类 等各种其他业务逻辑组件
	        job.setMapperClass(same7.sameMapper.class);
	        job.setReducerClass(same7.sameReducer.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/sameoutput7"));
			//最后向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: course +"\t"+ score    输入的value: name
	 private static class sameMapper 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 {
	        	//获取每一行内容  进行切分
	            String[] splits = value.toString().split(",");
	            String course = splits[0];
	            String score = splits[2];
	            String name = splits[1];
	          //key   value封装
	            keyOut.set(course +"\t"+ score);
	            valueOut.set(name);
	 
	            context.write(keyOut, valueOut);
	        }
	    }
	 //Reducer组件：
     //输入的key:     输入的values:
     //输出的key:  course + "\t" + score     输入的value:   number + "\t" + names
	 private static class sameReducer extends Reducer<Text, Text, Text, Text> {
		 
	        Text valueOut = new Text();
	 
	        @Override
	        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
	 
	            StringBuilder sb = new StringBuilder();
	            int number = 0;
	            for(Text t: values){
	                sb.append(t.toString()).append(",");//append方法是追加功能
	                number++;
	            }
	            if(number > 1){
	                String names = sb.toString().substring(0, sb.toString().length() - 1);//substring(参数)是java中截取字符串的一个方法
	                valueOut.set( "\t" + names +"\t" + "共有" + number + "人");
	                context.write(key, valueOut);
	            }
	        }
	    }
}

四、程序运行结果

• 程序处理完成之后的结果所保存的位置：hdfs://localhost:9000/user/hadoop/sameoutput7（这是自己定义的位置）。
• 程序的执行结果
  
  （可见，得到了我们想要的结果，课程中相同的分数，对应的学生和人数。）

---

本次的分享就到这里了，希望对大家有帮助，以上有任何错误，希望可以得到大佬们的指正，能够互相学习！✨

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

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

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

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

---