MapReduce的自定义分组
GroupingComparator是mapreduce当中reduce端的一个功能组件，主要的作用是决定哪些数据作为一组，调用一次reduce的逻辑，默认是每个不同的key，作为多个不同的组，每个组调用一次reduce逻辑，我们可以自定义GroupingComparator实现不同的key作为同一个组，调用一次reduce逻辑
需求
有如下订单数据
订单id | 商品id | 成交金额 |
|---|---|---|
Order_0000001 | Pdt_01 | 222.8 |
Order_0000001 | Pdt_05 | 25.8 |
Order_0000002 | Pdt_03 | 522.8 |
Order_0000002 | Pdt_04 | 122.4 |
Order_0000002 | Pdt_05 | 722.4 |
Order_0000003 | Pdt_01 | 222.8 |
现在需要求出每一个订单中成交金额最大的一笔交易
分析
1、利用“订单id和成交金额”作为key，可以将map阶段读取到的所有订单数据按照id分区，按照金额排序，发送到reduce
2、在reduce端利用groupingcomparator将订单id相同的kv聚合成组，然后取第一个即是最大值
实现
第一步：定义OrderBean
定义一个OrderBean，里面定义两个字段，第一个字段是我们的orderId，第二个字段是我们的金额（注意金额一定要使用Double或者DoubleWritable类型，否则没法按照金额顺序排序）
代码语言：javascript复制public class OrderBean implements WritableComparable<OrderBean> {
private String orderId;
private Double price;
@Override
public int compareTo(OrderBean o) {
//比较订单id的排序顺序
int i = this.orderId.compareTo(o.orderId);
if(i==0){
//如果订单id相同，则比较金额，金额大的排在前面
i = - this.price.compareTo(o.price);
}
return i;
}
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(orderId);
out.writeDouble(price);
}
@Override
public void readFields(DataInput in) throws IOException {
this.orderId = in.readUTF();
this.price = in.readDouble();
}
public OrderBean() {
}
public OrderBean(String orderId, Double price) {
this.orderId = orderId;
this.price = price;
}
public String getOrderId() {
return orderId;
}
public void setOrderId(String orderId) {
this.orderId = orderId;
}
public Double getPrice() {
return price;
}
public void setPrice(Double price) {
this.price = price;
}
@Override
public String toString() {
return orderId "t" price;
}
}第二步：自定义分区
自定义分区，按照订单id进行分区，把所有订单id相同的数据，都发送到同一个reduce中去
代码语言：javascript复制public class OrderPartition extends Partitioner<OrderBean,NullWritable> {
@Override
public int getPartition(OrderBean orderBean, NullWritable nullWritable, int i) {
//自定义分区，将相同订单id的数据发送到同一个reduce里面去
return (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE)%i;
}
}第三步：自定义groupingComparator
按照我们自己的逻辑进行分组，通过比较相同的订单id，将相同的订单id放到一个组里面去，进过分组之后当中的数据，已经全部是排好序的数据，我们只需要取前topN即可
代码语言：javascript复制/*
1: 继承WriteableComparator
2: 调用父类的有参构造
3: 指定分组的规则(重写方法)
*/
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
// 1: 继承WriteableComparator
public class OrderGroupComparator extends WritableComparator {
// 2: 调用父类的有参构造
public OrderGroupComparator() {
super(OrderBean.class,true);
}
//3: 指定分组的规则(重写方法)
@Override
public int compare(WritableComparable a, WritableComparable b) {
//3.1 对形参做强制类型转换
OrderBean first = (OrderBean)a;
OrderBean second = (OrderBean)b;
//3.2 指定分组规则
return first.getOrderId().compareTo(second.getOrderId());
}
}第四步：程序main函数入口
代码语言：javascript复制public class GroupingRunner {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//1、创建建一个job任务对象
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration, "grouping_demo");
//2、指定job所在的jar包
job.setJarByClass(GroupingRunner.class);
//3、指定源文件的读取方式类和源文件的读取路径
job.setInputFormatClass(TextInputFormat.class); //按照行读取
//TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/wordcount")); //只需要指定源文件所在的目录即可
TextInputFormat.addInputPath(job, new Path("file:///E:\input\grouping_demo")); //只需要指定源文件所在的目录即可
//4、指定自定义的Mapper类和K2、V2类型
job.setMapperClass(GroupingMapper.class); //指定Mapper类
job.setMapOutputKeyClass(OrderBean.class); //K2类型
job.setMapOutputValueClass(Text.class);//V2类型
//5、指定自定义分区类（如果有的话）
job.setPartitionerClass(MyPartitioner.class);
//6、指定自定义分组类（如果有的话）
job.setGroupingComparatorClass(GroupingComparator.class);
//7、指定自定义Combiner类(如果有的话)
//job.setCombinerClass(MyCombiner.class);
//设置ReduceTask个数
job.setNumReduceTasks(3);
//8、指定自定义的Reducer类和K3、V3的数据类型
job.setReducerClass(GroupingReducer.class); //指定Reducer类
job.setOutputKeyClass(Text.class); //K3类型
job.setOutputValueClass(NullWritable.class); //V3类型
//9、指定输出方式类和结果输出路径
job.setOutputFormatClass(TextOutputFormat.class);
//TextOutputFormat.setOutputPath(job, new Path("hdfs://node1:8020/output/wordcount")); //目标目录不能存在，否则报错
TextOutputFormat.setOutputPath(job, new Path("file:///E:\output\grouping_demo")); //目标目录不能存在，否则报错
//10、将job提交到yarn集群
boolean bl = job.waitForCompletion(true); //true表示可以看到任务的执行进度
//11.退出执行进程
System.exit(bl?0:1);
}
}
