使用Hadoop MapReduce计算成绩平均值

1.介绍

Hadoop MapReduce是一个开源的分布式计算框架，被广泛应用于处理海量数据集。在本文中，我们将深入探讨如何利用Hadoop MapReduce计算学生科目成绩的平均值。通过编写Map和Reduce两个类，我们能够实现这一功能，进而加深对大数据处理的理解和应用。这个过程将涉及到数据的切分、映射、排序、归约等多个步骤，展现了Hadoop在处理复杂计算任务时的强大性能和可扩展性。通过本文的学习，读者将能够更加熟练地运用Hadoop MapReduce框架处理实际场景中的数据分析和计算问题，为大数据应用的开发提供有力支持。

2.数据准备

本次博文的数据集路劲为hdfs上的/scoreinput/subject_score.csv，数据集内容大致如下：

一行包含了学生的学号、各科科目和相应的科目成绩。这些数据可以在Hadoop MapReduce框架下进行处理和分析，以获取有关学生成绩的更多信息，通过使用MapReduce框架，我们可以首先将数据集划分为多个小数据块，然后将这些小数据块分发到不同的计算节点上进行并行计算。

3.代码思路

3.1

在Map阶段，我们可以将每一行数据映射为(key, value)键值对。其中，学号可以作为键，科目和成绩组成的元组可以作为值。这样，每个学生的所有科目成绩都会被映射为一组键值对。

3.2

在Reduce阶段，我们可以对映射得到的键值对进行聚合和计算。例如，我们可以对每个学生的成绩进行求和，并计算平均成绩。这可以通过将学号作为键，将成绩作为值，然后在Reduce函数中对值进行累加并计数，最后求得平均值来实现。

通过使用Hadoop MapReduce框架，我们可以对学生的成绩数据集进行多种分析和处理，从而获得对学生学业表现的整体了解，并从中发现有价值的信息。这些信息可以为学校、教师和教育决策者提供数据支持和指导，帮助他们做出更准确的评估和决策。

4.代码实现

4.1mapper类

我们需要创建一个Mapper类，命名为ScoreMapper。该类继承自Mapper类，并重写map方法。map方法的输入参数是LongWritable、Text类型的key-value对，表示输入数据的偏移量和内容。map方法的输出参数是Text、IntWritable类型的key-value对，表示输出的键值对。在map方法中，我们首先将Text类型的value转换为字符串，并进行分割，得到学生姓名和成绩。然后，我们将姓名作为输出的key，成绩作为输出的value，通过context对象将它们写入输出流。

package com.atguigu.Score;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

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

public class ScoreMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        String line = new String(value.getBytes(),0,value.getLength(),"UTF-8");

        String str = "";
        String[] words = line.split(",");
        for (int i = 1;i<words.length;i++) {
            str+=words[i]+" ";
        }
        StringTokenizer itr = new StringTokenizer(str);

        String strName = itr.nextToken();
        String strScore = itr.nextToken();
        Text name = new Text(strName);
        int scoreInt = Integer.parseInt(strScore);
        context.write(name,new IntWritable(scoreInt));
    }
}

4.2 reduce类

我们需要创建一个Reducer类，命名为ScoreReduce。该类继承自Reducer类，并重写reduce方法。reduce方法的输入参数是Text、Iterable<IntWritable>类型的key-value对，表示相同科目的成绩集合。reduce方法的输出参数是Text、IntWritable类型的key-value对，表示输出的键值对。在reduce方法中，我们遍历输入的成绩集合，累加成绩并计数，最后求得平均值。然后，我们将科目名称作为输出的key，平均值作为输出的value，通过context对象将它们写入输出流。

package com.atguigu.Score;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

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

public class ScoreReduce extends Reducer<Text, IntWritable,Text,IntWritable> {
    @Override
    //value才是获取的一行数据，其中的key是数据的偏移量或者行数，Mapper后面的表示 声明Mapper 类的参数和上下文对象
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.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));
    }
}

4.3 Driver类

我们需要创建一个驱动类，命名为ScoreDriver。在main方法中，我们首先设置Hadoop的配置信息，包括用户名、文件系统等。然后，创建一个Job对象，并设置作业的名称和Jar包路径。接着，通过setMapperClass和setReducerClass方法，将Mapper类和Reducer类与Job关联起来。然后，我们设置Mapper的输出键值对类型和Reducer的输出键值对类型。接着，我们设置输入格式和输出格式。最后，设置输入路径和输出路径，并调用job.waitForCompletion方法提交作业。

package com.atguigu.Score;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.junit.Before;

import java.awt.peer.TextAreaPeer;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class ScoreDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        System.setProperty("HADOOP_USER_NAME", "wuhui");
        Configuration conf = new Configuration();
        conf.set("mapreduce.job.user.name", "wuhui");
        conf.set("fs.default","hdfs://192.168.88.110:8020");
        //获取一个job对象，在job对象中读取Hadoop的配置文件，jobName是该作业的名字
        Job job = Job.getInstance(conf,"Score Average");
        //获取jar包的路径
        job.setJarByClass(ScoreDriver.class);
        //对mapper和reduce进行关联
        job.setMapperClass(ScoreMapper.class);
        job.setReducerClass(ScoreReduce.class);
        //mapper的输出的kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //最终输出的kv类型
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        //设置输入和输出路径
        FileInputFormat.addInputPath(job,new Path("hdfs://192.168.88.110:8020/scoreinput/"));
        FileOutputFormat.setOutputPath(job,new Path("hdfs://192.168.88.110:8020/scoreoutput1/"));
        //判断作业是否完成，完成会传入一个true给函数，对此作业的进程进行堵塞，也就是程序执行完毕，反之则继续执行
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

注意：上述Driver类中我的hadoop用户名为wuhui，想借鉴我的代码的小伙伴可将hadoop用户名和ip改为自己的即可

5.代码结果展示：

在上述Driver代码中，我将输出结果保存到了hdfs上的/scoreinput目录中，在图中框起来的文件中就是mapredcue代码的输出结果

使用hdfs dfs -cat /scoreoutput/part-r-00000命令可查看目标文件内容

6.总结

通过以上步骤，我们成功编写了一个使用Hadoop MapReduce计算各科目成绩平均值的程序，并将打印结果输出到hdfs路径中。通过这种方式，我们可以方便地处理大规模的科目成绩数据，并计算出平均值。

希望本篇博客对小伙伴们有所帮助！如果有任何问题或建议，请随时提出。谢谢！