经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Hadoop » 查看文章
大数据-Hadoop生态(18)-MapReduce框架原理-WritableComparable排序和GroupingComparator分组
来源：cnblogs　　作者：nt杨　　时间：2018/12/13 9:20:17　　对本文有异议

1.排序概述

2.排序分类

 

3.WritableComparable案例

这个文件,是大数据-Hadoop生态(12)-Hadoop序列化和源码追踪的输出文件,可以看到,文件根据key,也就是手机号进行了字典排序

  1. 13470253144    180    180    360
  2. 13509468723    7335    110349    117684
  3. 13560439638    918    4938    5856
  4. 13568436656    3597    25635    29232
  5. 13590439668    1116    954    2070
  6. 13630577991    6960    690    7650
  7. 13682846555    1938    2910    4848
  8. 13729199489    240    0    240
  9. 13736230513    2481    24681    27162
  10. 13768778790    120    120    240
  11. 13846544121    264    0    264
  12. 13956435636    132    1512    1644
  13. 13966251146    240    0    240
  14. 13975057813    11058    48243    59301
  15. 13992314666    3008    3720    6728
  16. 15043685818    3659    3538    7197
  17. 15910133277    3156    2936    6092
  18. 15959002129    1938    180    2118
  19. 18271575951    1527    2106    3633
  20. 18390173782    9531    2412    11943
  21. 84188413    4116    1432    5548

字段含义分别为手机号,上行流量,下行流量,总流量

需求是根据总流量进行排序

 

Bean对象,需要实现序列化,反序列化和Comparable接口

  1. package com.nty.writableComparable;
  3. import org.apache.hadoop.io.WritableComparable;
  5. import java.io.DataInput;
  6. import java.io.DataOutput;
  7. import java.io.IOException;
  9. /**
  10.  * author nty
  11.  * date time 2018-12-12 16:33
  12.  */
  13.  
  14. /**
  15.  * 实现WritableComparable接口
  16.  * 原先将bean序列化时,需要实现Writable接口,现在再实现Comparable接口
  17.  * 
  18.  * public interface WritableComparable<T> extends Writable, Comparable<T>
  19.  * 
  20.  * 所以我们可以实现Writable和Comparable两个接口,也可以实现WritableComparable接口
  21.  */
  22. public class Flow implements WritableComparable<Flow> {
  24.   private long upflow;
  25.   private long downflow;
  26.   private long total;
  28.     public long getUpflow() {
  29.         return upflow;
  30.     }
  32.     public void setUpflow(long upflow) {
  33.         this.upflow = upflow;
  34.     }
  36.     public long getDownflow() {
  37.         return downflow;
  38.     }
  40.     public void setDownflow(long downflow) {
  41.         this.downflow = downflow;
  42.     }
  44.     public long getTotal() {
  45.         return total;
  46.     }
  48.     public void setTotal(long total) {
  49.         this.total = total;
  50.     }
  52.     //快速赋值
  53.     public void setFlow(long upflow, long downflow){
  54.         this.upflow = upflow;
  55.         this.downflow = downflow;
  56.         this.total = upflow + downflow;
  57.     }
  59.     @Override
  60.     public String toString() {
  61.         return upflow + "\t" + downflow + "\t" + total;
  62.     }
  64.     //重写compareTo方法
  65.     @Override
  66.     public int compareTo(Flow o) {
  67.         return Long.compare(o.total, this.total);
  68.     }
  70.     //序列化方法
  71.     @Override
  72.     public void write(DataOutput out) throws IOException {
  73.         out.writeLong(upflow);
  74.         out.writeLong(downflow);
  75.         out.writeLong(total);
  76.     }
  78.     //反序列化方法
  79.     @Override
  80.     public void readFields(DataInput in) throws IOException {
  81.         upflow = in.readLong();
  82.         downflow = in.readLong();
  83.         total = in.readLong();
  84.     }
  85. }

Mapper类

  1. package com.nty.writableComparable;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Mapper;
  7. import java.io.IOException;
  9. /**
  10.  * author nty
  11.  * date time 2018-12-12 16:47
  12.  */
  13. public class FlowMapper extends Mapper<LongWritable, Text, Flow, Text> {
  15.     private Text phone = new Text();
  17.     private Flow flow = new Flow();
  20.     @Override
  21.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  22.         //13470253144    180    180    360
  23.         //分割行数据
  24.         String[] flieds = value.toString().split("\t");
  26.         //赋值
  27.         phone.set(flieds[0]);
  29.         flow.setFlow(Long.parseLong(flieds[1]), Long.parseLong(flieds[2]));
  31.         //写出
  32.         context.write(flow, phone);
  33.     }
  34. }

