MapReduce实现二次排序

àì夳堔傛蜴生んèń 2022-12-05 10:12 155阅读 0赞

默认情况下，Map输出的结果会按照key进行排序，但在实际的应用中，有时间我们不仅要对key进行排序，同时还要对value进行排序，这时候就要用到mapreduce中的二次排序。

## **一、 二次排序原理分析** ##

我们对二次排序分为以下几个阶段:

**（1）Map起始阶段**

在Map阶段，通过job.setInputFormatClass()定义的InputFormat，将输入的数据集切分成小的split，同时InputFormat提供一个RecordReader的实现。RecordReader会将文本的行号最为Key，该行的内容作为Value，这也是Map的输入数据类型为<LongWritable, Text>的原因。接着会调用自定义的map方法，将读取到的<key, value>键值对输入给map方法。

**（2）Map的最后阶段**

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

**（3）Reducer阶段**

在Reducer阶段，自定义的reduce方法接收所有映射到这个reduce的map的输出，之后调用job.setSortComparatorClass()方法设置的key比较函数类，对所有数据进行排序。然后开始构造一个key对应的Value迭代器，这里就要用到分组，调用job.setGroupingComparatorClass()方法设置分组函数类，只要这个比较器比较的两个key相同，他们就属于同一个组，他们的value放在一个value迭代器中，而这个迭代器的key使用属于同一个的所有key的第一个key。最后进入Reducer的reduce方法，该方法的输入是所有的key和它的value迭代器。

## **二、二次排序样例** ##

假设要对如下文件进行排序，首先按照第一列字母进行排序，然后对每个字母所对应的第二列数字按大小进行排序。

a 5
    a 3
    b 2
    c 9
    b 10
    a 1
    c 8
    b 6

排序结果如下：

a  1
    a  3
    a  5
    b  2
    b  6
    b  10
    c  8
    c  9

**（1）自定义key**

所有自定义的key应该实现接口WritableComparable，因为它是可序列化的且可比较。实现该接口必须要实现如下函数：

// 序列化，将 IntPair 序列化成二进制流
    public void write(DataOutput dataOutput) throws IOException
    
    // 反序列化，将读取的二进制流转换成 IntPair
    public void readFields(DataInput dataInput) throws IOException
    
    // key 的比较
    public int compareTo(IntPair o)
    
    // 默认的分区类 HashPartitioner， 使用此方法
    public int hashCode()

**（2）自定义分区**

自定义分区类FirstPartitioner，是key的第一次比较，完成对所有的key排序

public class FirstPartitioner extends Partitioner<IntPair, IntWritable>

在job中使用setPartitionerClass()方法设置Partitioner

job.setPartitionerClass(FirstPartitioner.class);

**（3）key的比较类**

这是key的第二次比较，对所有的key进行排序，即同时完成IntPair中的first和second排序，可以通过如下两种方式实现：

*  继承WritableComparator

public static class KeyComparator extends WritableComparator

注意，必须要有一个构造函数，并且重载如下方法

public int compare(WritableComparable w1, WritableComparable w2);

*  实现接口RawComparator

以上两种实现方式，均通过job.setSortComparatorClass()方法来设置Key的比较类

job.setSortComparatorClass(KeyComparator.Class);

注：如果没有使用自定义的SortComparator类，则默认使用Key中comareTo()方法对key排序。

**（4）自定义分组类函数**

在reduce阶段，构造一个与key相对应的value迭代器时只要first相同就属于同一个组，放在同一个value迭代器中，实现fail分组类有一下两种方式：

*  继承WritableComparator

public class GroupingComparator extends WritableComparator

注意，必须要有一个构造函数，并且重载如下方法

public int compare(WritableComparable w1, WritableComparable w2)

*  实现接口RawComparator

以上两种实现方式，均通过job.setGroupingComparatorClass()方法来设置分组类

job.setGroupingComparatorClass(GroupingComparator.class);

## 举例 ##

具体代码如下：

/** IntPair.java */
    
    package secondarysort;
    
    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;
    
    
    /**
     * @author lyf 
     * @since 2018/12/01
     */
    public class IntPair implements WritableComparable<IntPair> {
    
        private Text first;
        private IntWritable second;
    
        public IntPair(){
            first = new Text();
            second = new IntWritable(0);
        }
    
        /**
         * key 的比较
         * @param o
         * @return
         */
        @Override
        public int compareTo(IntPair o) {
            int flag = first.compareTo(o.first);
            if (flag == 0) {
                flag = second.compareTo(o.second);
            }
            return flag;
        }
    
        /**
         * 默认的分区类 HashPartitioner， 使用此方法
         * @return
         */
        @Override
        public int hashCode() {
            return first.hashCode();
        }
    
        @Override
        public String toString() {
            return getFirst() + "|" + getSecond();
        }
    
        /**
         * 序列化， 将 IntPair 序列化成二进制流
         * @param dataOutput
         * @throws IOException
         */
        @Override
        public void write(DataOutput dataOutput) throws IOException {
            first.write(dataOutput);
            second.write(dataOutput);
        }
    
        /**
         * 反序列化，将二进制流反序列化成 IntPair
         * @param dataInput
         * @throws IOException
         */
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            first.readFields(dataInput);
            second.readFields(dataInput);
        }
    
        public void setFirst(Text first) {
            this.first = first;
        }
    
        public Text getFirst() {
            return first;
        }
    
        public void setSecond(IntWritable second) {
            this.second = second;
        }
    
        public IntWritable getSecond() {
            return second;
        }
    }

