hadoop程序只写逻辑，只写reducer的程序

Question

编写一个 map reduce 程序来打印文本文档中出现频率最高的单词。

可以固定阈值，输出频率超过阈值的词。 例如:如果 thereshold=100，并且“is”在文档中出现了 150 次，则必须在输出中打印它。

program :


    package org.myorg;

        import java.io.IOException;
        import java.util.*;

       import org.apache.hadoop.fs.Path;
       import org.apache.hadoop.conf.*;
       import org.apache.hadoop.io.*;
       import org.apache.hadoop.mapreduce.*;
       import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
       import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
       import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
       import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

   public class WordCount {

    public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
       private final static IntWritable one = new IntWritable(1);
       private Text word = new Text();

       public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
           String line = value.toString();
           StringTokenizer tokenizer = new StringTokenizer(line);
           while (tokenizer.hasMoreTokens()) {
               word.set(tokenizer.nextToken());
               context.write(word, one);
           }
       }
    } 

    public static class Reduce extends Reducer<Text, IntWritable, Text, Inritable> {

       public void reduce(Text key, Iterable<IntWritable> values, Context coext) 
         throws IOException, InterruptedException {
           int sum = 0;
           for (IntWritable val : values) {
               sum += val.get();
           }
           context.write(key, new IntWritable(sum));
       }
    }

    public static void main(String[] args) throws Exception {
       Configuration conf = new Configuration();

           Job job = new Job(conf, "wordcount");

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

      job.setMapperClass(Map.class);
       job.setReducerClass(Reduce.class);

      job.setInputFormatClass(TextInputFormat.class);
      job.setOutputFormatClass(TextOutputFormat.class);

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

       job.waitForCompletion(true);
    }
 }

Answer 1

完整代码如下

驱动类

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FrequentWordClassDriver extends Configured implements Tool{
    @Override
    public int run(String[] args) throws Exception {
        if(args.length != 2){
            return -1;
        }
        JobConf conf = new JobConf(getConf(), FrequentWordClassDriver.class);
        conf.setJobName(this.getClass().getName());
        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));
        conf.setMapperClass(FrequentWordClassMapper.class);
        conf.setReducerClass(FrequentWordClassReducer.class);
        conf.setMapOutputKeyClass(Text.class);
        conf.setMapOutputValueClass(IntWritable.class);
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(IntWritable.class);
        JobClient.runJob(conf);
        return 0;
    }
    public static void main(String[] args) throws Exception{
        int exitCode = ToolRunner.run(new FrequentWordClassDriver(), args);
        System.exit(exitCode);
    }
} 

映射器类

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

public class FrequentWordClassMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable>{
    @Override
    public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        String line = value.toString();
        for(String phrase : line.split(" ")){
            output.collect(new Text(phrase.toUpperCase()), new IntWritable(1)); 
        }
    }
}

reducer 类

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

public class FrequentWordClassReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable>{
    @Override
    public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException{
        int wordcount = 0, threshold = 100;
        while(values.hasNext()){
            IntWritable value = values.next();
            wordcount +=value.get();
        }
        if(wordcount >= threshold){
            output.collect(key, new IntWritable(wordcount));
        }   
    }
} 

Driver 类、Mapper 类 和Reducer 类 相当简单且不言自明。 mapper 类将每个句子拆分成单词，并以 的格式发送给 reducer 类。 reducer类会收到格式为 的数据，它会聚合统计每个词的出现次数，如果每个词的出现次数大于或等于阈值然后它会将单词作为输出发送。

希望对您有所帮助。