hadoop案例：groupcomparable（分组排序）

我就是我 2022-11-21 15:17 132阅读 0赞

### 文章目录 ###

*   *  输入数据
     *  期望结果
     *  需求分析
     *  自定义OrderBean
     *  自定义OrderComparator
     *  Mapper类
     *  Reducer类
     *  Driver类
     *  执行结果
     *  第二种方法（比较简单）
     *   *  Mapper类
         *  Reducer类
         *  Driver类

## 输入数据 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]

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

## 需求分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]

## 自定义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);
        }
    }

## 执行结果 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 2]

## 第二种方法（比较简单） ##

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/96080f914f7d4c049530c210e6a264e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/7741e970531e438cbcb4eca0858887c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/02b4650062b4429ea38c8dcad1b45486.png