hadoop的WordCount按照value降序排序

左手的ㄟ右手 2022-05-29 07:52 240阅读 0赞

**[https://www.cnblogs.com/booth/p/3274593.html][https_www.cnblogs.com_booth_p_3274593.html]**

数据源

A    2
    B    9
    C    4
    D    9
    Z    42

要实现的输出

Z    42
    D    9
    B    9
    C    4
    A    2

看字符顺序，其实什么也没有，只是按照后面的数字进行一次倒序排序，  
　　实现思路:  
　　　　**1利用hadoop自带的排序功能**，  
　　　　**2.KV互换**

public class SVJob { 
        public static void main(String[] args) throws IOException,
                InterruptedException, ClassNotFoundException {
            Configuration conf = new Configuration();
            conf.set("mapred.job.tracker", "192.168.9.181:9001");
            String[] ars = new String[] {
                    "hdfs://192.168.9.181:9000/user/hadoop/input/examples/SortByValue/",
                    "hdfs://192.168.9.181:9000/user/hadoop/output/examples/SortByValue" };
            String[] otherArgs = new GenericOptionsParser(conf, ars)
                    .getRemainingArgs();
            if (otherArgs.length != 2) {
                System.err.println("SortByValue: <in> <out>");
                System.exit(2);
            }
    
            Job job = new Job(conf, "SortByValue");
            job.setJarByClass(SVJob.class);
            job.setMapperClass(SVMapper.class);
            job.setReducerClass(SVReducer.class);
    
            job.setMapOutputKeyClass(IntWritable.class);
            job.setMapOutputValueClass(Text.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            job.setSortComparatorClass(IntWritableDecreasingComparator.class);
    
            FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
            FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
    }

public class SVMapper extends Mapper<Object, Text, IntWritable, Text> { 
        protected void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString(); 
            String[] keyValueStrings = line.split("\t");
            if(keyValueStrings.length != 2)
            {
                //新手，不知道怎么记录日志，也不清楚怎么退出 各位大神如果知道请通知我，谢谢
                System.err.println("string format error!!!!!");
                return;
            }
            int outkey = Integer.parseInt(keyValueStrings[1]);
            String outvalue = keyValueStrings[0];
            context.write(new IntWritable(outkey), new Text(outvalue));
        }
    }

public class SVReducer extends Reducer<IntWritable, Text, Text, IntWritable> { 
        protected void reduce(IntWritable key, Iterable<Text> values,Context context)throws IOException, InterruptedException {
            for(Text value : values){
                context.write(value, key);
            }
        }
    }

public class IntWritableDecreasingComparator extends Comparator { 
        @SuppressWarnings("rawtypes")
        public int compare( WritableComparable a,WritableComparable b){
            return -super.compare(a, b);
        }
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
            return -super.compare(b1, s1, l1, b2, s2, l2);
        }
    }

[https_www.cnblogs.com_booth_p_3274593.html]: https://www.cnblogs.com/booth/p/3274593.html