输入数据

group.txt

0000001	Pdt_01	222.8
0000002	Pdt_05	722.4
0000001	Pdt_02	33.8
0000003	Pdt_06	232.8
0000003	Pdt_02	33.8
0000002	Pdt_03	522.8
0000002	Pdt_04	122.4

期望结果

期望输出数据

1	222.8
2	722.4
3	232.8

需求分析

自定义OrderBean

package com.mr.groupcomparator;

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 int  orderId;
    private double money;

    public OrderBean() {
    }

    public OrderBean(int orderId, double money) {
        this.orderId = orderId;
        this.money = money;
    }

    public int getOrderId() {
        return orderId;
    }

    public void setOrderId(int orderId) {
        this.orderId = orderId;
    }

    public double getMoney() {
        return money;
    }

    public void setMoney(double money) {
        this.money = money;
    }

    @Override
    public String toString() {
        return
                " " + orderId +
                " "  +money
                ;
    }

    @Override
    public int compareTo(OrderBean orderBean) {
       int compare = Integer.compare(this.orderId, orderBean.orderId);
        if (compare == 0) {
            return Double.compare(orderBean.money, this.money);
       } else {
           return compare;
        }

    }


    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(orderId);
        out.writeDouble(money);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.orderId = in.readInt();
        this.money = in.readDouble();
    }
}

自定义OrderComparator

package com.mr.groupcomparator;

import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableComparable;

public class OrderComparator extends WritableComparator {


    protected OrderComparator() {
        super(OrderBean.class, true);
    }

    //  分组的逻辑：针对同一个订单id的数据，我们认为是一个组里面的
    @Override
     //Object--转换为WritebaleCompareble 不然没得比较哎
    public int compare(WritableComparable a, WritableComparable b) {

         //qiang
        OrderBean orderBean = (OrderBean)  a;
        OrderBean orderBean2 = (OrderBean) b;

         return Integer.compare(orderBean.getOrderId(), orderBean2.getOrderId());



    }
}

Mapper类

package com.mr.groupcomparator;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;


public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {

    OrderBean orderBean = new OrderBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] split = value.toString().split("\t");
        orderBean.setOrderId(Integer.parseInt(split[0]));
        orderBean.setMoney(Double.parseDouble(split[2]));

        context.write(orderBean, NullWritable.get());

    }
}

Reducer类

在这里插入代码片

Driver类

package com.mr.groupcomparator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;


public class OrderDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 0 指定路径
        args = new String[]{"E:/Hadoop/src/main/resources/input/groupcomparator", "E:/Hadoop/src/main/resources/ouput/groupcomparator"};

// 1 获取配置信息configuration以及封装任务job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

// 2 设置Driver加载路径 setJarByClass
        job.setJarByClass(OrderDriver.class);
// 3 设置map和reduce类 setMaper setReducer
        job.setMapperClass(OrderMapper.class);
        job.setReducerClass(OrderReducer.class);
// 4 设置map输出   setmapoutputkey  setmapoutputvalue
        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);
// 5 设置最终输出kv类型 (reducer的输出kv类型) setoutoutkey  setoutputvalue
        job.setOutputKeyClass(OrderBean.class);
        job.setOutputValueClass(NullWritable.class);
// 6 设置本地的输入和输出路径   fileinputformat.setinputpath
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

//        一定要手动设置一下分组的类
        job.setGroupingComparatorClass(OrderComparator.class);

// 7 提交
        boolean completion = job.waitForCompletion(true);
        System.exit(completion ? 0 : 1);
    }
}

执行结果

第二种方法（比较简单）

Mapper类

package com.mr.nogroupcomparator;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class GroupMapper extends Mapper<Text,Text,Text, DoubleWritable> {
    Text k = new Text ();
    DoubleWritable v = new DoubleWritable ();

    @Override
    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
        String id = key.toString ();
        // String number = id.substring (6, 7);
        k.set (id);

        String[] price_name = value.toString().split ("\t");
        String price = price_name[1];
        v.set (Double.parseDouble (price));

        context.write (k,v);
    }
}

Reducer类

package com.mr.nogroupcomparator;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class GroupReducer extends Reducer<Text, DoubleWritable,Text, DoubleWritable> {
    DoubleWritable doubleWritable = new DoubleWritable ();


    @Override
    protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
        double max=0;
        for (DoubleWritable value : values) {
            if (value.get ()>max){
                max = value.get ();
            }
            doubleWritable.set (max);

        }
        context.write (key,doubleWritable);
    }
}

Driver类

package com.mr.nogroupcomparator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueLineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class GroupDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        args = new String[]{"E:/Hadoop/src/main/resources/input/twogroup","E:/Hadoop/src/main/resources/ouput/twogroup"};
        Configuration conf = new Configuration ();
        conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t");
        //获取job对象
        Job job = Job.getInstance (conf);

        // 设置输入格式
        job.setInputFormatClass(KeyValueTextInputFormat.class);



        job.setJarByClass (GroupDriver.class);

        job.setMapperClass (GroupMapper.class);
        job.setReducerClass (GroupReducer.class);

        job.setMapOutputKeyClass (Text.class);
        job.setMapOutputValueClass (DoubleWritable.class);

        job.setOutputKeyClass (Text.class);
        job.setOutputValueClass (DoubleWritable.class);

        FileInputFormat.setInputPaths (job,new Path (args[0]));
        FileOutputFormat.setOutputPath (job,new Path (args[1]));


        boolean wait = job.waitForCompletion (true);
        System.exit (wait?0:1);

    }
}