Reducer类

  1. package com.nty.writableComparable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Reducer;
  6. import java.io.IOException;
  8. /**
  9.  * author nty
  10.  * date time 2018-12-12 16:47
  11.  */
  12. //注意一下输出类型
  13. public class FlowReducer extends Reducer<Flow, Text, Text, Flow> {
  15.     @Override
  16.     protected void reduce(Flow key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
  17.         for (Text value : values) {
  18.             //输出
  19.             context.write(value,key);
  20.         }
  21.     }
  22. }

Driver类

  1. package com.nty.writableComparable;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. /**
  11.  * author nty
  12.  * date time 2018-12-12 16:47
  13.  */
  14. public class FlowDriver {
  16.     public static void main(String[] args) throws  Exception {
  17.         //1. 获取Job实例
  18.         Configuration configuration = new Configuration();
  19.         Job instance = Job.getInstance(configuration);
  21.         //2. 设置类路径
  22.         instance.setJarByClass(FlowDriver.class);
  25.         //3. 设置Mapper和Reducer
  26.         instance.setMapperClass(FlowMapper.class);
  27.         instance.setReducerClass(FlowReducer.class);
  29.         //4. 设置输出类型
  30.         instance.setMapOutputKeyClass(Flow.class);
  31.         instance.setMapOutputValueClass(Text.class);
  33.         instance.setOutputKeyClass(Text.class);
  34.         instance.setOutputValueClass(Flow.class);
  36.         //5. 设置输入输出路径
  37.         FileInputFormat.setInputPaths(instance, new Path("d:\\Hadoop_test"));
  38.         FileOutputFormat.setOutputPath(instance, new Path("d:\\Hadoop_test_out"));
  40.         //6. 提交
  41.         boolean b = instance.waitForCompletion(true);
  42.         System.exit(b ? 0 : 1);
  43.     }
  44. }

 

结果

 

 4.GroupingComparator案例

     订单id           商品id          商品金额        

  1. 0000001    Pdt_01    222.8
  2. 0000002    Pdt_05    722.4
  3. 0000001    Pdt_02    33.8
  4. 0000003    Pdt_06    232.8
  5. 0000003    Pdt_02    33.8
  6. 0000002    Pdt_03    522.8
  7. 0000002    Pdt_04    122.4

求出每一个订单中最贵的商品

需求分析:

1) 将订单id和商品金额作为key,在Map阶段先用订单id升序排序,如果订单id相同,再用商品金额降序排序

2) 在Reduce阶段,用groupingComparator按照订单分组,每一组的第一个即是最贵的商品

 

先定义bean对象,重写序列化反序列话排序方法

  1. package com.nty.groupingComparator;
  3. import org.apache.hadoop.io.WritableComparable;
  5. import java.io.DataInput;
  6. import java.io.DataOutput;
  7. import java.io.IOException;
  9. /**
  10.  * author nty
  11.  * date time 2018-12-12 18:07
  12.  */
  13. public class Order implements WritableComparable<Order> {
  15.     private String orderId;
  17.     private String productId;
  19.     private double price;
  21.     public String getOrderId() {
  22.         return orderId;
  23.     }
  25.     public Order setOrderId(String orderId) {
  26.         this.orderId = orderId;
  27.         return this;
  28.     }
  30.     public String getProductId() {
  31.         return productId;
  32.     }
  34.     public Order setProductId(String productId) {
  35.         this.productId = productId;
  36.         return this;
  37.     }
  39.     public double getPrice() {
  40.         return price;
  41.     }
  43.     public Order setPrice(double price) {
  44.         this.price = price;
  45.         return this;
  46.     }
  48.     @Override
  49.     public String toString() {
  50.         return orderId + "\t" + productId + "\t" + price;
  51.     }
  54.     @Override
  55.     public int compareTo(Order o) {
  56.         //先按照订单排序,正序
  57.         int compare = this.orderId.compareTo(o.getOrderId());
  58.         if(0 == compare){
  59.             //订单相同,再比较价格,倒序
  60.             return Double.compare( o.getPrice(),this.price);
  61.         }
  62.         return compare;
  63.     }
  65.     @Override
  66.     public void write(DataOutput out) throws IOException {
  67.         out.writeUTF(orderId);
  68.         out.writeUTF(productId);
  69.         out.writeDouble(price);
  70.     }
  72.     @Override
  73.     public void readFields(DataInput in) throws IOException {
  74.         this.orderId = in.readUTF();
  75.         this.productId = in.readUTF();
  76.         this.price = in.readDouble();
  77.     }
  78. }

Mapper类

  1. package com.nty.groupingComparator;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.NullWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Mapper;
  8. import java.io.IOException;
  10. /**
  11.  * author nty
  12.  * date time 2018-12-12 18:07
  13.  */
  14. public class OrderMapper extends Mapper<LongWritable, Text, Order, NullWritable> {
  16.     private Order order = new Order();
  18.     @Override
  19.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  20.         //0000001    Pdt_01    222.8
  21.         //分割行数据
  22.         String[] fields = value.toString().split("\t");
  24.         //为order赋值
  25.         order.setOrderId(fields[0]).setProductId(fields[1]).setPrice(Double.parseDouble(fields[2]));
  27.         //写出
  28.         context.write(order,NullWritable.get());
  29.     }
  30. }

