Hello，大家好！博主上篇讲解了合并，这篇要讲的是辅助排序。如何讲解这个章节呢？首先先对什么是合并进行解释，然后通过案例进行证明。

一. GroupingComparator分组的简介

什么是GroupingComparator分组（辅助排序）？
  对Reduce阶段的数据根据某一个或几个字段进行分组。

分组排序的步骤：

• 1. 自定义类继承WritableComparator
• 2. 重写compare()方法

@Override
public int compare(WritableComparable a, WritableComparable b) {
		// 比较的业务逻辑
		return result;
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
 

• 3. 创建一个构造将比较对象的类传给父类

protected OrderGroupingComparator() {
		super(OrderBean.class, true);
}


  

 

  
1
  
2
  
3
  
4
 

二. 根据案例分析

2.1 需求

• 1. 有如下订单数据

现在需要求出每一个订单中最贵的商品。

• 2. 输入的数据
  
• 3. 期望输出数据

1	222.8
2	722.4
3	232.8

  

 

  
1
  
2
  
3
 

2.2 需求分析

（1）利用“订单id和成交金额”作为key，可以将Map阶段读取到的所有订单数据按照id升序排序，如果id相同再按照金额降序排序，发送到Reduce。
（2）在Reduce端利用groupingComparator将订单id相同的kv聚合成组，然后取第一个即是该订单中最贵商品，如下图所示所示。

2.3 代码实现

1. 定义订单信息OrderBean类

package com.buwenbuhuo.groupingcomparator;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/**
 * @author 卜温不火
 * @create 2020-04-24 23:43
 * com.buwenbuhuo.groupingcomparator - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class OrderBean implements WritableComparable<OrderBean> { private String orderId; private String productId; private double price; @Override public String toString() { return orderId + "\t" + productId + "\t" + price; } public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId = orderId; } public String getProductId() { return productId; } public void setProductId(String productId) { this.productId = productId; } public double getPrice() { return price; } public void setPrice(double price) { this.price = price; } @Override public int compareTo(OrderBean o) { int compare = this.orderId.compareTo(o.orderId); if (compare == 0) { return Double.compare(o.price, this.price); } else { return compare; } } @Override public void write(DataOutput out) throws IOException { out.writeUTF(orderId); out.writeUTF(productId); out.writeDouble(price); } @Override public void readFields(DataInput in) throws IOException { this.orderId = in.readUTF(); this.productId = in.readUTF(); this.price = in.readDouble(); }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
  
38
  
39
  
40
  
41
  
42
  
43
  
44
  
45
  
46
  
47
  
48
  
49
  
50
  
51
  
52
  
53
  
54
  
55
  
56
  
57
  
58
  
59
  
60
  
61
  
62
  
63
  
64
  
65
  
66
  
67
  
68
  
69
  
70
  
71
  
72
  
73
  
74
  
75
 

2. 编写OrderSortMapper类

package com.buwenbuhuo.groupingcomparator;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
/**
 * @author 卜温不火
 * @create 2020-04-24 23:43
 * com.buwenbuhuo.groupingcomparator - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> { private OrderBean orderBean = new OrderBean(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); orderBean.setOrderId(fields[0]); orderBean.setProductId(fields[1]); orderBean.setPrice(Double.parseDouble(fields[2])); context.write(orderBean, NullWritable.get()); }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
 

3. 编写OrderSortGroupingComparator类

package com.buwenbuhuo.groupingcomparator;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
/**
 * @author 卜温不火
 * @create 2020-04-24 23:43
 * com.buwenbuhuo.groupingcomparator - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class OrderComparator extends WritableComparator { protected OrderComparator() { super(OrderBean.class, true); } @Override public int compare(WritableComparable a, WritableComparable b) { OrderBean oa = (OrderBean) a; OrderBean ob = (OrderBean) b; return oa.getOrderId().compareTo(ob.getOrderId()); }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
 

4. 编写OrderSortReducer类

package com.buwenbuhuo.groupingcomparator;

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

import java.io.IOException;
import java.util.Iterator;
/**
 * @author 卜温不火
 * @create 2020-04-24 23:43
 * com.buwenbuhuo.groupingcomparator - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> { @Override protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { Iterator<NullWritable> iterator = values.iterator(); for (int i = 0; i < 2; i++) { if (iterator.hasNext()) { context.write(key, iterator.next()); } } }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
 

5. 编写OrderSortDriver类

package com.buwenbuhuo.groupingcomparator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
/**
 * @author 卜温不火
 * @create 2020-04-24 23:43
 * com.buwenbuhuo.groupingcomparator - the name of the target package where the new class or interface will be created.
 * mapreduce0422 - the name of the current project.
 */
public class OrderDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Job job = Job.getInstance(new Configuration()); job.setJarByClass(OrderDriver.class); job.setMapperClass(OrderMapper.class); job.setReducerClass(OrderReducer.class); job.setMapOutputKeyClass(OrderBean.class); job.setMapOutputValueClass(NullWritable.class); job.setGroupingComparatorClass(OrderComparator.class); job.setOutputKeyClass(OrderBean.class); job.setOutputValueClass(NullWritable.class); FileInputFormat.setInputPaths(job, new Path("d:\\input")); FileOutputFormat.setOutputPath(job, new Path("d:\\output")); boolean b = job.waitForCompletion(true); System.exit(b ? 0 : 1); }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
  
38
  
39
  
40
  
41
 

2.4 运行与结果实现

• 1. 运行
  
• 2. 结果与对比
  

本次的分享就到这里了,大家有什么疑惑或者好的建议可以在评论区积极留言。受益的小伙伴们不要忘了点赞关注我呀！！！

文章来源: buwenbuhuo.blog.csdn.net，作者：不温卜火，版权归原作者所有，如需转载，请联系作者。

原文链接：buwenbuhuo.blog.csdn.net/article/details/105740800