指对Reduce阶段的数据根据某一个或几个字段进行分组。

案例

需求

有如下订单数据



现在需要找出每一个订单中最贵的商品,如图

需求分析

  • 利用“订单id和成交金额”作为key,可以将Map阶段读取到的所有订单数据先按照订单id(升降序都可以),再按照acount(降序)排序,发送到Reduce。

  • 在Reduce端利用groupingComparator将订单id相同的kv聚合成组,然后取第一个成交金额即是最大值(若有多个成交金额并排第一,则都输出)。

  • Mapper阶段主要做三件事:

    keyin-valuein

    map()

    keyout-valueout

  • 期待shuffle之后的数据:

    10000001 Pdt_02 222.8

    10000001 Pdt_01 222.8

    10000001 Pdt_05 25.8

    10000002 Pdt_06 722.4

    10000002 Pdt_03 522.8

    10000002 Pdt_04 122.4

    10000003 Pdt_01 232.8

    10000003 Pdt_01 33.8

  • Reducer阶段主要做三件事:

    keyin-valuein

    reduce()

    keyout-valueout

  • 进入Reduce需要考虑的事

  1. 获取分组比较器,如果没设置默认使用MapTask排序时key的比较器
  2. 默认的比较器比较策略不符合要求,它会将orderId一样且acount一样的记录才认为是一组的
  3. 自定义分组比较器,只按照orderId进行对比,只要OrderId一样,认为key相等,这样可以将orderId相同的分到一个组!

    在组内去第一个最大的即可

编写程序

利用“订单id和成交金额”作为key,所以把每一行记录封装为bean。由于需要比较ID,所以实现了WritableComparable接口

OrderBean.java

public class OrderBean implements WritableComparable<OrderBean>{

	private String orderId;
private String pId;
private Double acount; public String getOrderId() {
return orderId;
}
public void setOrderId(String orderId) {
this.orderId = orderId;
}
public String getpId() {
return pId;
}
public void setpId(String pId) {
this.pId = pId;
}
public Double getAcount() {
return acount;
}
public void setAcount(Double acount) {
this.acount = acount;
}
public OrderBean() { } @Override
public String toString() {
return orderId + "\t" + pId + "\t" + acount ;
} @Override
public void write(DataOutput out) throws IOException {
out.writeUTF(orderId);
out.writeUTF(pId);
out.writeDouble(acount);
} @Override
public void readFields(DataInput in) throws IOException {
orderId=in.readUTF();
pId=in.readUTF();
acount=in.readDouble();
} // 二次排序,先按照orderid排序(升降序都可以),再按照acount(降序)排序
@Override
public int compareTo(OrderBean o) { //先按照orderid排序升序排序
int result=this.orderId.compareTo(o.getOrderId()); if (result==0) {//订单ID相同,就比较成交金额的大小
//再按照acount(降序)排序
result=-this.acount.compareTo(o.getAcount()); } return result;
}
}

自定义比较器,可以通过两种方法:

  • 继承WritableCompartor
  • 实现RawComparator

MyGroupingComparator.java

//实现RawComparator
public class MyGroupingComparator implements RawComparator<OrderBean>{ private OrderBean key1=new OrderBean();
private OrderBean key2=new OrderBean();
private DataInputBuffer buffer=new DataInputBuffer(); @Override
public int compare(OrderBean o1, OrderBean o2) {
return o1.getOrderId().compareTo(o2.getOrderId());
} @Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { try {
buffer.reset(b1, s1, l1); // parse key1
key1.readFields(buffer); buffer.reset(b2, s2, l2); // parse key2
key2.readFields(buffer); buffer.reset(null, 0, 0); // clean up reference
} catch (IOException e) {
throw new RuntimeException(e);
} return compare(key1, key2);
} }

MyGroupingComparator2.java

 //继承WritableCompartor
