MapReduce实战案例(3)

朱雀 2023-10-11 23:05 123阅读 0赞

![1127bab70e0ead8a9dc622c15cbf12fa.jpeg][]

#### 案例三: MR实战之TOPN(自定义GroupingComparator) ####

项目准备

1.  需求+测试数据

有如下订单数据

<table> 
 <thead> 
  <tr> 
   <th>订单id</th> 
   <th>商品id</th> 
   <th>成交金额</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Order_0000001</td> 
   <td>Pdt_01</td> 
   <td>222.8</td> 
  </tr> 
  <tr> 
   <td>Order_0000001</td> 
   <td>Pdt_05</td> 
   <td>25.8</td> 
  </tr> 
  <tr> 
   <td>Order_0000002</td> 
   <td>Pdt_03</td> 
   <td>522.8</td> 
  </tr> 
  <tr> 
   <td>Order_0000002</td> 
   <td>Pdt_04</td> 
   <td>122.4</td> 
  </tr> 
  <tr> 
   <td>Order_0000002</td> 
   <td>Pdt_05</td> 
   <td>722.4</td> 
  </tr> 
  <tr> 
   <td>Order_0000003</td> 
   <td>Pdt_01</td> 
   <td>222.8</td> 
  </tr> 
 </tbody> 
</table>

现在需要求出每一个订单中成交金额最大的一笔交易

1.  分析
    
    a) 利用“订单id和成交金额”作为key，可以将map阶段读取到的所有订单数据按照id分区，按照金额排序，发送到reduce
    
    b) 在reduce端利用groupingcomparator将订单id相同的kv聚合成组，然后取第一个即是最大值

项目实现

a)自定义groupingcomparator

/**
     * @Author 千锋大数据教学团队
     * @Company 千锋好程序员大数据
     * @Description 用于控制shuffle过程中reduce端对kv对的聚合逻辑
     */
    public class ItemidGroupingComparator extends WritableComparator {
    
        protected ItemidGroupingComparator() {
    
            super(OrderBean.class, true);
        }
    
    
        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            OrderBean abean = (OrderBean) a;
            OrderBean bbean = (OrderBean) b;
    
            //将item_id相同的bean都视为相同，从而聚合为一组
            return abean.getItemid().compareTo(bbean.getItemid());
        }
    }
    复制代码

**文末扫码领取福利！**

b)定义订单信息bean

/**
     * @Author 千锋大数据教学团队
     * @Company 千锋好程序员大数据
     * @Description 订单信息bean，实现hadoop的序列化机制
     */
    public class OrderBean implements WritableComparable<OrderBean>{
        private Text itemid;
        private DoubleWritable amount;
    
        public OrderBean() {
        }
        public OrderBean(Text itemid, DoubleWritable amount) {
            set(itemid, amount);
        }
    
        public void set(Text itemid, DoubleWritable amount) {
    
            this.itemid = itemid;
            this.amount = amount;
    
        }
    
        public Text getItemid() {
            return itemid;
        }
    
        public DoubleWritable getAmount() {
            return amount;
        }
    
        @Override
        public int compareTo(OrderBean o) {
            int cmp = this.itemid.compareTo(o.getItemid());
            if (cmp == 0) {
    
                cmp = -this.amount.compareTo(o.getAmount());
            }
            return cmp;
        }
    
        @Override
        public void write(DataOutput out) throws IOException {
            out.writeUTF(itemid.toString());
            out.writeDouble(amount.get());
    
        }
    
        @Override
        public void readFields(DataInput in) throws IOException {
            String readUTF = in.readUTF();
            double readDouble = in.readDouble();
    
            this.itemid = new Text(readUTF);
            this.amount= new DoubleWritable(readDouble);
        }
    
    
        @Override
        public String toString() {
            return itemid.toString() + "\t" + amount.get();
        }
    }
    复制代码

c) 编写MapReduce处理流程

/**
     * @Author 千锋大数据教学团队
     * @Company 千锋好程序员大数据
     * @Description 利用secondarysort机制输出每种item订单金额最大的记录
     */
    
    public class SecondarySort {
    
        static class SecondarySortMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable>{
    
            OrderBean bean = new OrderBean();
    
            @Override
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
                String line = value.toString();
                String[] fields = StringUtils.split(line, "\t");
    
                bean.set(new Text(fields[0]), new DoubleWritable(Double.parseDouble(fields[1])));
    
                context.write(bean, NullWritable.get());
    
            }
    
        }
    
        static class SecondarySortReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable>{
    
    
            //在设置了groupingcomparator以后，这里收到的kv数据 就是：  <1001 87.6>,null  <1001 76.5>,null  .... 
            //此时，reduce方法中的参数key就是上述kv组中的第一个kv的key：<1001 87.6>
            //要输出同一个item的所有订单中最大金额的那一个，就只要输出这个key
            @Override
            protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
                context.write(key, NullWritable.get());
            }
        }
    
    
        public static void main(String[] args) throws Exception {
    
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            job.setJarByClass(SecondarySort.class);
    
            job.setMapperClass(SecondarySortMapper.class);
            job.setReducerClass(SecondarySortReducer.class);
    
    
            job.setOutputKeyClass(OrderBean.class);
            job.setOutputValueClass(NullWritable.class);
    
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            //指定shuffle所使用的GroupingComparator类
            job.setGroupingComparatorClass(ItemidGroupingComparator.class);
            //指定shuffle所使用的partitioner类
            job.setPartitionerClass(ItemIdPartitioner.class);
    
            job.setNumReduceTasks(3);
    
            job.waitForCompletion(true);
    
        }
    
    }

![2fd9078adb8148c59ceb1ef1f93c44b6.jpeg][]

也可以观看视频：

千锋大数据Hadoop全新增强版-先导片

[1127bab70e0ead8a9dc622c15cbf12fa.jpeg]: https://img-blog.csdnimg.cn/img_convert/1127bab70e0ead8a9dc622c15cbf12fa.jpeg
[2fd9078adb8148c59ceb1ef1f93c44b6.jpeg]: https://img-blog.csdnimg.cn/2fd9078adb8148c59ceb1ef1f93c44b6.jpeg