Customize the FileInputFormat to Read Whole File

In MapReduce, the Map task reads input text file line by line, as following pattern

key_1   ..... line_1 in content......
key_2   ..... line_2 in content......
.
.
.
key_n   ..... line_n in content......
.
.
.

This is because the default TextInputFormat uses the LineRecordReader to read the input line by line. InLineRecordReader

  public boolean nextKeyValue() throws IOException {
    if (key == null) {
      key = new LongWritable();
    }
    // Using the position of line to be the key.
    key.set(pos);
    if (value == null) {
      value = new Text();
    }
    int newSize = 0;
    while (pos < end) {
      // Reading a line of input file.
      newSize = in.readLine(value, maxLineLength,
                            Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),
                                     maxLineLength));
      if (newSize == 0) {
        break;
      }
      pos += newSize;
      if (newSize < maxLineLength) {
        break;
      }

      // line too long. try again
      LOG.info("Skipped line of size " + newSize + " at pos " +
               (pos - newSize));
    }
    ...
 }

So, to read the whole file to Map task, we have to customize the methon nextKeyValue().

1. Creating an class WholeFileRecordReader extends the RecordReader.

   class WholeFileRecordReader extends RecordReader<NullWritable, Text>
   

   Note: In this example, we do not need the position and don't have to specify the key, so using NullWritable as the key type.
2. Implementing the method nextKeyValue() as following.

    public boolean nextKeyValue() throws IOException {
        if (!processed) {
            byte[] contents = new byte[(int) fileSplit.getLength()];
   
            Path file = fileSplit.getPath();
            FileSystem fs = file.getFileSystem(conf);
   
            FSDataInputStream in = null;
            try {
                in = fs.open(file);
                IOUtils.readFully(in, contents, 0, contents.length);
                value.set(contents, 0, contents.length);
            } finally {
                IOUtils.closeStream(in);
            }
            processed = true;
            return true;
        }
        return false;
    }
   

3. Creating WholeFileInputFormat extends the FileInputFormat, which is the class to configure and read the split.

   public class WholeFileInputFormat extends FileInputFormat<NullWritable, Text>
   

4. Implementing two method as following.

    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false;
    }
   
    @Override
    public RecordReader<NullWritable, Text> createRecordReader(
      InputSplit split, TaskAttemptContext context) {
        return new WholeFileRecordReader();
    }
   

5. Specifying the input format class as WholeFileInputFormat.

    Job job = Job.getInstance(conf, "Mapper Practice Three");
    job.setJarByClass(PracticeThree.class);
    job.setMapperClass(PracticeThreeMapper.class);
    job.setOutputFormatClass(NullOutputFormat.class);
   
    job.setNumReduceTasks(0);
    job.setInputFormatClass(WholeFileInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);