Hadoop入门（十）Mapreduce高级shuffle之Sort和Group

蔚落 2023-06-03 09:58 19阅读 0赞

### 一、排序分组概述 ###

MapReduce中排序和分组在哪里被执行

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21vYWt1bg_size_16_color_FFFFFF_t_70][]

第3步中需要对不同分区中的数据进行排序和分组，默认情况按照key进行排序和分组

### 二、排序 ###

在Hadoop默认的排序算法中，只会针对key值进行排序

任务：  
数据文件中，如果按照第一列升序排列，  
当第一列相同时，第二列升序排列  
如果当第一列相同时，求出第二列的最小值

![20191006122423829.png][]

**自定义排序**

1.封装一个自定义类型作为key的新类型：将第一列与第二列都作为key

WritableComparable接口

定义：

public interface WritableComparable<T> extends Writable, Comparable<T> {
    }

自定义类型MyNewKey实现了WritableComparable的接口，该接口中有一个compareTo()方法，当对key进行比较时会调用该方法，而我们将其改为了我们自己定义的比较规则，从而实现我们想要的效果

private static class MyNewKey implements WritableComparable<MyNewKey> {
            long firstNum;
            long secondNum;
     
            public MyNewKey() {
            }
     
            public MyNewKey(long first, long second) {
                firstNum = first;
                secondNum = second;
            }
     
            @Override
            public void write(DataOutput out) throws IOException {
                out.writeLong(firstNum);
                out.writeLong(secondNum);
            }
     
            @Override
            public void readFields(DataInput in) throws IOException {
                firstNum = in.readLong();
                secondNum = in.readLong();
            }
     
            /*
             * 当key进行排序时会调用以下这个compreTo方法
             */
            @Override
            public int compareTo(MyNewKey anotherKey) {
                long min = firstNum - anotherKey.firstNum;
                if (min != 0) {
                    // 说明第一列不相等，则返回两数之间小的数
                    return (int) min;
                } else {
                    return (int) (secondNum - anotherKey.secondNum);
                }
            }
        }

2.改写最初的MapReduce方法函数

public static class MyMapper extends
                Mapper<LongWritable, Text, MyNewKey, LongWritable> {
    
            protected void map(LongWritable key,Text value,
                    Mapper<LongWritable, Text, MyNewKey, LongWritable>.Context context)
                    throws java.io.IOException, InterruptedException {
    
                String[] spilted = value.toString().split("\t");
                long firstNum = Long.parseLong(spilted[0]);
                long secondNum = Long.parseLong(spilted[1]);
                // 使用新的类型作为key参与排序
                MyNewKey newKey = new MyNewKey(firstNum, secondNum);
     
                context.write(newKey, new LongWritable(secondNum));
            };
        }

### 三、分组 ###

在Hadoop中的默认分组规则中，也是基于Key进行的，会将相同key的value放到一个集合中去

目的：求出第一列相同时第二列的最小值  
上面的例子看分组，因为我们自定义了一个新的key，它是以两列数据作为key的，因此这6行数据中每个key都不相同产生6组  
它们是：1 1,2 1,2 2,3 1,3 2,3 3。

而实际上只可以分为3组，分别是1，2，3。现在首先改写一下reduce函数代码

public static class MyReducer extends
                Reducer<MyNewKey, LongWritable, LongWritable, LongWritable> {
            protected void reduce(
                    MyNewKey key,
                    java.lang.Iterable<LongWritable> values,
                    Reducer<MyNewKey, LongWritable, LongWritable, LongWritable>.Context context)
                    throws java.io.IOException, InterruptedException {
    
                long min = Long.MAX_VALUE;
                for (LongWritable number : values) {
                    long temp = number.get();
                    if (temp < min) {
                        min = temp;
                    }
                }
     
                context.write(new LongWritable(key.firstNum), new LongWritable(min));
            };
        }

![20191006122751967.png][]![20191006122801396.png][]

**自定义分组**

为了针对新的key类型作分组，我们也需要自定义一下分组规则：

private static class MyGroupingComparator implements
                RawComparator<MyNewKey> {
     
            /*
             * 基本分组规则：按第一列firstNum进行分组
             */
            @Override
            public int compare(MyNewKey key1, MyNewKey key2) {
                return (int) (key1.firstNum - key2.firstNum);
            }
     
            /*
             * @param b1 表示第一个参与比较的字节数组
             * 
             * @param s1 表示第一个参与比较的字节数组的起始位置
             * 
             * @param l1 表示第一个参与比较的字节数组的偏移量
             * 
             * @param b2 表示第二个参与比较的字节数组
             * 
             * @param s2 表示第二个参与比较的字节数组的起始位置
             * 
             * @param l2 表示第二个参与比较的字节数组的偏移量
             */
            @Override
            public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
                return WritableComparator.compareBytes(b1, s1, 8, b2, s2, 8);
            }
     
        }

自定义了一个分组比较器MyGroupingComparator，该类实现了RawComparator接口，而RawComparator接口又实现了Comparator接口，这两个接口的定义：

public interface RawComparator<T> extends Comparator<T> {
      public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2);
    }

public interface Comparator<T> {
        int compare(T o1, T o2);
        boolean equals(Object obj);
    }

分组实现步骤：

1.MyGroupingComparator实现这两个接口  
RawComparator中的compare()方法是基于字节的比较，  
Comparator中的compare()方法是基于对象的比较

由于在MyNewKey中有两个long类型，每个long类型又占8个字节。这里因为比较的是第一列数字，所以读取的偏移量为8字节。

2.添加对分组规则的设置：  
　　// 设置自定义分组规则  
   job.setGroupingComparatorClass(MyGroupingComparator.class);

3. 运行结果：

![20191006123443655.png][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21vYWt1bg_size_16_color_FFFFFF_t_70]: /images/20230601/67fabbb3c1dd4fcd9767fe2cb725ac0c.png
[20191006122423829.png]: /images/20230601/530ef2390650402296965f4d197c61de.png
[20191006122751967.png]: https://img-blog.csdnimg.cn/20191006122751967.png
[20191006122801396.png]: /images/20230601/e86d0c7a21de4b62820786b8e0659d21.png
[20191006123443655.png]: /images/20230601/f5dc52d4b0a04cf9be27fa1d87540c30.png