本文共 4058 字，大约阅读时间需要 13 分钟。

MapReduce Java API 中的 Partitioner 实现：按成绩分段输出到不同文件

场景

在 MapReduce Java API 中，Partitioner 是一个强大的工具，可以帮助我们将大数据集分成多个独立的分区，从而实现多输入路径的方式。例如，我们可以根据学生的成绩将数据分成三个分区：小于60分、大于等于60分小于等于80分，以及大于80分。

Partitioner 的作用

Partitioner 在 MapReduce 作业中扮演着关键角色。它的主要职责包括：

Partitioner 的创建流程

要实现分区功能，我们需要自定义一个 Partitioner 类。以下是具体的步骤说明：

实现步骤

1. 创建数据集

首先，我们需要准备一个数据集文件（例如 score.txt），文件内容应包含学生的成绩数据。

100 85 90 55 70 60

2. 自定义 Partitioner 类

编写一个自定义的 Partitioner 类，用于根据成绩分段。

package com.badao.muloutput;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;public class StudentPartitioner extends Partitioner
   
     {    @Override    public int getPartition(IntWritable intWritable, Text text, int i) {        int score = intWritable.get();        if (score < 60) {            return 0;        } else if (score <= 80) {            return 1;        } else {            return 2;        }    }}
   

3. 定义 Mapper 类

编写一个 Mapper 类，用于处理输入数据。

package com.badao.muloutput;import org.apache.commons.lang.StringUtils;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;public class MulOutputMapper extends Mapper
   
     {    @Override    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {        String[] studentArr = value.toString().split(" ");        if (StringUtils.isNotBlank(studentArr[1])) {            int score = Integer.parseInt(studentArr[1].trim());            context.write(new IntWritable(score), value);        }    }}
   

4. 定义 Reducer 类

编写一个 Reducer 类，用于处理每个分区的数据。

package com.badao.muloutput;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class MulOutputReducer extends Reducer
   
     {    @Override    public void reduce(IntWritable key, Iterable
    
      values, Context context) throws IOException, InterruptedException {        for (Text value : values) {            context.write(NullWritable.get(), value);        }    }}
    
   

5. 创建 Job 类

编写一个 Job 类，配置 MapReduce 作业。

package com.badao.muloutput;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.NullWritable;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.output.FileOutputFormat;import java.io.IOException;public class MulOutputJob {    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {        // 创建配置        Configuration conf = new Configuration();        conf.set("fs.defaultFS", "hdfs://192.168.148.128:9000");                // 创建作业        Job job = Job.getInstance(conf, "muloutput");        job.setJarByClass(MulOutputJob.class);        job.setMapperClass(MulOutputMapper.class);        job.setReducerClass(MulOutputReducer.class);        job.setPartitionerClass(StudentPartitioner.class);        job.setNumReduceTasks(3); // 设置 Reduce 的数量                // 设置输入路径        FileInputFormat.addInputPath(job, new Path("/score.txt"));        // 设置输出路径        FileOutputFormat.setOutputPath(job, new Path("/muloutput"));                // 等待作业完成        job.waitForCompletion(true);    }}

注意事项

通过以上步骤，我们可以实现 MapReduce Java API 中的多输入路径方式，将学生的成绩数据按指定分段输出到不同文件中。