0

我只是在使用 3 机器集群测试字数统计示例。我的代码与此示例相同，但以下内容除外：

我在“output.collect（key，new IntWritable（sum））”行之前的reducer代码中添加了两行代码：

System.out.println(key);
key.set(key + " - Key in Reducer");

然后我检查了我的减速器日志（最后 8 K，我发现了这个：

3M3WI - Key in Reducer - Key in Reducer
3M3WIG - Key in Reducer - Key in Reducer
3M3WL - Key in Reducer - Key in Reducer
3M3WNWPLG - Key in Reducer - Key in Reducer
3M3WQ - Key in Reducer - Key in Reducer
3M3WQNG.K78QJ0WN, - Key in Reducer - Key in Reducer
3M3WWR - Key in Reducer - Key in Reducer
3M3WX - Key in Reducer - Key in Reducer
3M3X - Key in Reducer - Key in Reducer
3M3X,. - Key in Reducer - Key in Reducer
3M3X.KZA8J - Key in Reducer - Key in Reducer
3M3X1 - Key in Reducer - Key in Reducer
3M3X8RC - Key in Reducer - Key in Reducer
3M3XC - Key in Reducer - Key in Reducer
3M3XCBD9R337PK - Key in Reducer - Key in Reducer
3M3XD - Key in Reducer - Key in Reducer
3M3XLW - Key in Reducer - Key in Reducer
3M3XML - Key in Reducer - Key in Reducer
3M3XN - Key in Reducer - Key in Reducer
3M3XU - Key in Reducer - Key in Reducer
3M3XX - Key in Reducer - Key in Reducer
3M3XZ - Key in Reducer - Key in Reducer
3M3Y - Key in Reducer - Key in Reducer
3M3YAIJL - Key in Reducer - Key in Reducer

这意味着我的减速器输出再次输入减速器。这应该是hadoop的工作方式吧？它不应该是迭代的......而且我的代码与 hadoop.apache.com 网站中的示例相同......

有没有人遇到同样的问题？

附上我所有的代码，大部分与示例相同。

package test;

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.mapred.*;
import org.apache.hadoop.util.*;

public class WordCount {

    public static class Map extends MapReduceBase implements 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, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
          word.set(tokenizer.nextToken());
          output.collect(word, one);
        }
      }
    }

    public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
      public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        int sum = 0;
        while (values.hasNext()) {
          sum += values.next().get();
        }
        System.out.println(key);
        key.set(key+" - Key in Reducer");
        output.collect(key, new IntWritable(sum));
      }
    }

    public static void main(String[] args) throws Exception {
      JobConf conf = new JobConf(WordCount.class);
      conf.setJobName("wordcount");

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

      conf.setMapperClass(Map.class);
      conf.setCombinerClass(Reduce.class);
      conf.setReducerClass(Reduce.class);

      conf.setInputFormat(TextInputFormat.class);
      conf.setOutputFormat(TextOutputFormat.class);

      FileInputFormat.setInputPaths(conf, new Path(args[0]));
      FileOutputFormat.setOutputPath(conf, new Path(args[1]));

      JobClient.runJob(conf);
    }
}
4

1 回答 1

2

注释掉conf.setCombinerClass(Reduce.class); 应该没问题。发生这种情况是因为您也将 Reducer 用作您的 Combiner。

当组合器可用时，map() 的输出首先被馈送到 combine()。然后将 combine() 函数的输出发送到 reducer 机器上的 reduce() 函数。因此，reduce() 的实际输入已经包含在 Reducer 中的 1 个 Key，在通过 reduce() 后会加倍。这就是为什么您两次获得Reducer 中的Key 。

于 2013-09-16T18:59:12.263 回答