Hadoop for Beginners (13): MapReduce Input Splits and Parallelism

Tutorial index: Big Data for Beginners: tutorial series

In Hadoop for Beginners (11): Writing MapReduce in Code — a WordCount Example we got hands-on with MapReduce by writing some code. The MapReduce workflow looks roughly like this:

• InputFormat: turn files into [K,V] values
• Shuffle: merge and organize data sharing the same key
• OutputFormat: write out the [K,V] values

InputFormat Data Input

InputFormat has to turn files into [K,V] values, and those files live as blocks spread across the cluster nodes. How does it stay efficient? By splitting files across the nodes so tasks run in parallel.

How Splits Are Made

An earlier post noted that a block defaults to 128M, meaning files are distributed one 128M block at a time over the cluster. So how should we slice to maximize efficiency?

Split evenly? Given a 300M file, divide it into three even 100M pieces? That looks like balanced load across nodes, but it isn’t:

The first block leaves 28M after taking 100M, and that remainder has to be shipped to the second node. The second node takes its 28M+128M, cuts 100M, and hands the remaining 56M to the next node. That’s a lot of extra network I/O — and network bandwidth is a precious resource. You waste bandwidth and increase processing time.

So the default split size is simply one block. The job plans its input splits on the client side at submission time, which you can see in the source. Here’s the trail I followed:

It starts at job.waitForCompletion(true), reaching org.apache.hadoop.mapreduce.Job#waitForCompletion, where this.submit() leads into org.apache.hadoop.mapreduce.Job#submit, which executes:

this.status = (JobStatus)this.ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
    public JobStatus run() throws IOException, InterruptedException, ClassNotFoundException {
        return submitter.submitJobInternal(Job.this, Job.this.cluster);
    }
});

The key call, submitter.submitJobInternal(Job.this, Job.this.cluster), lands in org.apache.hadoop.mapreduce.JobSubmitter#submitJobInternal, which runs int maps = this.writeSplits(job, submitJobDir) — that’s the splitting code, and maps is how many splits to make and MapTasks to launch. Drill in further:

private int writeSplits(JobContext job, Path jobSubmitDir) throws IOException, InterruptedException, ClassNotFoundException {
    JobConf jConf = (JobConf)job.getConfiguration();
    int maps;
    if (jConf.getUseNewMapper()) {
        maps = this.writeNewSplits(job, jobSubmitDir);
    } else {
        maps = this.writeOldSplits(jConf, jobSubmitDir);
    }
    return maps;
}

That calls maps = this.writeNewSplits(job, jobSubmitDir), i.e. org.apache.hadoop.mapreduce.JobSubmitter#writeNewSplits. Here’s the interesting part — watch the code:

private <T extends InputSplit> int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException, InterruptedException, ClassNotFoundException {
    Configuration conf = job.getConfiguration();
    InputFormat<?, ?> input = (InputFormat)ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
    List<InputSplit> splits = input.getSplits(job);
    T[] array = (InputSplit[])((InputSplit[])splits.toArray(new InputSplit[splits.size()]));
    Arrays.sort(array, new JobSubmitter.SplitComparator());
    JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array);
    return array.length;
}

List<InputSplit> splits = input.getSplits(job) is the crux — it runs org.apache.hadoop.mapreduce.InputFormat#getSplits, meaning InputFormat supplies the splitting logic. That brings us neatly back to the structure I opened with, and it means we should dig a bit into InputFormat. I was starting to think I’d wandered off topic.

org.apache.hadoop.mapreduce.InputFormat is abstract, so we need an implementation. The common one is file-oriented: org.apache.hadoop.mapreduce.lib.input.FileInputFormat. Looking at its getSplits — I’ll skip the full listing and jump to the important line: long splitSize = computeSplitSize(blockSize, minSize, maxSize);. Let’s see what that does:

protected long computeSplitSize(long blockSize, long minSize,
                                long maxSize) {
    return Math.max(minSize, Math.min(maxSize, blockSize));
}

What does that mean? By name we can guess: minimum, maximum, and block size. Taking the smaller of max and block size is obviously the block size, since it can’t exceed the max; then taking the larger of min and that block size is also the block size, since the block size can’t be under the min. So really it just takes the middle value of the three.

All of that was to show you what InputFormat does — and it matters. Since Hadoop’s InputFormat is abstract, what does that imply? Right: we can write our own InputFormat.

Writing a Custom InputFormat

I won’t belabor the built-in ones — read the source directly, or search online.

org.apache.hadoop.mapreduce.InputFormat really has two steps: getSplits handles splitting, and createRecordReader converts each record into the [K,V] values the Mapper receives.

This is a beginner tutorial so I won’t dig much deeper; here’s a small demo, and you can explore the fancier uses yourself.

Source for this post: https://github.com/renfei/demo/tree/master/hadoop/hadoop_api/src/main/java/net/renfei/hadoop/inputformat

Taking the lazy route to get a custom InputFormat working quickly, extend FileInputFormat. First create DemoInputFormat, extending FileInputFormat, and override createRecordReader:

public class DemoInputFormat extends FileInputFormat<Text, BytesWritable> {
    @Override
    public RecordReader createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        return new DemoRecordReader();
    }
}

We also need a RecordReader, so here’s DemoRecordReader:

public class DemoRecordReader extends RecordReader<Text, BytesWritable> {
    private boolean readed = false;
    private Text key = new Text();
    private BytesWritable value = new BytesWritable();
    private FileSplit fileSplit;
    private FSDataInputStream inputStream;
    /**
     * Initialization — called once during initialization
     *
     * @param split
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        // Cast the split type to a file split
        // Safe to cast to FileSplit because net.renfei.hadoop.inputformat.DemoInputFormat
        // extends org.apache.hadoop.mapreduce.lib.input.FileInputFormat
        fileSplit = (FileSplit) split;
        // Get the split path
        Path path = fileSplit.getPath();
        // Get the filesystem from the path
        FileSystem fileSystem = path.getFileSystem(context.getConfiguration());
        // Open a stream
        inputStream = fileSystem.open(path);
    }
    /**
     * Read the next key-value pair
     *
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (readed) {
            return false;
        } else {
            // Read the data
            // Read Key: since this is a demo, we just use the file path as the Text key, which isn't very meaningful
            key.set(fileSplit.getPath().toString());
            // Read Value: reading it all in one go, hence fileSplit.getLength()
            byte[] buf = new byte[(int) fileSplit.getLength()];
            inputStream.read(buf);
            value.set(buf, 0, buf.length);
            readed = true;
            return true;
        }
    }
    /**
     * Get the current key
     *
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return key;
    }
    /**
     * Get the current value
     *
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return value;
    }
    /**
     * How far along the read is
     *
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public float getProgress() throws IOException, InterruptedException {
        return readed ? 1 : 0;
    }
    /**
     * Close resources
     *
     * @throws IOException
     */
    @Override
    public void close() throws IOException {
        IOUtils.closeStream(inputStream);
    }
}

And create a Driver class:

public class DemoDriver {
    /**
     * Program entry point
     *
     * @param args
     */
    public static void main(String[] args) throws IOException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(DemoDriver.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(BytesWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BytesWritable.class);
        // Use our custom DemoInputFormat
        job.setInputFormatClass(DemoInputFormat.class);
        // Since DemoInputFormat extends FileInputFormat, FileInputFormat can set the paths
        FileInputFormat.setInputPaths(job, new Path("/Users/renfei/Downloads/demo.txt"));
        FileOutputFormat.setOutputPath(job, new Path("/Users/renfei/Downloads/demoout"));
    }
}