public class MyGroupingComparator2 extends WritableComparator{ public MyGroupingComparator2() {
super(OrderBean.class,null,true);
} public int compare(WritableComparable a, WritableComparable b) {
OrderBean o1=(OrderBean) a;
OrderBean o2=(OrderBean) b;
return o1.getOrderId().compareTo(o2.getOrderId());
}
}

OrderMapper.java

public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable>{

	private OrderBean out_key=new OrderBean();
private NullWritable out_value=NullWritable.get(); @Override
protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, OrderBean, NullWritable>.Context context)
throws IOException, InterruptedException { String[] words = value.toString().split("\t"); out_key.setOrderId(words[0]);
out_key.setpId(words[1]);
out_key.setAcount(Double.parseDouble(words[2])); context.write(out_key, out_value); }
}

OrderReducer.java

public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable>{

	/*
* OrderBean key-NullWritable nullWritable在reducer工作期间,
* 只会实例化一个key-value的对象!
* 每次调用迭代器迭代下个记录时,使用反序列化器从文件中或内存中读取下一个key-value数据的值,
* 封装到之前OrderBean key-NullWritable nullWritable在reducer的属性中
*/
@Override
protected void reduce(OrderBean key, Iterable<NullWritable> values,
Reducer<OrderBean, NullWritable, OrderBean, NullWritable>.Context context)
throws IOException, InterruptedException { Double maxAcount = key.getAcount(); for (NullWritable nullWritable : values) { if (!key.getAcount().equals(maxAcount)) {
break;
}
//复合条件的记录
context.write(key, nullWritable); } }
}

OrderBeanDriver.java

public class OrderBeanDriver {

	public static void main(String[] args) throws Exception {

		Path inputPath=new Path("E:\\mrinput\\groupcomparator");
Path outputPath=new Path("e:/mroutput/groupcomparator"); //作为整个Job的配置
Configuration conf = new Configuration(); //保证输出目录不存在
FileSystem fs=FileSystem.get(conf); if (fs.exists(outputPath)) { fs.delete(outputPath, true); } // ①创建Job
Job job = Job.getInstance(conf); // ②设置Job
// 设置Job运行的Mapper,Reducer类型,Mapper,Reducer输出的key-value类型
job.setMapperClass(OrderMapper.class);
job.setReducerClass(OrderReducer.class); // Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器,通过序列化器对输出的key-value进行序列化和反序列化
// 如果Mapper和Reducer输出的Key-value类型一致,直接设置Job最终的输出类型 job.setOutputKeyClass(OrderBean.class);
job.setOutputValueClass(NullWritable.class); // 设置输入目录和输出目录
FileInputFormat.setInputPaths(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath); // 设置自定义的分组比较器
job.setGroupingComparatorClass(MyGroupingComparator2.class); // ③运行Job
job.waitForCompletion(true); }
}

输出结果

