title: Hadoop: How to get each file path in CombineFileInputFormat?
body: <p>I have many files, some of which are very small. In order to reduce number of mappers, I'd like to use CombineFileInputFormat. 
  The file name will be used as part of the key of the mapper output. </p>
  
  <p>I have tried some methods as follows to get the file name of each chunk in CombineFileSplit, but all failed.</p>
  
  <p>1) I see <code>conf.set("map.input.file", split.getPath(idx).toString());</code> in the function </p>
  
  <p><code>initNextRecordReader()</code> of class <code>CombineFileRecordReader</code>. But <code>NullPointerException</code> </p>
  
  <p>happens in my <code>map()</code> function, as <code>context.getConfiguration().get("map.input.file")</code> </p>
  
  <p>returns <code>null</code>.</p>
  
  <p>2) I also try <code>((FileSplit) (context.getInputSplit())).getPath().getName()</code> in mapper, 
  but <code>java.lang.ClassCastException: org.apache.hadoop.mapreduce.lib.input.CombineFileSplit cannot be cast to org.apache.hadoop.mapreduce.lib.input.FileSplit</code> happens.</p>
  
  <p><strong>So how do I get each file name in the CombineFileSplit ?</strong></p>
  
  <p>============================================================</p>
  
  <p><strong>The input files are lzo compressed, and they are not indexed for now.</strong></p>
  
  <p>The following is my codes:</p>
  
  <p>I implement CombineFileInputFormat like this:</p>
  
  <pre><code>public class CombinedInputFormat extends CombineFileInputFormat&lt;LongWritable, Text&gt; {
  
      @Override
      public RecordReader&lt;LongWritable, Text&gt; createRecordReader(InputSplit arg0,
              TaskAttemptContext arg1) throws IOException {
          // TODO Auto-generated method stub
          return new CombineFileRecordReader&lt;LongWritable, Text&gt;((CombineFileSplit) arg0, arg1, CombineLzoLineRecordReader.class);
      }
  
  }
  </code></pre>
  
  <p>and this is CombineLzoLineRecordReader that extends LzoLineRecordReader:</p>
  
  <pre><code>public class CombineLzoLineRecordReader extends LzoLineRecordReader {
      private int index;
  
      public CombineLzoLineRecordReader(CombineFileSplit split, TaskAttemptContext context, Integer index)
          throws IOException, InterruptedException {
          this.index = index;
      }
  
      public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException, InterruptedException {
          CombineFileSplit combineSplit = (CombineFileSplit) genericSplit;
          FileSplit fileSplit = new FileSplit(combineSplit.getPath(index), combineSplit.getOffset(index), combineSplit.getLength(index), combineSplit.getLocations());
          super.initialize(fileSplit, context);
      }
  }
  </code></pre>
  
  <p>And my map methods is like this :</p>
  
  <pre><code>private String getName(String filePath) {
          String[] filePathDir = filePath.split("/");
          return filePathDir[filePathDir.length - 1];
      }
  
  public void map(LongWritable key, Text value, Context context)
              throws IOException, InterruptedException {
  
          String name = getName(context.getConfiguration().get("map.input.file"));
  
          line = new String(value.getBytes(), 0, value.getLength(), "ISO-8859-1");
          lineFields = line.split("\t",-1);
          if (lineFields != null &amp;&amp; lineFields.length &gt;= 20) { 
                      // do something ...
          }
      }
  </code></pre>
  
  <p>And error infomation :</p>
  
  <pre><code>13/06/14 17:02:50 INFO mapred.JobClient: Task Id : attempt_201209101415_762760_m_000000_0, Status : FAILED
  java.lang.NullPointerException
          at com.netease.devilfish.hadoop.job.LogAnalysisDailyMapper.getName(Unknown Source)
  </code></pre>