深入理解hadoop之排序

　　MapReduce的排序是默认按照Key排序的，也就是说输出的时候，key会按照大小或字典顺序来输出，比如一个简单的wordcount，出现的结果也会是左侧的字母按照字典顺序排列。下面我们主要聊聊面试中比较常见的全排序和二次排序

一、全排序

　　全排序的方法一般有以下几种：

　　　　1.使用一个分区。但是该方法在处理大型文件的时候效率极低，因为一台机器必须处理所有的输出文件，从而丧失了mapreduce提供的并行架构的优势。这个比较简单，只要在APP中设置分区数量为1就可以了。

　　　　2.自定义分区函数，自行设置分解区间。这个方法最关键的地方在于如何划分各分区，如果数据分布不均匀，分区函数设置不恰当，最后会产生数据倾斜。这个地方请看下面统计历年最高气温的例子。

　　　　气温数据：

　　　　Map端

package com.heima.hdfs.mr3;

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;

/**
 * Map实现对气温文本进行切割，输入key是偏移量，输入map是text,输出的key值是年份，输出的value是气温
 * 简单的对文本文档进行切割，
 */
public class MaxTempMapper extends Mapper<LongWritable ,Text,IntWritable,IntWritable>{
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] arr =value.toString().split(" ");
       context.write(new IntWritable(Integer.parseInt(arr[0])),new IntWritable(Integer.parseInt(arr[1])));

    }
}

　　　　Reduce端

package com.heima.hdfs.mr3;

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

import java.io.IOException;

/**
 * 这个地方要注意：相同的key值会进入同一个分区，同一个分区里的数据会进入同一个reduce里面
 */
public class MaxTempReducer extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {
    @Override
    protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int max = Integer.MIN_VALUE;
        for(IntWritable iw:values){
            max=max>iw.get()?max:iw.get();
        }
        context.write(key,new IntWritable(max));
     }
}

　　App端

package com.heima.hdfs.mr3;

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

/**
 * Created by Administrator on 2018/7/5 0005.
 */
public class MaxTempApp {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","file:///");
        Job job = Job.getInstance(conf);
        job.setJobName("MaxTempApp");
        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        job.setNumReduceTasks(3);
        job.setPartitionerClass(YearPartitioner.class);
        job.setJarByClass(MaxTempApp.class);
        job.setMapperClass(MaxTempMapper.class);
        job.setReducerClass(MaxTempReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.waitForCompletion(true);
    }
}

　　　　3.通过hadoop采样机制，对键空间进行采样，较为均匀的划分数据集，采样的核心思想是只查看一小部分键，获得键的近似分布，由此构建分区，在hadoop中已经自带了采样器，不需要开发人员自己编写

　　　　Map端

package com.heima.hdfs.allsort;

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

import java.io.IOException;

/**
 * Created by Administrator on 2018/7/5 0005.
 */
public class MaxTempMapper extends Mapper<IntWritable,IntWritable,IntWritable,IntWritable> {
    @Override
    protected void map(IntWritable key, IntWritable value, Context context) throws IOException, InterruptedException {
        context.write(key,value);
    }
}

　　Reduce端

package com.heima.hdfs.allsort;

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

import java.io.IOException;

/**
 * Created by Administrator on 2018/7/5 0005.
 */
public class MaxTempReducer extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {
    @Override
    protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int max = Integer.MIN_VALUE;
        for(IntWritable iw :values){
            max = max>iw.get()?max:iw.get();
        }
        context.write(key,new IntWritable(max));
    }
}

App端

package com.heima.hdfs.allsort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

import java.io.IOException;

/**
 * Created by Administrator on 2018/7/5 0005.
 */
