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().- Creating an class
WholeFileRecordReaderextends theRecordReader.class WholeFileRecordReader extends RecordReader<NullWritable, Text>Note: In this example, we do not need the position and don't have to specify thekey, so usingNullWritableas thekeytype. - 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; } - Creating
WholeFileInputFormatextends theFileInputFormat, which is the class to configure and read the split.public class WholeFileInputFormat extends FileInputFormat<NullWritable, Text> - 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(); } - 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);
