MapReduce之二次排序案例详解！

旧城等待， 2022-12-15 08:47 150阅读 0赞

### 目录 ###

*   *  一、需求分析
     *  二、方案一实现
     *  三、方案二实现

## 一、需求分析 ##

MR的二次排序的需求说明：  
在mapreduce操作时，shuffle阶段会多次根据key值排序。但是在shuffle分组后，相同key值的values序列的顺序是不确定的(如下图)。如果想要此时value值也是排序好的，这种需求就是二次排序。  
![在这里插入图片描述][20201015210829402.png_pic_center]

原始数据		无二次排序	有二次排序
    	a 12    	a 12    	a 12
    	b 34		b 34		b 13
    	c 90		b 23    	b 23
    	b 23		b 13    	b 34
    	b 13		c 90    	c 90

根据案例分析，我们要将下面数据key按照abc，value按照大小排序，这也就是一个典型的MR的二次排序的案例，准备原始数据：

a 20
    b 20
    a 5
    c 10
    c 8
    b 15
    a 10
    b 18
    c 29
    b 52

我们想要得到的结果：

a       5
    a       10
    a       20
    b       15
    b       18
    b       20
    b       52
    c       8
    c       10
    c       29

## 二、方案一实现 ##

先看方案一的实现思路：

input -> map -><a,20> -> shuffle -> <a,list(10, 5, 20)> -> reduce -> <a,5>
    			   <b,20>				<b,list(52, 18, 15, 20)>         <a,10>
       			   <a,5>				<c,list(29, 8, 10)>    		     <a,20>
       			   <c,10>											     <b,15>
       			   ...												     <b,18>
       			   													     <b,20>
       			   													      ...

**直接在reduce端对分组后的values进行排序**  
示例代码：

