MR基础案例（六）TOP N

朱雀 2024-04-18 12:55 60阅读 0赞

**任务需求**

需要求出wordcount的前3名 的数据  
TopN.java

package MR;
    
    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.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    import java.util.TreeSet;
    
    public class TopN {
        
        //自定义的mapper类
        public static class MyMapper extends Mapper<Object, Text, Text, LongWritable> {
        
            Text k = new Text();
            LongWritable v = new LongWritable(1L);
            /**
             * 抽象map函数   (map阶段的核心业务逻辑)
             */
            @Override
            protected void map(Object key, Text value,Context context) throws IOException, InterruptedException {
        
                //获取行值
                String row = value.toString();
                /**
                 * 
                 */
                //拆分行值
                String [] words = row.split(" ");
                for (String s : words) {
        
                    //输出
                    k.set(s);
                    context.write(k, v);
                }
            }
        }
    
    
        //自定义reducer类
        public static class MyReducer extends Reducer<Text, LongWritable, Text, IntWritable> {
        
            IntWritable v = new IntWritable();
            final static int N = 3;
            /**
             * 抽象的reduce函数 (reduce阶段的核心业务逻辑)
             */
            //定义一个存储前N个 3
            TreeSet<TopNWritable> ts = new TreeSet<TopNWritable>();
            @Override
            protected void reduce(Text key, Iterable<LongWritable> value,Context context) throws IOException, InterruptedException {
        
    
                //定义计数器
                int counter = 0;
                for (LongWritable i : value) {
        
                    counter += i.get();
                }
                v.set(counter);
    
                //将统计值封装对象并存储ts中
                TopNWritable tw = new TopNWritable(key.toString(),counter);
                //对象添加到ts
                ts.add(tw);
                if(ts.size() > N){
        
                    ts.remove(ts.last());
                }
                //reduce的输出
                //context.write(key, v);
            }
    
            @Override
            protected void cleanup(Context context) throws IOException, InterruptedException {
        
    
                for (TopNWritable tn : ts){
        
                    context.write(new Text(tn.getWord()), new IntWritable(tn.getValue()));
                }
            }
        }
    
    
    
        /**
         * 驱动方法
         * @param args
         */
        public static void main(String[] args) {
        
            try {
        
                //1、获取配置对象并进行属性设置
                Configuration conf = new Configuration();
                //对conf设置
                //2、获取job
                Job job = Job.getInstance(conf, "topN");
                //3、对job设置运行主类
                job.setJarByClass(TopN.class);
    
                //4、对job的map端属性设置
                job.setMapperClass(MyMapper.class);
                job.setMapOutputKeyClass(Text.class);
                job.setMapOutputValueClass(LongWritable.class);
    
                //5、对job的reduce端的属性设置
                job.setReducerClass(MyReducer.class);
                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(IntWritable.class);
    
    
                //6、设置job的输入路径和输出路径
                FileInputFormat.addInputPath(job, new Path(args[0]));
                FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
                //7、提交作业
                int success = job.waitForCompletion(true) ? 0 : 1;
                //8、退出
                System.exit(success);
            } catch (IOException | ClassNotFoundException | InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

TopNWritable.java

package MR;
    
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    import java.util.Objects;
    
    /**
     *
     */
    public class TopNWritable implements WritableComparable<TopNWritable> {
        
        private String word;
        private int value;
    
        public TopNWritable(){
        }
    
        public TopNWritable(String word, int value) {
        
            this.word = word;
            this.value = value;
        }
    
        public void write(DataOutput out) throws IOException {
        
            out.writeUTF(this.word);
            out.writeInt(this.value);
        }
    
        public void readFields(DataInput in) throws IOException {
        
            this.word = in.readUTF();
            this.value = in.readInt();
        }
    
        public String getWord() {
        
            return word;
        }
    
        public void setWord(String word) {
        
            this.word = word;
        }
    
        public int getValue() {
        
            return value;
        }
    
        public void setValue(int value) {
        
            this.value = value;
        }
    
        public int compareTo(TopNWritable o) {
        
            int tmp =  o.value - this.value;   //降序
            if(tmp != 0){
        
                return tmp;
            }
            return this.word.compareTo(o.word);
            //return o.second - this.second; //降序
        }
    
        @Override
        public boolean equals(Object o) {
        
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            TopNWritable that = (TopNWritable) o;
            return value == that.value &&
                    Objects.equals(word, that.word);
        }
    
        @Override
        public int hashCode() {
        
    
            return Objects.hash(word, value);
        }
    }