前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >2021年大数据Hadoop(二十二):MapReduce的自定义分组

2021年大数据Hadoop(二十二):MapReduce的自定义分组

作者头像
Lansonli
发布2021-10-11 15:47:06
4190
发布2021-10-11 15:47:06
举报
文章被收录于专栏:Lansonli技术博客Lansonli技术博客

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

    }

}

  • 📢博客主页:https://lansonli.blog.csdn.net
  • 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正!
  • 📢本文由 Lansonli 原创,首发于 CSDN博客🙉
  • 📢大数据系列文章会每天更新,停下休息的时候不要忘了别人还在奔跑,希望大家抓紧时间学习,全力奔赴更美好的生活✨
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2021-05-31 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • MapReduce的自定义分组
    • ​​​​​​​需求
      • ​​​​​​​分析
        • 实现
          • 第一步:定义OrderBean
          • ​​​​​​​第二步:自定义分区
          • ​​​​​​​第三步:自定义groupingComparator
          • ​​​​​​​第四步:程序main函数入口
      领券
      问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档