clase de clave incorrecto: El texto no es IntWritable

Question

Esto puede parecer una pregunta estúpida, pero no veo el problema en mis tipos en mi código mapreduce para hadoop

Como se indica en la pregunta el problema es que es esperando IntWritable pero le estoy pasando un objeto de texto en collector.collect del reductor.

Mi configuración del trabajo tiene las siguientes clases de salida asignador:

conf.setMapOutputKeyClass(IntWritable.class); 
conf.setMapOutputValueClass(IntWritable.class); 

Y las siguientes clases de salida del reductor:

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

Mi clase de mapeo tiene la siguiente definición:

public static class Reduce extends MapReduceBase implements Reducer<IntWritable, IntWritable, Text, IntWritable> 

con la función requerida:

public void reduce(IntWritable key, Iterator<IntWritable> values, OutputCollector<Text,IntWritable> output, Reporter reporter) 

Y entonces se produce un error cuando llamo:

output.collect(new Text(),new IntWritable()); 

Soy bastante nuevo para mapear reducir, pero todos los tipos parecen coincidir, compila pero luego falla en esa línea diciendo que su esperando una IntWritable como la clave para la clase de reducción. Si importa que estoy usando la versión 0.21 de Hadoop

Aquí está mi clase de mapa:

public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, IntWritable, IntWritable> { 
    private IntWritable node = new IntWritable(); 
    private IntWritable edge = new IntWritable(); 

    public void map(LongWritable key, Text value, OutputCollector<IntWritable, IntWritable> output, Reporter reporter) throws IOException { 
     String line = value.toString(); 
     StringTokenizer tokenizer = new StringTokenizer(line); 

     while (tokenizer.hasMoreTokens()) { 
      node.set(Integer.parseInt(tokenizer.nextToken())); 
      edge.set(Integer.parseInt(tokenizer.nextToken())); 
      if(node.get() < edge.get()) 
       output.collect(node, edge); 
     } 
    } 
} 

y mi clase de reducir:

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

    IntWritable $ = new IntWritable(Integer.MAX_VALUE); 
    Text keyText = new Text(); 

    public void reduce(IntWritable key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { 
     ArrayList<IntWritable> valueList = new ArrayList<IntWritable>(); 

     //outputs original edge pair as key and $ for value 
     while (values.hasNext()) { 
      IntWritable value = values.next(); 
      valueList.add(value); 
      keyText.set(key.get() + ", " + value.get()); 
      output.collect(keyText, $); 
     } 

     //outputs all the 2 length pairs 
     for(int i = 0; i < valueList.size(); i++) 
      for(int j = i+1; i < valueList.size(); j++) 
       output.collect(new Text(valueList.get(i).get() + ", " + valueList.get(j).get()), key); 
    } 
} 

y la configuración de mi trabajo:

JobConf conf = new JobConf(Triangles.class); 
conf.setJobName("mapred1"); 

conf.setMapOutputKeyClass(IntWritable.class); 
conf.setMapOutputValueClass(IntWritable.class); 

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("mapred1")); 

JobClient.runJob(conf); 

Answer 1

Su problema es que establece la clase Reducir como un combinador

conf.setCombinerClass(Reduce.class); 

combinadores ejecutar en la fase mapa y que necesitan para emitir el mismo tipo de clave/valor (IntWriteable, IntWritable en su caso) elimine esta línea y que debe estar bien