2021年大数据Hadoop（二十二）：MapReduce的自定义分组

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;

    }

}

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;

    }

}

/*



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

    }

}

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

    }

}