【基于MapReduce的成绩分析系统】——查找（输入一个学生的姓名，输出该生姓名以及其参加考试的课程和成绩）
程序员文章站
2022-03-24 13:45:47
...
| 本次用 MapReduce 查找（输入一个学生的姓名，输出该生姓名以及其参加考试的课程和成绩）是我们《大数据基础》课程的期末大作业的功能需求之一。 临近期末，在这里记录一下自己的学习收获，希望大家在浏览的过程中有所收获。 由于能力有限，博客中难免会存在一些不足，有纰漏之处恳请大佬指正，不胜感激… …???? 博客主页：爱跑步的mango ???? |
MapReduce练习目录
一、数据及字段说明
这个数据与另一篇博客的输入数据是一样的，点击下方即可查看：
???? 【基于MapReduce的成绩分析系统】——计算每门课程的平均成绩、最高成绩、最低成绩 ????
二、过程分析及解题思路
-
需求: 查找（输入一个学生的姓名，输出该生姓名以及其参加考试的课程和成绩）
返回结果格式举例：olive:english,48;PE,78;music,48;chinese,42;（这里的olive是我们指定的要查找的学生姓名，后面其参加考试的课程和对应的成绩，可以指定查找其他学生。） - 解题思路： 查找（输入一个学生的姓名，输出该生姓名以及其参加考试的课程和成绩），在main方法中传递变量属性，将要查询的学生姓名通过参数设置，在map阶段取出，主要考虑找到聚合的key，然后reduce方法去拉取数据的时候会进行归并排序，这里用自定义Student类实现WritableComparable接口的compareTo方法定义的逻辑来比较是否是相同的数据，也就是学生姓名。这样相同的key数据就会在同一个reduce方法中执行，这时候我们在reduce方法中进行相关的过滤，就可以找到目标数据了。
- 关键： mapper阶段和reducer阶段的输入和输出是什么？
- 对于mapper阶段，map方法输出的key-value分别是
key：Student
value： NullWritable - 对于reducer阶段，reduce方法输出的key-value分别是
key：Student
values： NullWritable
三、具体代码实现
- 自定义数据类型：Student 的代码
package Mapreduce.mark6;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class Student implements WritableComparable<Student> {
private StringBuilder courseName;//课程名
private String studentName;//学生姓名
private Integer score;
public Student() {
}
public Student(StringBuilder courseName, String studentName, Integer score) {
this.courseName = courseName;
this.studentName = studentName;
this.score = score;
}
public StringBuilder getCourseName() {
return courseName;
}
public void setCourseName(StringBuilder courseName) {
this.courseName = courseName;
}
public String getStudentName() {
return studentName;
}
public void setStudentName(String studentName) {
this.studentName = studentName;
}
public Integer getScore() {
return score;
}
public void setScore(Integer score) {
this.score = score;
}
//比较规则
@Override
public int compareTo(Student o) {
return this.studentName.compareTo(o.studentName);
}
// 序列化
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(courseName.toString());
out.writeUTF(studentName);
out.writeInt(score);
}
//反序列化
@Override
public void readFields(DataInput in) throws IOException {
this.courseName = new StringBuilder(in.readUTF());
this.studentName = in.readUTF();
this.score = in.readInt();
}
//通过toString方法自定义输出类型
@Override
public String toString() {
return studentName + ':' + courseName;
}
}
- MapReduce的代码
package Mapreduce.mark6;
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.NullWritable;
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;
//hadoop jar hadoop.jar Mapreduce.mark6.FindCoursesMR hdfs://localhost:9000/user/hadoop/markinput hdfs://localhost:9000/user/hadoop/findoutout6 olive
public class FindCoursesMR {
public static void main(String[] args) throws Exception {
if (args.length<3) {
System.out.printf("Usage:%s <input> <output>\n");
}
Configuration conf = new Configuration();
conf.set("studentName",args[2]);//args[2]是传入的第3个参数
Job job = Job.getInstance(conf);
//设置jar包所在路径
job.setJarByClass(FindCoursesMR.class);
job.setMapperClass(FCMapper.class);
job.setReducerClass(FCReducer.class);
//设置我们的业务逻辑Mapper 类的输出 key 和 value 的数据类型
job.setMapOutputKeyClass(Student.class);
job.setMapOutputValueClass(NullWritable.class);
//这两个方法对map和reduce阶段的输出都能起到作用
job.setOutputKeyClass(Student.class);
job.setOutputValueClass(NullWritable.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);
//boolean waitForCompletion = job.waitForCompletion(true);
// System.out.println("job is finished!");
//System.exit(job.waitForCompletion(true)?0:1);
System.out.println(job.waitForCompletion(true)?1:0);
}
static class FCMapper extends Mapper<LongWritable,Text, Student, NullWritable> {
//map 方法的生命周期: 框架每传一行数据就被调用一次
//key : 这一行的起始点在文件中的偏移量
//value : 这一行的内容
@Override
protected void map(LongWritable key, Text line, Context context) throws IOException, InterruptedException {
String[] fields = line.toString().split(",");
if (fields.length==3){
context.write(new Student(new StringBuilder(fields[0]),fields[1],Integer.parseInt(fields[2])),NullWritable.get());
}
}
}
static class FCReducer extends Reducer<Student,NullWritable,Student,NullWritable> {
private static String studentName = "";
private static StringBuilder stringBuilder = null;
private static Student stu = null;
@Override
protected void setup(Context context) throws IOException, InterruptedException {
//setup是在map执行前加载的方法,读取数据
studentName = context.getConfiguration().get("studentName");
stringBuilder = new StringBuilder();
stu = new Student();
stu.setStudentName(studentName);
}
@Override
protected void reduce(Student key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
if (studentName.equals(key.getStudentName())){
for (NullWritable value : values) {
stringBuilder.append(key.getCourseName() + ","+key.getScore()+";");
}
stu.setCourseName(stringBuilder);
context.write(stu,NullWritable.get());
}
}
}
}
四、程序运行结果
- 程序处理完成之后的结果所保存的位置：hdfs://localhost:9000/user/hadoop/findoutout6（这是自己定义的位置）。
- 程序的执行结果
（可见，得到了我们要查找的学生olive考试的课程和对应的成绩。）
本次的分享就到这里就结束了，这里要感谢老师和某个大佬，才让我可以完成，同时也教会了我很多东西。以上有任何错误，希望可以得到大佬们的指正，互相学习！✨
???? 基于MapReduce的成绩分析系统如果想看更多功能实现的学习体验，欢迎访问：
【基于MapReduce的成绩分析系统】——计算每门课程的平均成绩、最高成绩、最低成绩
【基于MapReduce的成绩分析系统】——计算每门课程学生的平均成绩，并将平均成绩从高到低输出
【基于MapReduce的成绩分析系统】——求该成绩表每门课程当中出现了相同分数的分数，出现的姓名以及该相同分数的人数