二次排序发生的阶段，肯定是在map端，数据从环形缓冲区溢出写入到磁盘之前，这里有一个处理key的方法，非常值得初学者思考，这种组合新key的方法，在实际mapreduce开发中非常值得借鉴。

概述

MapReduce框架对处理结果的输出会根据key值进行默认的排序，这个默认的排序可以满足一部分需求，但是也是十分有限的，在我们实际需求当中，往往有要对reduce输出结果进行二次排序的需求。

需求描述

1.输入数据

sort1   1  
sort2   3  
sort2   88  
sort2   54  
sort1   2  
sort6   22  
sort6   888  
sort6   58

2.目标输出

sort1   1,2  
sort2   3,54,88  
sort6   22,58,888

解决思路

1.首先分析一下，二次排序在mapreduce处理流程的哪个阶段进行的，这样我们就应该先深刻的理解mapreduce处理数据的整个流程，这是最基础的。下面描述一下mapreduce处理数据的大概流程：首先，mapreduce框架通过getSplits()方法实现对原始文件的切分之后，每一个切片对应着一个MapTask，InputSplit输入到map()函数进行处理，中间结果经过环形缓冲区的排序，然后分区，自定义二期排序(如果有的话)和合并，再通过shuffle操作将数据传输到reduce task端，reduce端也存在着缓冲区，数据也会在缓冲区和磁盘中进行合并排序等操作，然后对数据按照key值进行分组，然后每处理完一个分组之后就会去调用一次reduce()函数，最终输出结果。大概流程如下图:

image.png

2.具体解决思路
(1)map端处理
根据上面的需求，我们有一个非常明确的目标就是要对第一列相同的记录，并且对合并后的数字进行排序。我们都知道MapReduce框架不管是默认排序或者是自定义排序都只是对key值进行排序，现在的情况是这些数据不是key值，怎么办？其实我们可以将原始数据的key值和其对应的数据组合成一个新的key值，然后新的key值对应的value还是原始数据中的valu。那么我们就可以将原始数据的map输出变成类似下面的数据结构：

{[sort1,1],1}  
{[sort2,3],3}  
{[sort2,88],88}  
{[sort2,54],54}  
{[sort1,2],2}  
{[sort6,22],22}  
{[sort6,888],888}  
{[sort6,58],58} 

那么我们只需要对[]里面的心key值进行排序就OK了，然后我们需要自定义一个分区处理器，因为我的目标不是想将新key相同的记录传到一个reduce中，而是想将新key中第一个字段相同的记录放到同一个reduce中进行分组合并，所以我们需要根据新key值的第一个字段来自定义一个分区处理器。通过分区操作后，得到的数据流如下：

Partition1：{[sort1,1],1}、{[sort1,2],2}  
  
Partition2：{[sort2,3],3}、{[sort2,88],88}、{[sort2,54],54}  
  
Partition3：{[sort6,22],22}、{[sort6,888],888}、{[sort6,58],58} 

分区操作完成之后，我调用自己的自定义排序器对新的key值进行排序

{[sort1,1],1}  
{[sort1,2],2}  
{[sort2,3],3}  
{[sort2,54],54}  
{[sort2,88],88}  
{[sort6,22],22}  
{[sort6,58],58}  
{[sort6,888],888} 

(2)reduce端处理
经过Shuffle处理之后，数据传输到Reducer端了。在Reducer端按照组合键的第一个字段进行分组，并且每处理完一次分组之后就会调用一次reduce函数来对这个分组进行处理和输出。最终各个分组的数据结果变成类似下面的数据结构：

sort1   1,2  
sort2   3,54,88  
sort6   22,58,888  

代码实现

package com.lucus.secondsort;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * Created by lucus on 2018/3/12 0012.
 */
//自定义组合键
public class CombinationKey implements WritableComparable<CombinationKey> {

    private Text firstKey;
    private IntWritable secondKey;

    //无构造函数
    public CombinationKey() {
        this.firstKey = new Text();
        this.secondKey = new IntWritable();
    }

    //有参构造函数
    public CombinationKey(Text firstKey, IntWritable secondKey) {
        this.firstKey = firstKey;
        this.secondKey = secondKey;
    }

    public Text getFirstKey() {
        return firstKey;
    }

    public void setFirstKey(Text firstKey) {
        this.firstKey = firstKey;
    }

    public IntWritable getSecondKey() {
        return secondKey;
    }

