MapReduce快速入门系列(10) | 二次排序和辅助排序案例(GroupingComparator分组)

举报
不温卜火 发表于 2020/12/02 22:45:26 2020/12/02
5.8k+ 0 0
【摘要】 Hello,大家好!博主上篇讲解了合并,这篇要讲的是辅助排序。如何讲解这个章节呢?首先先对什么是合并进行解释,然后通过案例进行证明。 目录 一. GroupingComparator分组的简介二. 根据案例分析2.1 需求2.2 需求分析2.3 代码实现1. 定义订单信息OrderBean类2. 编写OrderSortMapper类3. 编写Orde...

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


一. GroupingComparator分组的简介

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

分组排序的步骤:

  • 1. 自定义类继承WritableComparator
  • 2. 重写compare()方法
@Override
public int compare(WritableComparable a, WritableComparable b) {
		// 比较的业务逻辑
		return result;
}
  
 
  • 3. 创建一个构造将比较对象的类传给父类
protected OrderGroupingComparator() {
		super(OrderBean.class, true);
}
  
 

二. 根据案例分析

2.1 需求

  • 1. 有如下订单数据
订单id 商品id 成交金额
0000001 Pdt_01 222.8
Pdt_02 33.8
0000001 Pdt_03 522.8
Pdt_04 122.4
Pdt_05 722.4
0000001 Pdt_06 232.8
Pdt_02 33.8

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

  • 2. 输入的数据
    1
  • 3. 期望输出数据
1	222.8
2	722.4
3	232.8

  
 

2.2 需求分析

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

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(); }
}
  
 

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()); }
}
  
 

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()); }
}
  
 

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()); } } }
}
  
 

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); }
}
  
 

2.4 运行与结果实现

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

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

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

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

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

作者其他文章

评论(0

抱歉,系统识别当前为高风险访问,暂不支持该操作

    全部回复

    上滑加载中

    设置昵称

    在此一键设置昵称,即可参与社区互动!

    *长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

    *长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。