美文网首页
MapReduce 基础 (十)分组

MapReduce 基础 (十)分组

作者: 做个合格的大厂程序员 | 来源:发表于2020-06-17 11:12 被阅读0次
    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
    

    假设有一组数据,我们要统计订单号金额最大的一笔给区分出来应该怎么做?

    这里是一个综合案例,我们既要将数据转换,也要将数据给分组。首先我们需要将数据拆分成一个对象,这个对象的一个属性是订单号,另一个属性是金额。

    OrderBean

    package cn.leon.mygrouping;
    
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public class OrderBean  implements WritableComparable<OrderBean>{
        private  String orderId;
        private  Double 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;
        }
    
        //指定排序规则
        @Override
        public int compareTo(OrderBean orderBean) {
            //先比较订单ID,如果订单ID一致,则排序订单金额(降序)
            int i = this.orderId.compareTo(orderBean.orderId);
            if(i == 0){
                i = this.price.compareTo(orderBean.price) * -1;
            }
    
            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();
        }
    }
    

    其次需要定义个Mapper将数据拆分

    GroupMapper

    package cn.leon.mygrouping;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    public class GroupMapper extends Mapper<LongWritable,Text,OrderBean,Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //1:拆分行文本数据,得到订单的ID,订单的金额
            String[] split = value.toString().split("\t");
    
            //2:封装OrderBean,得到K2
            OrderBean orderBean = new OrderBean();
            orderBean.setOrderId(split[0]);
            orderBean.setPrice(Double.valueOf(split[2]));
    
            //3:将K2和V2写入上下文中
            context.write(orderBean, value);
        }
    }
    

    进入shuffle阶段,第一步我们需要分区计算,我们可以将订单号作为分区的名称进行分区。

    OrderPartition

    package cn.leon.mygrouping;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class OrderPartition extends Partitioner<OrderBean,Text> {
        //分区规则: 根据订单的ID实现分区
    
        /**
         *
         * @param orderBean K2
         * @param text  V2
         * @param i  ReduceTask个数
         * @return 返回分区的编号
         */
        @Override
        public int getPartition(OrderBean orderBean, Text text, int i) {
            return (orderBean.getOrderId().hashCode() & 2147483647) % i;
        }
    }
    

    排序阶段在orderBean的对象中我们已经进行了就不必纠结了。规约在这里不需要考虑,最后应该考虑的就是分组了。

    OrderGroupComparator

    package cn.leon.mygrouping;
    
    /*
    
      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());
        }
    }
    

    分组阶段我们可以将订单号相同的数据进行合并。作为一组来考虑。所以我们这里用compare来进行订单号的对比。

    最后进入Reducer阶段直接将排序和分组之后的数据转换为K3,V3进行输出。这里因为我们只需要统计每个订单开销最大的数据。所以我们在此只执行一次循环即可。如果需要统计前2条我们只需要循环2次即可。

    GroupReducer

    package cn.leon.mygrouping;
    
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class GroupReducer extends Reducer<OrderBean,Text,Text,NullWritable> {
        @Override
        protected void reduce(OrderBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            int i = 0;
            for (Text value : values) {
                context.write(value, NullWritable.get());
                i++;
                if(i >= 1){
                    break;
                }
            }
        }
    }
    

    最后在主执行类中我们设置好输出,输入和分组,分区之后就可以正常的输出数据了。

    package cn.leon.mygrouping;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    
    public class JobMain extends Configured implements Tool {
        @Override
        public int run(String[] args) throws Exception {
            //1:获取Job对象
            Job job = Job.getInstance(super.getConf(), "mygroup_job");
    
            //2:设置job任务
                //第一步:设置输入类和输入路径
                job.setInputFormatClass(TextInputFormat.class);
                TextInputFormat.addInputPath(job, new Path("file:///D:\\input\\mygroup_input"));
    
                //第二步:设置Mapper类和数据类型
                job.setMapperClass(GroupMapper.class);
                job.setMapOutputKeyClass(OrderBean.class);
                job.setMapOutputValueClass(Text.class);
    
                //第三,四,五,六
                //设置分区
                job.setPartitionerClass(OrderPartition.class);
                //设置分组
                job.setGroupingComparatorClass(OrderGroupComparator.class);
    
                //第七步:设置Reducer类和数据类型
                job.setReducerClass(GroupReducer.class);
                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(NullWritable.class);
    
                //第八步:设置输出类和输出的路径
                job.setOutputFormatClass(TextOutputFormat.class);
                TextOutputFormat.setOutputPath(job, new Path("file:///D:\\out\\mygroup_out"));
    
            //3:等待job任务结束
            boolean bl = job.waitForCompletion(true);
    
    
    
            return bl ? 0: 1;
        }
    
        public static void main(String[] args) throws Exception {
            Configuration configuration = new Configuration();
    
            //启动job任务
            int run = ToolRunner.run(configuration, new JobMain(), args);
    
            System.exit(run);
        }
    }
    

    相关文章

      网友评论

          本文标题:MapReduce 基础 (十)分组

          本文链接:https://www.haomeiwen.com/subject/wewjxktx.html