    public void setSecondKey(IntWritable secondKey) {
        this.secondKey = secondKey;
    }


    @Override
    public void write(DataOutput out) throws IOException {
        this.firstKey.write(out);
        this.secondKey.write(out);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.firstKey.readFields(in);
        this.secondKey.readFields(in);
    }

    /**
     * 自定义比较策略
     * 注意：该比较策略用于MapReduce的第一次默认排序
     * 也就是发生在Map端的sort阶段
     * 发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整)
     *
     * **/
    @Override
    public int compareTo(CombinationKey combinationKey) {
        System.out.println("------------------------CombineKey flag-------------------");
        return this.firstKey.compareTo(combinationKey.getFirstKey());
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        CombinationKey other = (CombinationKey) obj;
        if (firstKey == null) {
            if (other.firstKey != null)
                return false;
        } else if (!firstKey.equals(other.firstKey))
            return false;
        return true;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((firstKey == null) ? 0 : firstKey.hashCode());
        return result;
    }
}

package com.lucus.secondsort;

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

/**
 * Created by lucus on 2018/3/12 0012.
 */
//自定义分区
public class DefinedPartition extends Partitioner<CombinationKey, IntWritable> {
    /**
     * 数据输入来源：map输出 我们这里根据组合键的第一个值作为分区
     * 如果不自定义分区的话，MapReduce会根据默认的Hash分区方法
     * 将整个组合键相等的分到一个分区中，这样的话显然不是我们要的效果
     * @param key map输出键值
     * @param value map输出value值
     * @param numPartitions 分区总数，即reduce task个数
     */
    @Override
    public int getPartition(CombinationKey key, IntWritable value, int numPartitions) {
        System.out.println("---------------------进入自定义分区---------------------");
        System.out.println("---------------------结束自定义分区---------------------");
        return (key.getFirstKey().hashCode() & Integer.MAX_VALUE) % numPartitions;
    }
}

package com.lucus.secondsort;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

/**
 * Created by lucus on 2018/3/12 0012.
 */
//自定义比较器
public class DefinedComparator extends WritableComparator {
    protected DefinedComparator() {
        super(CombinationKey.class,true);
    }
    /**
     * 第一列按升序排列，第二列也按升序排列
     */
    public int compare(WritableComparable a, WritableComparable b) {
        System.out.println("------------------进入二次排序-------------------");
        CombinationKey c1 = (CombinationKey) a;
        CombinationKey c2 = (CombinationKey) b;
        int minus = c1.getFirstKey().compareTo(c2.getFirstKey());

        if (minus != 0){
            System.out.println("------------------结束二次排序-------------------");
            return minus;
        } else {
            System.out.println("------------------结束二次排序-------------------");
            return c1.getSecondKey().get() -c2.getSecondKey().get();
        }
    }
}

package com.lucus.secondsort;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

/**
 * Created by lucus on 2018/3/12 0012.
 */
//自定义分组有中方式，一种是继承WritableComparator
//另外一种是实现RawComparator接口
public class DefinedGroupSort extends WritableComparator {

    protected DefinedGroupSort() {
        super(CombinationKey.class,true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        System.out.println("---------------------进入自定义分组---------------------");
        CombinationKey combinationKey1 = (CombinationKey) a;
        CombinationKey combinationKey2 = (CombinationKey) b;
        System.out.println("---------------------分组结果：" + combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey()));
        System.out.println("---------------------结束自定义分组---------------------");
        //自定义按原始数据中第一个key分组
        return combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey());
    }

}

package com.lucus.secondsort;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.net.URI;

/**
 * Created by lucus on 2018/3/12 0012.
 */
public class SecondSortMapReduce {

    public static void main(String[] args)  throws Exception {
        Configuration conf = new Configuration();
        conf.set("mapreduce.app-submission.cross-platform", "true");
        conf.set("mapreduce.framework.name", "yarn");
        conf.set("mapreduce.job.jar","D:\\workspace\\SecondSortMapReduce\\target\\SecondSortMapReduce-1.0-SNAPSHOT-jar-with-dependencies.jar");
        // 设置输入输出文件目录
        String[] ioArgs = new String[] { "hdfs://rsct0:8020/input/secondsort.txt", "hdfs://rsct0:8020/output/secondsort" };
        String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage:  <in> <out>");
            System.exit(2);
        }

