java - Reducer setup() 的 Mapper 是做什么用的？

Question

设置和清理方法究竟用于什么？我试图找出它们的意思，但还没有人准确描述它们的作用。例如，设置方法如何使用来自输入拆分的数据？它把它当作一个整体吗？还是逐行？

Answer 1

如前所述，setup() 和 cleanup() 是您可以重写的方法，如果您选择的话，它们可用于初始化和清理您的映射/ reduce task 。在这些阶段，您实际上无法直接访问来自输入拆分的任何数据。 map/reduce 任务的生命周期是(从程序员的角度来看):

设置 -> map -> 清理

设置 -> 减少 -> 清理

在 setup() 期间通常发生的事情是您可以从配置对象中读取参数以自定义您的处理逻辑。

cleanup() 期间通常发生的事情是清理您可能已分配的所有资源。还有其他用途，就是清除任何累积的聚合结果。

setup() 和 cleanup() 方法对您(开发人员/程序员)来说只是“钩子(Hook)”，让您有机会在您的操作之前和之后做一些事情映射/ reduce task 。

例如，在规范的字数统计示例中，假设您要将某些字词排除在外(例如，“the”、“a”、“be”等停用词)。当您配置 MapReduce 作业时，您可以将这些单词的列表(以逗号分隔)作为参数(键值对)传递到配置对象中。然后在您的 map 代码中，在 setup() 期间，您可以获取停用词并将它们存储在某个全局变量( map task 的全局变量)中，并在您的 map 逻辑中排除对这些词的计数。这是来自 http://wiki.apache.org/hadoop/WordCount 的修改示例.

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();
    private Set<String> stopWords;

    protected void setup(Context context) throws IOException, InterruptedException {
        Configuration conf = context.getConfiguration();

        stopWords = new HashSet<String>();
        for(String word : conf.get("stop.words").split(",")) {
            stopWords.add(word);
        }
    }

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

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

    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
      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();
    conf.set("stop.words", "the, a, an, be, but, can");

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