title: Getting Filename/FileData as key/value input for Map when running a Hadoop MapReduce Job
body: <p>I went through the question <a href="https://stackoverflow.com/questions/5716138/how-to-get-filename-file-contents-as-key-value-input-for-map-when-running-a-hado">How to get Filename/File Contents as key/value input for MAP when running a Hadoop MapReduce Job?</a> here. Though it explains the concept, I am unable to successfully transform it to code.</p>
  
  <p>Basically, I want the file name as key and the file data as value. For that I wrote a custom <code>RecordReader</code> as recommended in the aforementioned question. But I couldn't understand how to get the file name as the key in this class. Also, while writing the custom <code>FileInputFormat</code> class, I couldn't understand how to return the custom <code>RecordReader</code> I wrote previously.</p>
  
  <p>The <code>RecordReader</code> code is:</p>
  
  <pre><code>import java.io.IOException;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.InputSplit;
  import org.apache.hadoop.mapreduce.RecordReader;
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  
  public class CustomRecordReader extends RecordReader&lt;Text, Text&gt; {
  
      private static final String LINE_SEPARATOR = System.getProperty("line.separator");
  
      private StringBuffer valueBuffer = new StringBuffer("");
      private Text key = new Text();
      private Text value = new Text();
      private RecordReader&lt;Text, Text&gt; recordReader;
  
      public SPDRecordReader(RecordReader&lt;Text, Text&gt; recordReader) {
          this.recordReader = recordReader;
      }
  
      @Override
      public void close() throws IOException {
          recordReader.close();
      }
  
      @Override
      public Text getCurrentKey() throws IOException, InterruptedException {
          return key;
      }
  
      @Override
      public Text getCurrentValue() throws IOException, InterruptedException {
          return value;
      }
  
      @Override
      public float getProgress() throws IOException, InterruptedException {
          return recordReader.getProgress();
      }
  
      @Override
      public void initialize(InputSplit arg0, TaskAttemptContext arg1)
              throws IOException, InterruptedException {
          recordReader.initialize(arg0, arg1);
      }
  
      @Override
      public boolean nextKeyValue() throws IOException, InterruptedException {
  
          if (valueBuffer.equals("")) {
              while (recordReader.nextKeyValue()) {
                  valueBuffer.append(recordReader.getCurrentValue());
                  valueBuffer.append(LINE_SEPARATOR);
              }
              value.set(valueBuffer.toString());
              return true;
          }
          return false;
      }
  
  }
  </code></pre>
  
  <p>And the incomplete <code>FileInputFormat</code> class is:</p>
  
  <pre><code>import java.io.IOException;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.InputSplit;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.RecordReader;
  import org.apache.hadoop.mapred.Reporter;
  
  public class CustomFileInputFormat extends FileInputFormat&lt;Text, Text&gt; {
  
      @Override
      protected boolean isSplitable(FileSystem fs, Path filename) {
          return false;
      }
  
      @Override
      public RecordReader&lt;Text, Text&gt; getRecordReader(InputSplit arg0, JobConf arg1,
              Reporter arg2) throws IOException {
          return null;
      }
  }
  </code></pre>