title: Blank / Empty Values of Key and Value retrieved in the Mapper class
body: <p>I have written a MapReduce code for running it on a CDH4 cluster. My requirement was to read the complete file as the value and the file name as the key. For that I wrote custom InputFormat and RecordReader classes.</p>
  
  <p>Custom input format class: FullFileInputFormat.java</p>
  
  <pre><code>import java.io.*;
  
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.*;
  
  import FullFileRecordReader;
  
  public class FullFileInputFormat extends FileInputFormat&lt;Text, Text&gt; {
  
      @Override
      public RecordReader&lt;Text, Text&gt; getRecordReader(InputSplit split, JobConf jobConf, Reporter reporter) throws IOException {
          reporter.setStatus(split.toString());
          return new FullFileRecordReader((FileSplit) split, jobConf);
      }
  }
  </code></pre>
  
  <p>And the custom RecordReader class: FullFileRecordReader.java</p>
  
  <pre><code>import java.io.BufferedReader;
  import java.io.IOException;
  
  import org.apache.hadoop.fs.FSDataInputStream;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.*;
  
  public class FullFileRecordReader implements RecordReader&lt;Text, Text&gt; {
  
      private BufferedReader in;
      private boolean processed = false;
      private int processedBytes = 0;
  
      private FileSplit fileSplit;
      private JobConf conf;
  
      public FullFileRecordReader(FileSplit fileSplit, JobConf conf) {
          this.fileSplit = fileSplit;
          this.conf = conf;
      }
  
      @Override
      public void close() throws IOException {
          if (in != null) {
              in.close();
          }
      }
  
      @Override
      public Text createKey() {
          return new Text("");
      }
  
      @Override
      public Text createValue() {
          return new Text("");
      }
  
      @Override
      public long getPos() throws IOException {
          return processedBytes;
      }
  
      @Override
      public boolean next(Text key, Text value) throws IOException {
          Path filePath = fileSplit.getPath();
  
          if (!processed) {
              key = new Text(filePath.getName());
  
              value = new Text("");
              FileSystem fs = filePath.getFileSystem(conf);
              FSDataInputStream fileIn = fs.open(filePath);
              byte[] b = new byte[1024];
              int numBytes = 0;
  
              while ((numBytes = fileIn.read(b)) &gt; 0) {
                  value.append(b, 0, numBytes);
                  processedBytes += numBytes;
              }
              processed = true;
              return true;
          }
          return false;
      }
  
      @Override
      public float getProgress() throws IOException {
          return 0;
      }
  }
  </code></pre>
  
  <p>Though whenever I try to print the key-value in the RecordReader class, I get their values, but when I print the same in the mapper class, I see blank values for them. I am unable to understand why the Mapper class is unable to get any data for keys and values.</p>
  
  <p>Currently I have only a Map job and no reduce job. The code is:</p>
  
  <pre><code>import java.io.IOException;
  import java.util.Iterator;
  
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.Text;
  
  import org.apache.hadoop.mapred.*;
  
  import FullFileInputFormat;
  
  public class Source {
  
      public static class Map extends MapReduceBase implements Mapper&lt;Text, Text, Text, Text&gt; {
  
          public void map(Text key, Text value, OutputCollector&lt;Text, Text&gt; output, Reporter reporter) throws java.io.IOException {
              System.out.println("Processing " + key.toString());
              System.out.println("Value: " + value.toString());
          }
      }
  
      public static void main(String[] args) throws Exception {
          JobConf job = new JobConf(Source.class);
          job.setJobName("Source");
  
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(Text.class);
          job.setJarByClass(Source.class);
          job.setInputFormat(FullFileInputFormat.class);
          job.setMapperClass(Map.class);
          FileInputFormat.addInputPath(job, new Path(args[0]));
          FileOutputFormat.setOutputPath(job, new Path(args[1]));
          JobClient.runJob(job);
      }
  }
  </code></pre>