mapreduce二次排序

Myth丶恋晨 2022-07-16 04:22 174阅读 0赞

本代码几乎涵盖mapreduce的全部过程

package secondarSort;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Partitioner;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    
    public class SecondarSort { 
    
        /** * 自定义数据类型 * @author root * */
        public static class MyInt implements WritableComparable<MyInt>{ 
            private int firstInt;
            private int secondInt;
            public MyInt(){}
    
            public void set(int firstInt,int secondInt){
                this.firstInt = firstInt;
                this.secondInt = secondInt;
            }
    
            public MyInt(int firstInt,int secondInt){
                super();
                this.firstInt = firstInt;
                this.secondInt = secondInt;
    
            }
    
            @Override
            public String toString() {
                // TODO Auto-generated method stub
                return firstInt+"\t"+secondInt;
            }
    
            @Override
            public void write(DataOutput out) throws IOException {
                out.writeInt(firstInt);
                out.writeInt(secondInt);
    
            }
    
            @Override
            public void readFields(DataInput in) throws IOException {
                firstInt=in.readInt();
                secondInt = in.readInt();
    
            }
    
            @Override
            public int compareTo(MyInt myInt) {
                return this.getFirstInt()-myInt.getFirstInt();
            }
    
    
            //自动生成的set和get方法
            public int getFirstInt() {
                return firstInt;
            }
    
            public void setFirstInt(int firstInt) {
                this.firstInt = firstInt;
            }
    
            public int getSecondInt() {
                return secondInt;
            }
    
            public void setSecondInt(int secondInt) {
                this.secondInt = secondInt;
            }   
        }
    
        /** * map函数 * @author root * */
        public static class SSMap extends Mapper<LongWritable, Text, MyInt, IntWritable>{ 
            private MyInt myInt = new MyInt();
            private IntWritable intw = new IntWritable();
            @Override
            protected void map(LongWritable key, Text value,
                    Mapper<LongWritable, Text, MyInt, IntWritable>.Context context)
                            throws IOException, InterruptedException {
                String[] strs = value.toString().split(" ");
                myInt.set(Integer.parseInt(strs[0]), Integer.parseInt(strs[1]));
                intw.set(Integer.parseInt(strs[1]));
                context.write(myInt, intw);
            }
        }
    
        /** * 分区 * @author root * */
        public static class MyPartitioner extends Partitioner<MyInt, IntWritable>{ 
            @Override
            public int getPartition(MyInt key, IntWritable value, int numPartitions) {
    
                return key.getFirstInt();
            }   
        }
        /** * 分组，将值变成<k,list> 在reduce之前进行执行 构造方法 * @author root * */
        public static class MyGrouper extends WritableComparator{ 
    
            public MyGrouper(){
                super(MyInt.class,true);
            }
            @Override
            public int compare(WritableComparable a, WritableComparable b) {
                MyInt aa = (MyInt)a;
                MyInt bb = (MyInt)b;
                //返回0则分到同一个组里
                return aa.getFirstInt()-bb.getFirstInt()>0?1:-1;
            }
    
        }
    
        /** * 比较器 * @author root * */
        public static class MyComparator extends WritableComparator{ 
            public MyComparator(){
                super(MyInt.class,true);
            }
            @Override
            public int compare(WritableComparable a, WritableComparable b) {
                MyInt aa = (MyInt)a;
                MyInt bb = (MyInt)b;
    
                if(aa.getFirstInt()==bb.getFirstInt()){
                    return aa.getSecondInt()-bb.getSecondInt();
                }
    
                return aa.getFirstInt()-bb.getFirstInt();
            }
        }
    
        /** * reduce方法 * @author root * */
        public static class SSReduce extends Reducer<MyInt, IntWritable, Text, Text>{ 
            private Text k=new Text();
            private Text v=new Text();
    
            @Override
            protected void reduce(MyInt key, Iterable<IntWritable> value,
                    Reducer<MyInt, IntWritable, Text, Text>.Context context) throws IOException, InterruptedException {
                for(IntWritable i : value){
                    k.set(key.getFirstInt()+"");
                    v.set(i.toString());
                    context.write(k, v);
                }
            }
    
        }
    
        /** * main方法，启动程序 * @param args * @throws Exception */
        public static void main(String[] args) throws Exception {
            Configuration config = new Configuration();
            config.set("fs.defaultFS", "hdfs://192.168.8.8:9000/");
    
            Job job=Job.getInstance(config);
            //设置主方法
            job.setJarByClass(SecondarSort.class);
            //设置map方法
            job.setMapperClass(SSMap.class);
            //设置分区方法
            job.setPartitionerClass(MyPartitioner.class);
            //设置分组方法
            job.setGroupingComparatorClass(MyGrouper.class);
            //设置比较器
            job.setSortComparatorClass(MyComparator.class);
            //设置reduce 方法
            job.setReducerClass(SSReduce.class);
    
            job.setMapOutputKeyClass(MyInt.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
    
            FileInputFormat.addInputPath(job, new Path("/root/secondarySort.txt"));
            FileOutputFormat.setOutputPath(job, new Path("/result/secondarySort"));
    
            System.exit(job.waitForCompletion(true)?0:1);
        }
    }