MapReduce快速入门系列(8) | Shuffle之排序(sort)——区内排序

不温卜火

发布于 2020-10-28 15:29:07

3360

发布于 2020-10-28 15:29:07

文章被收录于专栏：不温卜火

上一篇博文讲了Shuffle排序的相关概念以及全排序的操作，这篇博文继续分享的是排序的另一种操作：区内排序。

一. 需求分析

基于前一个需求，增加自定义分区类，分区按照省份手机号设置。

1. 把原数据排序后

2. 期望数据输出

二. 代码实现

2.1 增加自定义分区类MyPartitioner2

package com.buwenbuhuo.WritableComparable2;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * @author 卜温不火
 * @create 2020-04-24 18:14
 * com.buwenbuhuo.WritableComparable2 - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class MyPartitioner2 extends Partitioner<com.buwenbuhuo.WritableComparable.FlowBean, Text> {
    @Override
    public int getPartition(com.buwenbuhuo.WritableComparable.FlowBean flowBean, Text text, int numPartitions) {
        switch (text.toString().substring(0, 3)) {
            case "136":
                return 0;
            case "137":
                return 1;
            case "138":
                return 2;
            case "139":
                return 3;
            default:
                return 4;
        }
    }
}

2.2 在驱动类中添加分区类

// 加载自定义分区类
job.setPartitionerClass(ProvincePartitioner.class);

// 设置Reducetask个数
job.setNumReduceTasks(5);

此部分的完整代码如下:

package com.buwenbuhuo.WritableComparable2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;

/**
 * @author 卜温不火
 * @create 2020-04-24 18:19
 * com.buwenbuhuo.WritableComparable2 - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class SortDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());

        job.setJarByClass(com.buwenbuhuo.WritableComparable2.SortDriver.class);
        job.setMapperClass(com.buwenbuhuo.WritableComparable.SortMapper.class);
        job.setReducerClass(com.buwenbuhuo.WritableComparable.SortReducer.class);

        job.setMapOutputKeyClass(com.buwenbuhuo.WritableComparable.FlowBean.class);
        job.setMapOutputValueClass(Text.class);

        job.setPartitionerClass(MyPartitioner2.class);
        job.setNumReduceTasks(5);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(com.buwenbuhuo.WritableComparable.FlowBean.class);

        FileInputFormat.setInputPaths(job, new Path("d:\\output"));
        FileOutputFormat.setOutputPath(job, new Path("d:\\output2"));

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}