        //创建文件系统
        FileSystem fileSystem = FileSystem.get(new URI(otherArgs[1]),conf);
        //如果输入目录文件存在，就删除
        if(fileSystem.exists(new Path(otherArgs[1]))){
            fileSystem.delete(new Path(otherArgs[1]),true);
        }
        //创建任务
        Job job = Job.getInstance(conf);

        //1.1设置输入目录和设置输入数据格式化的类
        FileInputFormat.setInputPaths(job,new Path(otherArgs[0]));
        job.setInputFormatClass(KeyValueTextInputFormat.class);

        //1.2设置自定义Mapper类和设置map函数输出数据的key和value的类型
        job.setMapperClass(SecondSortMapper.class);
        job.setMapOutputKeyClass(CombinationKey.class);
        job.setMapOutputValueClass(IntWritable.class);

        //1.3设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)
        job.setPartitionerClass(DefinedPartition.class);
        job.setNumReduceTasks(1);

        //设置自定义分组策略
        job.setGroupingComparatorClass(DefinedGroupSort.class);
        //设置自定义比较策略(因为我的CombineKey重写了compareTo方法，所以这个可以省略)
        job.setSortComparatorClass(DefinedComparator.class);

        //1.4   排序
        //1.5   归约
        //2.1   Shuffle把数据从map端拷贝到reduce端
        //2.2   指定Reducer类和输出key和value的类型
        job.setReducerClass(SecondSortReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //2.3   指定输出的路径和设置输出的格式化类
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        job.setOutputFormatClass(TextOutputFormat.class);

        //提交作业 退出
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class SecondSortMapper extends Mapper<Text, Text, CombinationKey, IntWritable> {
        /**
         * 这里要特殊说明一下，为什么要将这些变量写在map函数外边
         * 对于分布式的程序，我们一定要注意到内存的使用情况，对于MapReduce框架
         * 每一行的原始记录的处理都要调用一次map()函数，假设，这个map()函数要处理1一亿
         * 条输入记录，如果将这些变量都定义在map函数里面则会导致这4个变量的对象句柄
         * 非常的多(极端情况下将产生4*1亿个句柄，当然java也是有自动的GC机制的，一定不会达到这么多)
         * 导致栈内存被浪费掉，我们将其写在map函数外面，顶多就只有4个对象句柄
         */
        private CombinationKey combinationKey = new CombinationKey();
        Text sortName = new Text();
        IntWritable score = new IntWritable();
        String[] splits = null;

        @Override
        protected void map(Text key, Text value, Mapper<Text, Text, CombinationKey, IntWritable>.Context context)
                throws IOException, InterruptedException {
            System.out.println("---------------------进入map()函数---------------------");
            if (key == null || value == null || key.toString().equals("")){
                return;
            }
            //构造相关属性
            sortName.set(key.toString());
            score.set(Integer.parseInt(value.toString()));
            //设置联合key
            combinationKey.setFirstKey(sortName);
            combinationKey.setSecondKey(score);

            //通过context把map处理后的结果输出
            context.write(combinationKey, score);
            System.out.println("---------------------结束map()函数---------------------");
        }
    }

    public static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text> {
        StringBuffer sb = new StringBuffer();
        Text score = new Text();
        /**
         * 这里要注意一下reduce的调用时机和次数：
         * reduce每次处理一个分组的时候会调用一次reduce函数。
         * 所谓的分组就是将相同的key对应的value放在一个集合中
         * 例如：<sort1,1> <sort1,2>
         * 分组后的结果就是
         * <sort1,{1,2}>这个分组会调用一次reduce函数
         */
        @Override
        protected void reduce(CombinationKey key, Iterable<IntWritable> values, Reducer<CombinationKey, IntWritable, Text, Text>.Context context)
                throws IOException, InterruptedException {
            //先清除上一个组的数据
            sb.delete(0,sb.length());

            for(IntWritable val : values){
                sb.append(val.get() + ",");
            }

            //取出最后一个逗号
            if(sb.length() > 0) {
                sb.deleteCharAt(sb.length() - 1);
            }

            //设置写出去的value
            score.set(sb.toString());

            //将联合key的第一个元素作为新的key，将score作为value写出去
            context.write(key.getFirstKey(),score);

            System.out.println("---------------------进入reduce()函数---------------------");
            System.out.println("---------------------{[" + key.getFirstKey()+"," + key.getSecondKey() + "],[" +score +"]}");
            System.out.println("---------------------结束reduce()函数---------------------");
        }
    }

}