MapReduce之GroupingComparator分组(辅助排序、二次排序)的更多相关文章

  1. mapreduce编程(一)-二次排序

    转自:http://blog.csdn.net/heyutao007/article/details/5890103 mr自带的例子中的源码SecondarySort,我重新写了一下,基本没变. 这个 ...

  2. spark分组统计及二次排序案例一枚

    组织数据形式: aa 11 bb 11 cc 34 aa 22 bb 67 cc 29 aa 36 bb 33 cc 30 aa 42 bb 44 cc 49 需求: 1.对上述数据按key值进行分组 ...

  3. Spark基础排序+二次排序(java+scala)

    1.基础排序算法 sc.textFile()).reduceByKey(_+_,).map(pair=>(pair._2,pair._1)).sortByKey(false).map(pair= ...

  4. Hadoop MapReduce编程 API入门系列之二次排序(十六)

    不多说,直接上代码. -- ::, INFO [org.apache.hadoop.metrics.jvm.JvmMetrics] - Initializing JVM Metrics with pr ...

  5. Hadoop MapReduce 二次排序原理及其应用

    关于二次排序主要涉及到这么几个东西: 在0.20.0 以前使用的是 setPartitionerClass setOutputkeyComparatorClass setOutputValueGrou ...

  6. 详细讲解MapReduce二次排序过程

    我在15年处理大数据的时候还都是使用MapReduce, 随着时间的推移, 计算工具的发展, 内存越来越便宜, 计算方式也有了极大的改变. 到现在再做大数据开发的好多同学都是直接使用spark, hi ...

  7. Haoop MapReduce 的Partition和reduce端的二次排序

    先贴一张原理图(摘自hadoop权威指南第三版) 实际中看了半天还是不太理解其中的Partition,和reduce端的二次排序,最终根据实验来结果来验证自己的理解 1eg 数据如下 20140101 ...

  8. Hadoop案例(八)辅助排序和二次排序案例(GroupingComparator)

    辅助排序和二次排序案例(GroupingComparator) 1.需求 有如下订单数据 订单id 商品id 成交金额 0000001 Pdt_01 222.8 0000001 Pdt_05 25.8 ...

  9. Hadoop Mapreduce分区、分组、二次排序过程详解[转]

    原文地址:Hadoop Mapreduce分区.分组.二次排序过程详解[转]作者: 徐海蛟 教学用途 1.MapReduce中数据流动   (1)最简单的过程:  map - reduce   (2) ...

  10. Hadoop Mapreduce分区、分组、二次排序

    1.MapReduce中数据流动   (1)最简单的过程:  map - reduce   (2)定制了partitioner以将map的结果送往指定reducer的过程: map - partiti ...

随机推荐

  1. JS实战 &#183; 仿css样式选择器

    代码如下: <html> <head>     <meta http-equiv="Content-Type" content="text/ ...

  2. Android 四大组件之再论BroadCast

    BroadCast 是android提供的跨进程通讯的有一利器. 1.异步执行onReceiver @Nullable public abstract Intent registerReceiver( ...

  3. OC字符串NSString

    ========================== 面向对象编程进阶和字符串 ========================== Δ一.类的设计模式—单例 [单例]程序允许过程中,有且仅有一块内存 ...

  4. 同一台服务器启动多个driver负载机实例

    COSBench添加driver负载机 说明:Driver是COSBench测试工具中对负载机的一种标记,相当于loadrunner中的负载发生器. 在进行测试时,不管出于什么原因,我有时候就想单台服 ...

  5. SQLServer学习笔记&lt;&gt;日期和时间数据的处理(cast转化格式、日期截取、日期的加减)和 case表达式

    日期和时间数据的处理. (1)字符串日期 ‘20080301’,这一串为字符串日期,但必须保证为四位的年份,两位的月份,两位的日期.例如,查询订单表日期大于‘20080301’.可以这样写: 1 se ...

  6. Android系统中 setprop,getprop,watchprops命令的使用

    如:在frameworks/opt/net/ims/src/java/com/android/ims/ImsManager.java if (SystemProperties.get("pe ...

  7. JAVA线程优化

    Java并发编程:线程池的使用 在前面的文章中,我们使用线程的时候就去创建一个线程,这样实现起来非常简便,但是就会有一个问题: 如果并发的线程数量很多,并且每个线程都是执行一个时间很短的任务就结束了, ...

  8. C#实现Zip压缩解压实例【转】

    本文只列举一个压缩帮助类,使用的是有要添加一个dll引用ICSharpCode.SharpZipLib.dll[下载地址]. 另外说明一下的是,这个类压缩格式是ZIP的,所以文件的后缀写成 .zip. ...

  9. SPFile上传文件到文档库

    , dataLen);            SPSite sps = SPControl.GetContextSite(Context);            sps.AllowUnsafeUpd ...

  10. Error Handling in ASP.NET Core

    Error Handling in ASP.NET Core 前言  在程序中,经常需要处理比如 404,500 ,502等错误,如果直接返回错误的调用堆栈的具体信息,显然大部分的用户看到是一脸懵逼的 ...