java - hadoop レデューサーの出力がレデューサーに繰り返し読み込まれました

Question

3 台のマシン クラスターを使用して単語カウントの例をテストしているところです。私のコードは、次を除いてこの例と同じです:

「output.collect(key, new IntWritable(sum))」行の前に、リデューサー コードに 2 行のコードを追加します。

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 Webサイトの例と同じです...

誰も同じ問題に遭遇しますか?

例とほとんど同じで、すべてのコードを添付しました。

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