The Map Phase

Analogy categoryAnalogies

Overview of the Map Phase

The map phase is the first stage of a MapReduce job, where the framework launches one map task per InputSplit, and each task's RecordReader feeds a stream of key-value pairs into the user-defined map() method one record at a time. A single MapReduce job typically runs hundreds or thousands of map tasks in parallel across the cluster, and because each task only ever sees its own split of the data, there is no coordination needed between mappers during this phase, which is what allows map tasks to scale near-linearly with cluster size.

🏏

Cricket analogy: In a domestic tournament like the Ranji Trophy, dozens of matches are played simultaneously across different grounds, and each umpire only scores the match in front of them without needing updates from any other ground, just like each map task only processes its own split.

InputSplits and RecordReaders

An InputSplit is a logical chunk of the input, typically sized close to the HDFS block size (128MB by default), and the InputFormat's getSplits() method computes these splits by consulting the file's block locations so the JobTracker or ApplicationMaster can schedule each map task near its data. It's important to note that a split is logical, not physical — a TextInputFormat split may straddle a block boundary, in which case the RecordReader transparently reads a few extra bytes from the neighboring block to complete the last line, ensuring no record is truncated or duplicated across splits.

🏏

Cricket analogy: An innings is logically divided into overs of six balls each, and if a ball's outcome (like a boundary that needs a replay check) spans into the next over's review, the third umpire still resolves it correctly without losing the delivery, just as a RecordReader completes a record spanning a block boundary.

Mapper Lifecycle and Local Aggregation

A Mapper instance goes through a setup() call once before any records are processed, then map() is invoked once per input record, and finally cleanup() runs once after all records in that task are done — this lifecycle lets you open a database connection or load a lookup file in setup() rather than reopening it on every record. As map() emits intermediate pairs, they are written to an in-memory circular buffer (sized by mapreduce.task.io.sort.mb, default 100MB); when the buffer crosses a threshold (mapreduce.map.sort.spill.percent, default 0.80), a background thread sorts and optionally combines the buffered pairs and spills them to a local disk file, and multiple spill files are later merged into one sorted output per map task.

🏏

Cricket analogy: A bowler's spell has a warm-up (setup, like a couple of loosening deliveries), then each ball bowled is the repeated action (map), followed by walking back to mark their run-up done for the day (cleanup), rather than re-warming up before every single delivery.

java

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private static final IntWritable ONE = new IntWritable(1);
    private final Text word = new Text();

    @Override
    protected void setup(Context context) {
        // e.g. load a stop-word list once per task
    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        for (String token : value.toString().split("\\s+")) {
            if (!token.isEmpty()) {
                word.set(token.toLowerCase());
                context.write(word, ONE);
            }
        }
    }
}

The ApplicationMaster (or JobTracker in MapReduce 1) tries to schedule each map task on a NodeManager that already holds a replica of the InputSplit's HDFS block. When that's not possible, it falls back to a node in the same rack, and only as a last resort schedules off-rack, since data locality is one of the biggest levers for map phase performance.

Partitioning Output for the Shuffle

Before intermediate output leaves a map task, a Partitioner decides which reducer each key-value pair should go to, and the default HashPartitioner computes (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks, which spreads keys roughly evenly across reducers assuming a reasonable hash distribution. Custom partitioners are useful when you need all records for a related set of keys to land on the same reducer — for example, routing all transactions for a given customer ID to one reducer regardless of the exact key, which the default hash-based scheme wouldn't guarantee without a custom key design.

🏏

Cricket analogy: A tournament's fixture scheduler assigns each match to one of several stadiums using a rotation so no single venue is overloaded, similar to how HashPartitioner spreads keys evenly across reducers by default.

Configuring too many tiny InputSplits — for example, thousands of small files each under a few MB — creates one map task per split, and the JVM startup and scheduling overhead per task can dwarf the actual processing time. Using CombineFileInputFormat or pre-merging small files into larger ones before ingestion avoids this 'small files problem' and keeps map task overhead proportional to real work.

One map task is launched per InputSplit, and mappers run independently with no coordination between them.
InputSplits are logical chunks sized near the HDFS block size; RecordReaders stitch records that cross block boundaries.
A Mapper's lifecycle is setup() once, map() per record, cleanup() once, enabling efficient reuse of expensive resources.
Map output is buffered in memory, sorted, optionally combined, and spilled to disk once a fill threshold is reached.
Data locality scheduling places map tasks on or near the node holding the relevant HDFS block whenever possible.
A Partitioner determines which reducer each intermediate key-value pair is routed to, defaulting to a hash-based scheme.
Too many small InputSplits create excessive per-task overhead; combining small files avoids the 'small files problem'.