/** FirstParatitioner.java */
    
    package secondarysort;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    
    /**
     * @author lyf 
     * @since 2018/12/01
     */
    public class FirstPartitioner extends Partitioner<IntPair, IntWritable> {
    
        @Override
        public int getPartition(IntPair intPair, IntWritable intWritable, int i) {
            return (intPair.getFirst().hashCode() & Integer.MAX_VALUE) % i;
        }
    }

/** GroupingComparator.java */
    
    package secondarysort;
    
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    /**
     * @author lyg
     * @since 2018/12/01
     */
    public class GroupingComparator extends WritableComparator {
    
        public GroupingComparator(){ super(IntPair.class, true);}
    
        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            IntPair a1 = (IntPair)a;
            IntPair b1 = (IntPair)b;
            String first1 = String.valueOf(a1.getFirst());
            String second = String.valueOf(b1.getFirst());
    
            return first1.compareTo(second);
        }
    }

/** SecondarySortMapper.java */
    
    package secondarysort;
    
    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;
    
    
    /**
     * @author lyf
     * @since 2018/12/01
     */
    public class SecondarySortMapper extends Mapper<LongWritable, Text, IntPair, IntWritable> {
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            String first;
            int second = 0;
            if (tokenizer.hasMoreTokens()) {
                first = tokenizer.nextToken();
                if (tokenizer.hasMoreTokens()) {
                    second = Integer.parseInt(tokenizer.nextToken());
                }
    
                IntPair intPair = new IntPair();
                intPair.setFirst(new Text(first));
                intPair.setSecond(new IntWritable(second));
    
                context.write(intPair, new IntWritable(second));
            }
        }
    }

/** SecondarySortReducer.java */
    
    package secondarysort;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    
    /**
     * @author lyf
     * @since 2018/12/01
     */
    public class SecondarySortReducer extends Reducer<IntPair, IntWritable, Text, IntWritable> {
    
        @Override
        protected void reduce(IntPair key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    
            for (IntWritable val : values){
                context.write(new Text(String.valueOf(key.getFirst())), val);
            }
    
        }
    }

/** SecondarySortJob.java */
    
    package secondarysort;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    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.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    
    /**
     * @author lyf
     * @since 2018/12/01
     */
    public class SecondarySortJob extends Configured implements Tool {
    
        @Override
        public int run(String[] args) throws Exception {
    
            Configuration conf = getConf();
    
    //        String input_path = args[0];
    //        String output_path = args[1];
            String input_path = "input_data";
            String output_path = "output_data";
    
            Path out_path = new Path(output_path);
            FileSystem fileSystem = out_path.getFileSystem(conf);
            if (fileSystem.exists(out_path)) {
                fileSystem.delete(out_path, true);
            }
    
            Job job = Job.getInstance(conf);
    
            job.setJarByClass(SecondarySortJob.class);
            job.setMapperClass(SecondarySortMapper.class);
            job.setReducerClass(SecondarySortReducer.class);
            job.setNumReduceTasks(1);
    
            job.setMapOutputKeyClass(IntPair.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            job.setPartitionerClass(FirstPartitioner.class);
            job.setGroupingComparatorClass(GroupingComparator.class);
    
            job.setInputFormatClass(TextInputFormat.class);
            job.setOutputFormatClass(TextOutputFormat.class);
    
            FileInputFormat.setInputPaths(job, new Path(input_path));
            FileOutputFormat.setOutputPath(job, out_path);
    
    
            boolean ret = job.waitForCompletion(true);
            return ret ? 0 :1;
        }
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            System.exit(ToolRunner.run(conf, new SecondarySortJob(), args));
        }
    }

输出结果：

a  1
    a  3
    a  5
    b  2
    b  6
    b  10
    c  8
    c  9

如果想将不同的分组输出到不同的文件中，可设置reduceTask的个数大于等于分组数，这个需要预先估计出，如将reduceTask的个数设置为3，输出结果将会有三个文件，分别存放不同的分组排序结果。

参考：

\[1\] [https://www.cnblogs.com/codeOfLife/p/5568786.html][https_www.cnblogs.com_codeOfLife_p_5568786.html]

\[2\] [https://blog.csdn.net/gamer\_gyt/article/details/47339755][https_blog.csdn.net_gamer_gyt_article_details_47339755]

[https_www.cnblogs.com_codeOfLife_p_5568786.html]: https://www.cnblogs.com/codeOfLife/p/5568786.html
[https_blog.csdn.net_gamer_gyt_article_details_47339755]: https://blog.csdn.net/gamer_gyt/article/details/47339755