public class MaxTempApp {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","file:///");
        Job job = Job.getInstance(conf);
        job.setJobName("MaxTempApp");
        job.setNumReduceTasks(3);
        job.setInputFormatClass(SequenceFileInputFormat.class);
        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        job.setJarByClass(MaxTempApp.class);
        job.setMapperClass(MaxTempMapper.class);
        job.setReducerClass(MaxTempReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        //设置全排序分区
        job.setPartitionerClass(TotalOrderPartitioner.class);
        //创建采样器这里概率是1，6000个key会全部取出来
        InputSampler.Sampler<IntWritable,IntWritable> sampler =new InputSampler.RandomSampler<IntWritable,IntWritable>(1,100000,3);
        TotalOrderPartitioner.setPartitionFile(job.getConfiguration(),new Path("e:/mr/tmp/par.lst"));
        InputSampler.writePartitionFile(job,sampler);
        job.waitForCompletion(true);
    }
}

Map起始阶段

在Map阶段，使用job.setInputFormatClass()定义的InputFormat，将输入的数据集分割成小数据块split，同时InputFormat提供一个RecordReader的实现。在这里我们使用的是TextInputFormat，它提供的RecordReader会将文本的行号作为Key，这一行的文本作为Value。这就是自定 Mapper的输入是<LongWritable,Text> 的原因。然后调用自定义Mapper的map方法，将一个个<LongWritable,Text>键值对输入给Mapper的map方法

Map最后阶段

在Map阶段的最后，会先调用job.setPartitionerClass()对这个Mapper的输出结果进行分区，每个分区映射到一个Reducer。每个分区内又调用job.setSortComparatorClass()设置的Key比较函数类排序。可以看到，这本身就是一个二次排序。如果没有通过job.setSortComparatorClass()设置 Key比较函数类，则使用Key实现的compareTo()方法

Reduce阶段

在Reduce阶段，reduce()方法接受所有映射到这个Reduce的map输出后，也会调用job.setSortComparatorClass()方法设置的Key比较函数类，对所有数据进行排序。然后开始构造一个Key对应的Value迭代器。这时就要用到分组，使用 job.setGroupingComparatorClass()方法设置分组函数类。只要这个比较器比较的两个Key相同，它们就属于同一组，它们的 Value放在一个Value迭代器，而这个迭代器的Key使用属于同一个组的所有Key的第一个Key。最后就是进入Reducer的 reduce()方法，reduce()方法的输入是所有的Key和它的Value迭代器，同样注意输入与输出的类型必须与自定义的Reducer中声明的一致

二、二次排序

　　我们都知道map端的输出结果经过partition()分区函数之后会对key值进行排序，经过shuffle阶段之后，向相同的key值会进入到同一个分组中去，也就是说key的排序是有序的，但是有时候需要对Key排序的同时还需要对Value进行排序，比如上面求每年最高气温的案例时，这时候就要用到二次排序了。经过本人的理解，二次排序可以大致分为以下几个阶段。

Map起始阶段

Map最后阶段

Reduce阶段

　　排序的案例仍然为上述求取每年最高气温的案例

二次排序的具体流程

在本例中要比较两次。先按照第一字段排序，然后再对第一字段相同的按照第二字段排序。根据这一点，我们可以构造一个复合类key，它有两个字段，先利用分区对第一字段排序，再利用分区内的比较对第二字段排序。二次排序的流程分为以下几步。

1、自定义 key

所有自定义的组合key应该实现接口WritableComparable，WritableComparable接口继承自writable和comparable这两个接，口因为writable接口是可序列化的并且可比较的。WritableComparable。组合key按照年份升序按照气温降序，实现的代码如下

public class Combokey implements WritableComparable<Combokey> {
    private int year ;

    public int getYear() {
        return year;
    }

    public void setYear(int year) {
        this.year = year;
    }

    public int getTemp() {
        return temp;
    }

    public void setTemp(int temp) {
        this.temp = temp;
    }

    private int temp;
    /*
    * 对key进行比较实现
    * */
    @Override
    public int compareTo(Combokey o) {
        System.out.println("Combokey.compareTo()"+o.toString());
         int y0 =o.getYear();
        int t0=o.getTemp();
        //年份相同(s升序)
        if(year==y0){
            //气温降序
            return -(temp-t0);
        }else{
            return (year-y0);
        }
    }
    /*
    * 串行化过程
    * */
    @Override
    public void write(DataOutput out) throws IOException {
        //年份
        out.writeInt(year);
        //气温
        out.writeInt(temp);
    }
    //反串行化的过程
    @Override
    public void readFields(DataInput in) throws IOException {
        year = in.readInt();
        temp = in.readInt();
    }
    public  String toString(){
        return  year+":"+temp;
    }
}

