title: Debugging in Hadoop - MapReduce . Mapper not being called?
body: <p>I'v been teaching myself Hadoop over the past few days and was trying to implement a basic BFS algorithms according to the information given at <a href="http://hadooptutorial.wikispaces.com/Iterative+MapReduce+and+Counters#discussion" rel="nofollow">this webpage</a>. I had to do some modifications and additions to get the code compiled. I got the following error on running and even after spending hours debugging I could not solve this. Could someone help me with this?</p>
  
  <p>ERROR:</p>
  
  <pre><code>15/05/11 03:04:20 WARN mapred.LocalJobRunner: job_local934121164_0001
  java.lang.Exception: java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.Text, received org.apache.hadoop.io.LongWritable
      at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
      at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:522)
  Caused by: java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.Text, received org.apache.hadoop.io.LongWritable
      at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1072)
      at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:715)
      at org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write(TaskInputOutputContextImpl.java:89)
      at org.apache.hadoop.mapreduce.lib.map.WrappedMapper$Context.write(WrappedMapper.java:112)
      at org.apache.hadoop.mapreduce.Mapper.map(Mapper.java:125)
      at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:146)
      at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:787)
      at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
      at org.apache.hadoop.mapred.LocalJobRunner$Job$MapTaskRunnable.run(LocalJobRunner.java:243)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  15/05/11 03:04:21 INFO mapreduce.Job: Job job_local934121164_0001 running in uber mode : false
  15/05/11 03:04:21 INFO mapreduce.Job:  map 0% reduce 0%
  </code></pre>
  
  <p>Which should not happen as in the mapper as well as the reducer I'm following the same key , value types as you can see below. The only thing which I think is happning here is that my mapper class is not being used and instead the default one is used (which emits <strong>LongWritable</strong> key). I'm not sure what I'm doing wrong here.</p>
  
  <p><strong>SearchMapper.java</strong></p>
  
  <pre><code>import java.io.IOException;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Mapper.Context;
  
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.io.LongWritable;
  public class SearchMapper extends Mapper&lt;Object, Text, Text, Text&gt; {
  
      // Types of the input key, input value and the Context object through which 
      // the Mapper communicates with the Hadoop framework
      public void map(Object key, Text value, Context context, Node inNode)
              throws IOException, InterruptedException {
  
          // For each GRAY node, emit each of the adjacent nodes as a new node
          // (also GRAY) if the adjacent node is already processed
          // and colored BLACK, the reducer retains the color BLACK
          // Note that the mapper is not differentiating between BLACK GREY AND WHITE
  
          if (inNode.getColor() == Node.Color.GRAY) {
              for (String neighbor : inNode.getEdges()) { 
                  Node adjacentNode = new Node();
  
                  // Remember that the current node only has the value the id 
                  // of its neighbour, and not the object itself. Therefore at 
                  // this stage there is no way of knowing and assigning any of
                  // its other properties. Also remember that the reducer is doing
                  // the 'clean up' task and not the mapper.
                  adjacentNode.setId(neighbor); 
                  adjacentNode.setDistance(inNode.getDistance() + 1);
                  adjacentNode.setColor(Node.Color.GRAY);
                  adjacentNode.setParent(inNode.getId());
                  context.write(new Text(adjacentNode.getId()), adjacentNode.getNodeInfo()); // Get nodeinfo returns a Text Object
              }
              inNode.setColor(Node.Color.BLACK);
          }
          // Emit the input node, other wise the BLACK color change(if it happens)
          // Wont be persistent 
          context.write(new Text(inNode.getId()), inNode.getNodeInfo());
  
      }
  }
  </code></pre>
  
  <p><strong>SearchReducer.java</strong></p>
  
  <pre><code>import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.io.Text;
  import java.io.IOException;
  
  public class SearchReducer extends Reducer&lt;Text, Text, Text, Text&gt; {
  
      // Types of the input key, the values associated with the key, the Context object for Reducers communication
      // with the Hadoop framework and the node whose information has to be output
      // the return type is a Node
      public Node reduce(Text key, Iterable&lt;Text&gt; values, Context context, Node outNode)
              throws IOException, InterruptedException {
  
          // set the node id as the key
          outNode.setId(key.toString());
  
          // TODO : (huh?) Since the values are of the type Iterable, iterate through the values associated with the key
          // for all the values corresponding to a particular node id
  
          for (Text value : values) {
  
              Node inNode = new Node(key.toString() + "\t" + value.toString());
  
              // Emit one node after combining all the mapper outputs
  
              // Only one node(the original) will have a non-null adjascency list
              if (inNode.getEdges().size() &gt; 0) {
                  outNode.setEdges(inNode.getEdges());
              }
  
              // Save the minimum distance and parent
              if (inNode.getDistance() &lt; outNode.getDistance()) {
                  outNode.setDistance(inNode.getDistance());
                  outNode.setParent(inNode.getParent());
              }
  
              // Save the darkest color
              if (inNode.getColor().ordinal() &gt; outNode.getColor().ordinal()) {
                  outNode.setColor(inNode.getColor());
              }        
          }
          context.write(key, new Text(outNode.getNodeInfo()));      
          return outNode;
      }
  }
  </code></pre>
  
  <p><strong>BaseJob.java</strong> ( a generic class the website mentioned follows which basically sets the job up)</p>
  
  <pre><code>import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.io.Text;
  import java.io.IOException;
  
  public abstract class BaseJob extends Configured implements Tool {
  
          protected Job setupJob(String jobName,JobInfo jobInfo) throws Exception {
  
          Job job = new Job(new Configuration(), jobName);
          job.setJarByClass(jobInfo.getJarByClass());
  
          job.setMapperClass(jobInfo.getMapperClass());
          if (jobInfo.getCombinerClass() != null)
              job.setCombinerClass(jobInfo.getCombinerClass());
          job.setReducerClass(jobInfo.getReducerClass());
  
          // TODO : set number of reducers as required
          job.setNumReduceTasks(3);
  
          job.setOutputKeyClass(jobInfo.getOutputKeyClass());
          job.setOutputValueClass(jobInfo.getOutputValueClass());
         /*
          job.setJarByClass(SSSPJob.class);
          job.setMapperClass(SearchMapper.class);
          job.setReducerClass(SearchReducer.class);
          job.setNumReduceTasks(3);
          job.setOutputValueClass(Text.class);
          job.setOutputKeyClass(Text.class);*/
          return job;
      }
  
     // Implement an abstract class for JobInfo object
      protected abstract class JobInfo {
          public abstract Class&lt;?&gt; getJarByClass();
          public abstract Class&lt;? extends Mapper&gt; getMapperClass();
          public abstract Class&lt;? extends Reducer&gt; getCombinerClass();
          public abstract Class&lt;? extends Reducer&gt; getReducerClass();
          public abstract Class&lt;?&gt; getOutputKeyClass();
          public abstract Class&lt;?&gt; getOutputValueClass();
  
      }
  }
  </code></pre>
  
  <p><strong>SSSPJob.java</strong> (the driver)</p>
  
  <pre><code>import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Counters;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  import org.apache.hadoop.fs.Path;
  
  public class SSSPJob extends BaseJob {
      // method to set the configuration for the job and the mapper and the reducer classes
      private Job getJobConf(String[] args) 
          throws Exception {
  
      // Defining the abstract class objects
          JobInfo jobInfo = new JobInfo() {
              @Override
              public Class&lt;? extends Reducer&gt; getCombinerClass() {
                  return null;
              }
  
              @Override
              public Class&lt;?&gt; getJarByClass() {
                  return SSSPJob.class;
              }
  
              @Override
              public Class&lt;? extends Mapper&gt; getMapperClass() {
                  return SearchMapper.class;
              }
  
              @Override
              public Class&lt;?&gt; getOutputKeyClass() {
                  return Text.class;
              }
  
              @Override
              public Class&lt;?&gt; getOutputValueClass() {
                  return Text.class;
              }
  
              @Override
              public Class&lt;? extends Reducer&gt; getReducerClass() {
                  return SearchReducer.class;
              }
          };
  
          return setupJob("ssspjob", jobInfo);
  
      }
  
      // the driver to execute the job and invoke the map/reduce functions
  
      public int run(String[] args) throws Exception {
          int iterationCount = 0; 
          Job job;
          // No of grey nodes
          long terminationValue = 1;
  
          while( terminationValue &gt;0){
              job = getJobConf(args); 
              String input, output;
  
              // Setting the input file and output file for each iteration
              // During the first time the user-specified file will be the
              // input whereas for the subsequent iterations
              // the output of the previous iteration will be the input
              // NOTE: Please be clear of how the input output files are set
              //       before proceding.
  
              // for the first iteration the input will be the first input argument
              if (iterationCount == 0) 
                  input = args[0];
              else
                  // for the remaining iterations, the input will be the output of the previous iteration
                  input = args[1] + iterationCount;
  
              output = args[1] + (iterationCount + 1);
  
              FileInputFormat.setInputPaths(job, new Path(input));
              FileOutputFormat.setOutputPath(job, new Path(output));
  
              job.waitForCompletion(true); 
  
              Counters jobCntrs = job.getCounters();
              terminationValue = jobCntrs.findCounter(MoreIterations.numberOfIterations).getValue();
              // if the counter's value is incremented in the reducer(s), then there are more
              // GRAY nodes to process implying that the iteration has to be continued.
              iterationCount++;
          }
          return 0;
      }
  
      public static void main(String[] args) throws Exception {
  
          int res = ToolRunner.run(new Configuration(), new SSSPJob(), args);
          if(args.length != 2){
              System.err.println("Usage: &lt;in&gt; &lt;output name&gt; ");
              System.exit(1);
              System.out.println("Huh?");
          }
          System.exit(res);
      }
  
  }
  </code></pre>
  
  <p>And furthermore, Im not sure how debugging is done on hadoop. All my debug print statements seems not to have any effect and I suspect the hadoop framework writes the log messages to some other place or file.
  Thanks :)</p>