Tipo mancata corrispondenza nella chiave dalla mappa: previsto org.apache.hadoop.io.Text, ricevuto org.apache.hadoop.io.LongWritable

Question

Sto provando a eseguire una mappa/riduttore in java. Qui di seguito sono i miei file

WordCount.java

package counter; 


public class WordCount extends Configured implements Tool { 

public int run(String[] arg0) throws Exception { 
    Configuration conf = new Configuration(); 

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

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

    job.setMapperClass(WordCountMapper.class); 
    job.setReducerClass(WordCountReducer.class); 

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

    FileInputFormat.addInputPath(job, new Path("counterinput")); 
    // Erase previous run output (if any) 
    FileSystem.get(conf).delete(new Path("counteroutput"), true); 
    FileOutputFormat.setOutputPath(job, new Path("counteroutput")); 

    job.waitForCompletion(true); 
    return 0; 
} 

public static void main(String[] args) throws Exception { 
    int res = ToolRunner.run(new Configuration(), new WordCount(), args); 
    System.exit(res); 

    } 
} 

WordCountMapper.java

public class WordCountMapper 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, OutputCollector<Text,IntWritable> output, Reporter reporter) 
    throws IOException, InterruptedException { 
     System.out.println("hi"); 
    String line = value.toString(); 
    StringTokenizer tokenizer = new StringTokenizer(line); 
    while (tokenizer.hasMoreTokens()) { 
     word.set(tokenizer.nextToken()); 
     output.collect(word, one); 
     } 
    } 
} 

WordCountReducer.java

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { 
    public void reduce(Text key, Iterator<IntWritable> values, 
     OutputCollector<Text,IntWritable> output, Reporter reporter) throws IOException, InterruptedException { 
     System.out.println("hello"); 
     int sum = 0; 
     while (values.hasNext()) { 
      sum += values.next().get(); 
     } 
     output.collect(key, new IntWritable(sum)); 
    } 
} 

sto ottenendo seguente errore

13/06/23 23:13:25 INFO jvm.JvmMetrics: Initializing JVM Metrics with 
processName=JobTracker, sessionId= 

13/06/23 23:13:25 WARN mapred.JobClient: Use GenericOptionsParser for parsing the 
arguments. Applications should implement Tool for the same. 
13/06/23 23:13:26 INFO input.FileInputFormat: Total input paths to process : 1 
13/06/23 23:13:26 INFO mapred.JobClient: Running job: job_local_0001 
13/06/23 23:13:26 INFO input.FileInputFormat: Total input paths to process : 1 
13/06/23 23:13:26 INFO mapred.MapTask: io.sort.mb = 100 
13/06/23 23:13:26 INFO mapred.MapTask: data buffer = 79691776/99614720 
13/06/23 23:13:26 INFO mapred.MapTask: record buffer = 262144/327680 
13/06/23 23:13:26 WARN mapred.LocalJobRunner: job_local_0001 
java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.Text, 
recieved org.apache.hadoop.io.LongWritable 
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:845) 
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:541) 
at org. 
apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80) 
at org.apache.hadoop.mapreduce.Mapper.map(Mapper.java:124) 
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144) 
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:621) 
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:305) 
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:177) 
13/06/23 23:13:27 INFO mapred.JobClient: map 0% reduce 0% 
13/06/23 23:13:27 INFO mapred.JobClient: Job complete: job_local_0001 
13/06/23 23:13:27 INFO mapred.JobClient: Counters: 0 

Penso che non sia in grado di trovare la classe Mapper e riduttore. Ho scritto il codice nella classe principale, Si sta ottenendo la classe predefinita di mapper e riduttore.

Answer 1

aggiungere queste 2 righe nel codice:

job.setMapOutputKeyClass(Text.class); 
job.setMapOutputValueClass(IntWritable.class); 

Si utilizza TextOutputFormat che emette chiave LongWritable e valore del testo di default, ma si emettono testo come chiave e IntWritable come valore. Devi dirlo ai lavori di famiglia.

HTH

Answer 2

Questo non può essere il problema, ma ho avuto questo problema una volta stupido. Assicurati di non mescolare le vecchie e le nuove librerie, ovvero mapred vs mapreduce. Annotare @Overide sulla mappa e ridurre i metodi. Se vedi errori, non stai sovrascrivendo correttamente i metodi.

Answer 3

ho ottenuto un simile stack eccezione a causa di improprio classe Mapper impostare nel mio codice (errore di battitura :))

job.setMapperClass(Mapper.class) // Set to org.apache.hadoop.mapreduce.Mapper due to type 

Si noti che per errore stavo usando classe Mapper dal pacchetto di MapReduce, ho cambiato il mio personalizzato classe mapper:

job.setMapperClass(LogProcMapperClass.class) // LogProcMapperClass is my custom mapper. 

L'eccezione viene risolta dopo aver corretto la classe mapper.