　2.自定义分区

自定义分区函数类FirstPartitioner，是key的第一次比较，完成对所有key的排序。该分区类按照年份进行分区，相同的年份会进入到同一个分区中去。

public class YearPartitioner extends Partitioner<Combokey,NullWritable>{
    @Override
    public int getPartition(Combokey key, NullWritable nullWritable, int numPartitions) {
        System.out.println("YearPartitioner.getPartition"+key);
        int year = key.getYear();
        return  year%numPartitions;
    }
}

3、Key的比较类CombokeyComparator

这是Key的第二次比较，这个类继承自WirtableComparator这个类，对所有的Key进行排序，即同时完成Combokey中的first和second排序。

public class CombokeyComparator extends WritableComparator{
    protected CombokeyComparator(){
        super(Combokey.class,true);
    }
    public int compare(WritableComparable a,WritableComparable b){
        System.out.println("CombokeyComparator"+a+","+b);
        Combokey k1 = (Combokey)a;
        Combokey k2 = (Combokey)b;
        return k1.compareTo(k2);
    }
}

4、定义分组类函数YearGroupComparator

在Reduce阶段，构造一个与 Key 相对应的 Value 迭代器的时候，只要year相同就属于同一个组，放在一个Value迭代器,不同的year按照年份升序进行排序。

public class YearGroupComparator extends WritableComparator{
    protected YearGroupComparator(){
        super(Combokey.class,true);
    }
    public int compare(WritableComparable a,WritableComparable b){
        System.out.println("YearGroupComparator"+a+","+b);
        Combokey key1 = (Combokey)a;
        Combokey key2 = (Combokey)b;
        return  key1.getYear()-key2.getYear();
    }
}

5.Map端，输入的(key,value)缩进长度和文本文档，输出的key是组合key,value值是控值

public class MaxTempMapper extends Mapper<LongWritable,Text,Combokey,NullWritable>{
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        System.out.println("MaxTempMapper.map");
        String[] arr = value.toString().split(" ");
        Combokey keyout = new Combokey();
        keyout.setYear(Integer.parseInt(arr[0]));
        keyout.setTemp(Integer.parseInt(arr[1]));
        context.write(keyout,NullWritable.get());
    }
}

6.Reduce端，将组合key切割成key为year，value为气温的一个列表

public class MaxTempReducer extends Reducer<Combokey,NullWritable,IntWritable,IntWritable> {
    @Override
    protected void reduce(Combokey key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        int year = key.getYear();
        int temp = key.getTemp();
        System.out.println("MaxTempReducer.reduce"+year+","+temp);
        context.write(new IntWritable(year),new IntWritable(temp));
    }
}

7.APP端

public class MaxTempApp {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","file:///");
        Job job = Job.getInstance(conf);
        job.setJobName("MaxTempApp");
        FileInputFormat.addInputPath(job,new Path("e:/mr/tmp/1.txt"));
        FileOutputFormat.setOutputPath(job,new Path("e:/mr/tmp/out"));
        job.setJarByClass(MaxTempApp.class);
        //设置Map类
        job.setMapperClass(MaxTempMapper.class);
        //设置Reduce类
        job.setReducerClass(MaxTempReducer.class);
        //设置Map输出类型
        job.setMapOutputKeyClass(Combokey.class);
        job.setMapOutputValueClass(NullWritable.class);
        //设置reduce输出类型
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        //设置分区类
        job.setPartitionerClass(YearPartitioner.class);
        //设置分组对比器
        job.setGroupingComparatorClass(YearGroupComparator.class);
        //设置排序对比器
        job.setSortComparatorClass(CombokeyComparator.class);
        job.setNumReduceTasks(3);
        job.waitForCompletion(true);
    }
}