package com.kfk.hadoop.mr.sort;
    
    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.LongWritable;
    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.output.FileOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.List;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/9 * @time : 7:07 下午 */
    
    public class SortMR extends Configured implements Tool { 
    
        /** * map * TODO */
        public static class TemplateMapper extends Mapper<LongWritable, Text,Text, IntWritable>{ 
    
            // 创建map输出的对象
            private static final Text mapOutKey = new Text();
            private static final IntWritable mapOutValue = new IntWritable();
    
            @Override
            public void setup(Context context) throws IOException, InterruptedException { 
                // TODO
            }
    
            @Override
            public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
    
                // 将每一行数据按空格拆开
                String[] values = value.toString().split(" ");
    
                // 数据预处理，将数组超过2的过滤掉
                if (values.length != 2){ 
                    return;
                }
                mapOutKey.set(values[0]);
                mapOutValue.set(Integer.valueOf(values[1]));
                context.write(mapOutKey,mapOutValue);
    
            }
    
            @Override
            public void cleanup(Context context) throws IOException, InterruptedException { 
                // TODO
            }
        }
    
        /** * reduce * TODO */
        public static class TemplateReducer extends Reducer<Text,IntWritable,Text,IntWritable>{ 
    
            // 创建reduceout端的对象
            private static final IntWritable outputValue = new IntWritable();
    
            @Override
            public void setup(Context context) throws IOException, InterruptedException { 
                // TODO
            }
    
            @Override
            public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
                List<Integer> valueList = new ArrayList<Integer>();
    
                // 取出value
                for (IntWritable value:values){ 
                    valueList.add(value.get());
                }
    
    			// 打印出reduce输入的key和valueList
                System.out.println("Reduce in == KeyIn: "+key+" ValueIn: "+valueList);
                // 进行排序
                Collections.sort(valueList);
    
    			/* valueList:表示上面已经排序好的列表，即需要遍历列表中的值作为reduce的输出 key不需要改变，即可作为reduce的输出 */
                for (Integer value : valueList){ 
                    outputValue.set(value);
                    context.write(key,outputValue);
                }
            }
    
            @Override
            public void cleanup(Context context) throws IOException, InterruptedException { 
                // TODO
            }
        }
    
        /** * run * @param args * @return * @throws IOException * @throws ClassNotFoundException * @throws InterruptedException */
        public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            // 1) get conf
            Configuration configuration = this.getConf();
    
            // 2) create job
            Job job = Job.getInstance(configuration,this.getClass().getSimpleName());
            job.setJarByClass(this.getClass());
    
            // 3.1) input，指定job的输入
            Path path = new Path(args[0]);
            FileInputFormat.addInputPath(job,path);
    
            // 3.2) map，指定job的mapper和输出的类型
            job.setMapperClass(TemplateMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            // 1.分区
            // job.setPartitionerClass();
    
            // 2.排序
            // job.setSortComparatorClass();
    
            // 3.combiner -可选项
            // job.setCombinerClass(WordCountCombiner.class);
    
            // 4.compress -可配置
            // configuration.set("mapreduce.map.output.compress","true");
            // 使用的SnappyCodec压缩算法
            // configuration.set("mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");
    
            // 5.分组
            // job.setGroupingComparatorClass();
    
            // 6.设置reduce的数量
            // job.setNumReduceTasks(2);
    
            // 3.3) reduce，指定job的reducer和输出类型
            job.setReducerClass(TemplateReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            // 3.4) output，指定job的输出
            Path outpath = new Path(args[1]);
            FileOutputFormat.setOutputPath(job,outpath);
    
            // 4) commit，执行job
            boolean isSuccess = job.waitForCompletion(true);
            // 如果正常执行返回0，否则返回1
            return (isSuccess) ? 0 : 1;
        }
    
        public static void main(String[] args) { 
    
            // 添加输入，输入参数
            args = new String[]{ 
                "hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/secondSort",
                "hdfs://bigdata-pro-m01:9000/user/caizhengjie/mr/output"
            };
    // WordCountUpMR wordCountUpMR = new WordCountUpMR();
            Configuration configuration = new Configuration();
    
            try { 
                // 判断输出的文件存不存在，如果存在就将它删除
                Path fileOutPath = new Path(args[1]);
                FileSystem fileSystem = FileSystem.get(configuration);
                if (fileSystem.exists(fileOutPath)){ 
                    fileSystem.delete(fileOutPath,true);
                }
                // 调用run方法
                int status = ToolRunner.run(configuration,new SortMR(),args);
                // 退出程序
                System.exit(status);
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (ClassNotFoundException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

运行结果：

a       5
    a       10
    a       20
    b       15
    b       18
    b       20
    b       52
    c       8
    c       10
    c       29

很容易发现，这样把排序工作都放到reduce端完成，当values序列长度非常大的时候，会对CPU和内存造成极大的负载。

**注意的地方（容易被“坑”）**  
在reduce端对values进行迭代的时候，不要直接存储value值或者key值，因为reduce方法会反复执行多次，但key和value相关的对象只有两个，reduce会反复重用这两个对象。需要用相应的数据类型.get()取出后再存储。

## 三、方案二实现 ##

方案二的解决思路:

原始数据 			自定义newkey 在shuffle中排序  reduce输入				     reduce输出
    	a 12    		a#12,12    a#12,12
    	b 34			b#34,34	   b#13,13
    	c 90 -> map ->	c#90,90    b#23,23       b#,List(13,23,34)-> reduce ->  b,13 b,23 b,34
    	b 23			b#23,23    b#34,34	
    	b 13			b#13,13    c#90,90

我们可以把key和value联合起来作为新的key，记作newkey。这时，newkey含有两个字段，假设分别是k,v。这里的k和v是原来的key和value。原来的value还是不变。这样，value就同时在newkey和value的位置。我们再实现newkey的比较规则，先按照key排序，在key相同的基础上再按照value排序。在分组时，再按照原来的key进行分组，就不会影响原有的分组逻辑了。最后在输出的时候，只把原有的key、value输出，就可以变通的实现了二次排序的需求。

需要自定义的地方  
　　1.自定义数据类型实现组合key  
　　　　实现方式：继承WritableComparable  
　　2.自定义partioner，形成newKey后保持分区规则任然按照key进行。保证不打乱原来的分区。  
　　　　实现方式：继承Partitioner  
　　3.自定义分组，保持分组规则任然按照key进行。不打乱原来的分组  
　　　　实现方式：继承RawComparator

**自定义数据类型代码：**

package com.kfk.hadoop.mr.secondsort;
    
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/15 * @time : 6:16 下午 */
    public class PairWritable implements WritableComparable<PairWritable> { 
    
        // 组合key:a#12,12
        private String first;
        private int second;
    
        public PairWritable() { 
    
        }
    
        public PairWritable(String first, int second) { 
            this.set(first,second);
        }
    
        /** * 方便设置字段 */
        public void set(String first, int second){ 
            this.first = first;
            this.second = second;
        }
    
        public String getFirst() { 
            return first;
        }
    
        public void setFirst(String first) { 
            this.first = first;
        }
    
        public int getSecond() { 
            return second;
        }
    
        public void setSecond(int second) { 
            this.second = second;
        }
    
        /** * 重写比较器 */
        public int compareTo(PairWritable o) { 
    
            int comp = this.getFirst().compareTo(o.getFirst());
            if (0 == comp){ 
                // 若第一个字段相等，则比较第二个字段
                return Integer.valueOf(this.getSecond()).compareTo(o.getSecond());
            }
            return comp;
        }
    
        /** * 序列化 */
        public void write(DataOutput out) throws IOException { 
            out.writeUTF(first);
            out.writeInt(second);
        }
    
        /** * 反序列化 */
        public void readFields(DataInput in) throws IOException { 
            this.first = in.readUTF();
            this.second = in.readInt();
        }
    
        @Override
        public String toString() { 
            return "PairWritable{" +
                    "first='" + first + '\'' +
                    ", second=" + second +
                    '}';
        }
    
        @Override
        public boolean equals(Object o) { 
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
    
            PairWritable that = (PairWritable) o;
    
            if (second != that.second) return false;
            return first != null ? first.equals(that.first) : that.first == null;
        }
    
        @Override
        public int hashCode() { 
            int result = first != null ? first.hashCode() : 0;
            result = 31 * result + second;
            return result;
        }
    }

**自定义分区代码：**

package com.kfk.hadoop.mr.secondsort;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/15 * @time : 7:09 下午 */
    public class FristPartitioner extends Partitioner<PairWritable, IntWritable> { 
    
        public int getPartition(PairWritable key, IntWritable intWritable, int numPartitions) { 
           /* * 默认的实现 (key.hashCode() & Integer.MAX_VALUE) % numPartitions * 让key中first字段作为分区依据 */
            return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;
        }
    }

**自定义分组比较器代码：**

package com.kfk.hadoop.mr.secondsort;
    
    import org.apache.hadoop.io.RawComparator;
    
    import org.apache.hadoop.io.WritableComparator;
    
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/15 * @time : 6:59 下午 */
    public class FristGrouping implements RawComparator<PairWritable> { 
    
        /* * 字节比较 * bytes1,bytes2为要比较的两个字节数组 * i,i1表示第一个字节数组要进行比较的收尾位置，i2,i3表示第二个 * 从第一个字节比到组合key中second的前一个字节，因为second为int型，所以长度为4 */
    
        public int compare(byte[] bytes1, int i, int i1, byte[] bytes2, int i2, int i3) { 
            return WritableComparator.compareBytes(bytes1,0,i1-4,bytes2,0,i3-4);
        }
    
        /* * 对象比较 */
        public int compare(PairWritable o1, PairWritable o2) { 
            return o1.getFirst().compareTo(o2.getFirst());
        }
    }

二次排序实现代码：

package com.kfk.hadoop.mr.secondsort;
    
    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.LongWritable;
    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.output.FileOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    import java.io.IOException;
    
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/9 * @time : 7:07 下午 */
    
    public class SecondSortMR extends Configured implements Tool { 
    
        /** * map * TODO */
        public static class TemplateMapper extends Mapper<LongWritable, Text,PairWritable, IntWritable>{ 
    
            // 创建map输出的对象
            private static final PairWritable mapOutKey = new PairWritable();
            private static final IntWritable mapOutValue = new IntWritable();
    
            @Override
            public void setup(Context context) { 
                // TODO
            }
    
            @Override
            public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
    
                // 将每一行数据按空格拆开
                String[] values = value.toString().split(" ");
    
                // 数据预处理，将数组超过2的过滤掉
                if (values.length != 2){ 
                    return;
                }
                mapOutKey.set(values[0],Integer.parseInt(values[1]));
                mapOutValue.set(Integer.parseInt(values[1]));
                context.write(mapOutKey,mapOutValue);
    
                System.out.println("Map out == KeyOut: "+mapOutKey+" ValueOut: "+mapOutValue);
    
            }
    
            @Override
            public void cleanup(Context context) { 
                // TODO
            }
        }
    
        /** * reduce * TODO */
        public static class TemplateReducer extends Reducer<PairWritable,IntWritable,Text,IntWritable>{ 
    
            // 创建reduce output端的对象
            private static final IntWritable outputValue = new IntWritable();
            private static final Text outputKey = new Text();
    
            @Override
            public void setup(Context context) throws IOException, InterruptedException { 
                // TODO
            }
    
            @Override
            public void reduce(PairWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
    
    			/* values表示reduce端输入已经排序好的列表，即需要遍历values每一个值作为reduce输出即可 key表示为自定义的key（newkey），即需要取出newkey的第一部分，也就是原始的key，作为reduce的输出 */
                for (IntWritable value:values){ 
                    outputKey.set(key.getFirst());
                    context.write(outputKey,value);
                }
            }
    
            @Override
            public void cleanup(Context context) throws IOException, InterruptedException { 
                // TODO
            }
        }
    
        /** * run * @param args * @return * @throws IOException * @throws ClassNotFoundException * @throws InterruptedException */
        public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            // 1) get conf
            Configuration configuration = this.getConf();
    
            // 2) create job
            Job job = Job.getInstance(configuration,this.getClass().getSimpleName());
            job.setJarByClass(this.getClass());
    
            // 3.1) input，指定job的输入
            Path path = new Path(args[0]);
            FileInputFormat.addInputPath(job,path);
    
            // 3.2) map，指定job的mapper和输出的类型
            job.setMapperClass(TemplateMapper.class);
            job.setMapOutputKeyClass(PairWritable.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            // 1.分区
            job.setPartitionerClass(FristPartitioner.class);
    
            // 2.排序
            // job.setSortComparatorClass();
    
            // 3.combiner -可选项
            // job.setCombinerClass(WordCountCombiner.class);
    
            // 4.compress -可配置
            // configuration.set("mapreduce.map.output.compress","true");
            // 使用的SnappyCodec压缩算法
            // configuration.set("mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");
    
            // 5.分组
            job.setGroupingComparatorClass(FristGrouping.class);
    
            // 6.设置reduce的数量
            // job.setNumReduceTasks(2);
    
            // 3.3) reduce，指定job的reducer和输出类型
            job.setReducerClass(TemplateReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            // 3.4) output，指定job的输出
            Path outpath = new Path(args[1]);
            FileOutputFormat.setOutputPath(job,outpath);
    
            // 4) commit，执行job
            boolean isSuccess = job.waitForCompletion(true);
            // 如果正常执行返回0，否则返回1
            return (isSuccess) ? 0 : 1;
        }
    
        public static void main(String[] args) { 
    
            // 添加输入，输入参数
            args = new String[]{ 
                "hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/secondSort",
                "hdfs://bigdata-pro-m01:9000/user/caizhengjie/mr/output"
            };
    // WordCountUpMR wordCountUpMR = new WordCountUpMR();
            Configuration configuration = new Configuration();
    
            try { 
                // 判断输出的文件存不存在，如果存在就将它删除
                Path fileOutPath = new Path(args[1]);
                FileSystem fileSystem = FileSystem.get(configuration);
                if (fileSystem.exists(fileOutPath)){ 
                    fileSystem.delete(fileOutPath,true);
                }
                // 调用run方法
                int status = ToolRunner.run(configuration,new SecondSortMR(),args);
                // 退出程序
                System.exit(status);
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (ClassNotFoundException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]

a       5
    a       10
    a       20
    b       15
    b       18
    b       20
    b       52
    c       8
    c       10
    c       29

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[20201015210829402.png_pic_center]: /images/20221123/4ae155be46c241838bc37d1c18bc2308.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/9e0bb6d3659a43d698b1c8c5d3f6e3b5.png