title: Hadoop mapper and reducer output mismatch
body: <p>I am trying to configure different mapper and reducer output type by using
  <code>setMapOutputKeyClass</code>, <code>setMapOutputValueClass</code>, <code>setMapKeyClass</code>, and
  <code>setMapValueClass</code>. However, even after I called these functions, I still get
  error messages during the runtime. </p>
  
  <p>Here is my code:</p>
  
  <pre><code>package org.myorg;
  
  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 Sort {
  
      public static class Map extends MapReduceBase implements Mapper&lt;LongWritable, Text, Text, IntWritable&gt; {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
  
        public void map(LongWritable key, Text value, OutputCollector&lt;Text, IntWritable&gt; 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&lt;Text, IntWritable, Text, LongWritable&gt; {
        public void reduce(Text key, Iterator&lt;IntWritable&gt; values, OutputCollector&lt;Text, LongWritable&gt; output, Reporter reporter) throws IOException {
          int sum = 0;
          while (values.hasNext()) {
            sum += values.next().get();
          }
          output.collect(key, new LongWritable(sum));
        }
      }
  
      public static void main(String[] args) throws Exception {
        JobConf conf = new JobConf(Sort.class);
        conf.setJobName("sort");
  
        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]));
  
        conf.setMapOutputKeyClass(Text.class);
        conf.setMapOutputValueClass(IntWritable.class);
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(LongWritable.class);
  
        JobClient.runJob(conf);
      }
  }
  </code></pre>
  
  <p>The error message I got:</p>
  
  <pre><code>java.lang.Exception: java.io.IOException: wrong value class: class org.apache.hadoop.io.LongWritable is not class org.apache.hadoop.io.IntWritable
  at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:354)
  Caused by: java.io.IOException: wrong value class: class org.apache.hadoop.io.LongWritable is not class org.apache.hadoop.io.IntWritable
  at org.apache.hadoop.mapred.IFile$Writer.append(IFile.java:168)
  at org.apache.hadoop.mapred.Task$CombineOutputCollector.collect(Task.java:1160)
  at org.myorg.Sort$Reduce.reduce(Sort.java:34)
  at org.myorg.Sort$Reduce.reduce(Sort.java:28)
  at org.apache.hadoop.mapred.Task$OldCombinerRunner.combine(Task.java:1436)
  at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.sortAndSpill(MapTask.java:1441)
  at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.flush(MapTask.java:1303)
  at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:431)
  at org.apache.hadoop.mapred.MapTask.run(MapTask.java:366)
  at org.apache.hadoop.mapred.LocalJobRunner$Job$MapTaskRunnable.run(LocalJobRunner.java:223)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
  at java.util.concurrent.FutureTask.run(FutureTask.java:262)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:724)
  13/10/12 14:08:11 INFO mapred.JobClient:  map 0% reduce 0%
  13/10/12 14:08:11 INFO mapred.JobClient: Job complete: job_local599611407_0001
  13/10/12 14:08:11 INFO mapred.JobClient: Counters: 0
  13/10/12 14:08:11 INFO mapred.JobClient: Job Failed: NA
  Exception in thread "main" java.io.IOException: Job failed!
  at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1357)
  at org.myorg.Sort.main(Sort.java:57)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:606)
  at org.apache.hadoop.util.RunJar.main(RunJar.java:160)
  </code></pre>
  
  <p>Did I do anything wrong? Thanks for your help!</p>