GroupingComparator类

  1. package com.nty.groupingComparator;
  3. import org.apache.hadoop.io.WritableComparable;
  4. import org.apache.hadoop.io.WritableComparator;
  6. /**
  7.  * author nty
  8.  * date time 2018-12-12 18:08
  9.  */
  10. public class OrderGroupingComparator extends WritableComparator {
  12.     //用作比较的对象的具体类型
  13.     public OrderGroupingComparator() {
  14.         super(Order.class,true);
  15.     }
  17.     //重写的方法要选对哦,一共有三个,选择参数为WritableComparable的方法
  18.     //默认的compare方法调用的是a,b对象的compare方法,但是现在我们排序和分组的规则不一致,所以要重写分组规则
  19.     @Override
  20.     public int compare(WritableComparable a, WritableComparable b) {
  21.         Order oa = (Order) a;
  22.         Order ob = (Order) b;
  23.         //按照订单id分组
  24.         return oa.getOrderId().compareTo(ob.getOrderId());
  25.     }
  26. }

Reducer类

  1. package com.nty.groupingComparator;
  3. import org.apache.hadoop.io.NullWritable;
  4. import org.apache.hadoop.mapreduce.Reducer;
  6. import java.io.IOException;
  8. /**
  9.  * author nty
  10.  * date time 2018-12-12 18:07
  11.  */
  12. public class OrderReducer extends Reducer<Order, NullWritable,Order, NullWritable> {
  14.     @Override
  15.     protected void reduce(Order key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
  16.         //每一组的第一个即是最高价商品,不需要遍历
  17.         context.write(key, NullWritable.get());
  18.     }
  19. }

Driver类

  1. package com.nty.groupingComparator;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.NullWritable;
  7. import org.apache.hadoop.mapreduce.Job;
  8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  9. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  11. import java.io.IOException;
  13. /**
  14.  * author nty
  15.  * date time 2018-12-12 18:07
  16.  */
  17. public class OrderDriver {
  19.     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
  20.         //1获取实例
  21.         Configuration configuration = new Configuration();
  22.         Job job = Job.getInstance(configuration);
  24.         //2设置类路径
  25.         job.setJarByClass(OrderDriver.class);
  27.         //3.设置Mapper和Reducer
  28.         job.setMapperClass(OrderMapper.class);
  29.         job.setReducerClass(OrderReducer.class);
  31.         //4.设置自定义分组类
  32.         job.setGroupingComparatorClass(OrderGroupingComparator.class);
  34.         //5. 设置输出类型
  35.         job.setMapOutputKeyClass(Order.class);
  36.         job.setMapOutputValueClass(NullWritable.class);
  38.         job.setOutputKeyClass(Order.class);
  39.         job.setOutputValueClass(NullWritable.class);
  41.         //6. 设置输入输出路径
  42.         FileInputFormat.setInputPaths(job, new Path("d:\\Hadoop_test"));
  43.         FileOutputFormat.setOutputPath(job, new Path("d:\\Hadoop_test_out"));
  45.         //7. 提交
  46.         boolean b = job.waitForCompletion(true);
  47.         System.exit(b ? 0 : 1);
  48.     }
  49. }

输出结果

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Hadoop热门文章

  • Hive中的去重 ：distinct,group by与ROW_Number()窗口函数
  • HBASE的安装过程及运行HBASE程序的需要配置的内容
  • hadoop--Unable to load native-hadoop library for your platform解决方法
  • Hive sampling 语法之TABLESAMPLE用法理解
  • idea项目运行时报错提示Could not locate executable null\bin\winutils.exe in the Hadoop binaries
  • hadoop 无法访问50070端口问题解决
  • Hadoop运行时遇到java.io.FileNotFoundException错误的解决方法
  • KUDU数据导入尝试一：TextFile数据导入Hive,Hive数据导入KUDU
  • Hive安装
  • Hadoop 学习笔记 生态

    • Hadoop推荐文章

  • 基于Hive的大数据分析系统
  • HBase 在统一内容平台业务的优化实践
  • hive窗口分析函数使用详解系列一
  • 两台实体机器4个虚拟机节点的Hadoop集群搭建（Ubuntu版）
  • 在阿里云和腾讯云的轻量应用服务器上搭建Hadoop集群
  • Haproxy搭建 Web 群集实现负载均衡
  • HBase Compaction 原理与线上调优实践
  • Hive执行计划之hive依赖及权限查询和常见使用场景
  • 工良出品：包教会，Hadoop、Hive 搭建部署简易教程
  • Centos7系统编译Hadoop3.3.4

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号