The Reduce Phase

Analogy categoryAnalogies

Overview of the Reduce Phase

The reduce phase is the final stage of a MapReduce job and is itself commonly described as three sub-phases: copy (fetching partitioned map output), sort (merging fetched segments into one grouped stream), and reduce (invoking the user's reduce() method once per distinct key). Only once all the values for a key have been merged and grouped does the framework call reduce(key, values, context) — this guarantee is what allows a reducer to compute things like a running total, a maximum, or a distinct count correctly, since it is certain it has seen every value for that key before finishing.

🏏

Cricket analogy: A tournament's official statistician doesn't publish a batsman's final tournament average until every single match's scoresheet has been collected and combined, ensuring the number reflects all innings, just as reduce() only runs once all values for a key are gathered.

The Reducer Contract

A Reducer's reduce() method receives a key and an Iterable<VALUEIN> representing every value emitted for that key across all mappers, and it can emit zero, one, or many output key-value pairs via context.write(). A crucial and often-missed detail is that the values Iterable is single-pass and backed by data that gets overwritten as iteration proceeds — Hadoop reuses the same Writable object for efficiency, so if you need to retain values beyond a single pass (say, to compute both a min and a max in ways that need the raw list), you must explicitly copy them rather than just storing the Writable reference, or every stored 'copy' will end up holding the last value seen.

🏏

Cricket analogy: A commentator calling out a bowler's figures live only gets to watch each ball's replay once as it streams by; if they want to reference an earlier delivery later, they must write it down themselves rather than expect the broadcast feed to replay it, mirroring the single-pass values iterator.

Number of Reducers and Its Impact

The number of reduce tasks is set explicitly via job.setNumReduceTasks() (or mapreduce.job.reduces) rather than derived automatically from data size, and this choice has real consequences: too few reducers under-parallelizes the reduce phase and can create huge output files, while too many reducers creates many small output files and adds scheduling and merge overhead disproportionate to the work each one does. Setting the reducer count to zero produces a map-only job, useful for pure filtering or transformation tasks where no cross-record aggregation by key is needed, and skips the shuffle-and-sort phase entirely since there's nothing to group.

🏏

Cricket analogy: A tournament organizer deciding how many finals venues to book has to balance too few venues causing overcrowding against too many venues splitting attendance too thinly, mirroring the tradeoff in choosing reducer count.

java

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    private final IntWritable result = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get(); // values Iterable is single-pass; read once
        }
        result.set(sum);
        context.write(key, result);
    }
}

A reasonable starting heuristic for choosing reducer count is roughly 0.95 to 1.75 times (number of NodeManagers × mapreduce.tasktracker.reduce.tasks.maximum), which lets the fastest reducers pick up a second wave of work while slower nodes finish their first, keeping the cluster busy without over-fragmenting output files.

Output Committing and Fault Tolerance

Each reduce task writes its output not directly to the final output directory but to a task-attempt-specific temporary directory (something like _temporary/1/_temporary/attempt_.../), and only after the task completes successfully does the OutputCommitter promote that directory's contents to the job's final output location — this is what makes speculative execution and task retries safe, since two attempts at the same task can write to separate temp locations without colliding, and only one attempt's output is ever actually committed. The FileOutputCommitter protocol coordinates this two-phase commit (task commit, then job commit) so that even if the ApplicationMaster itself fails partway through, no partial or duplicate output is visible in the final directory.

🏏

Cricket analogy: A DRS review keeps the on-field umpire's original call unofficial until the third umpire's review is finalized and only then updates the official scorecard, similar to how a task's output stays in a temporary location until it's officially committed.

Never write output files directly from within map() or reduce() using raw file I/O outside the OutputCommitter's managed path (for example, opening a FileOutputStream to a fixed path). Because speculative execution can run two attempts of the same task simultaneously, and failed attempts can be retried, direct side-effect writes bypass Hadoop's commit protocol and can produce duplicated, corrupted, or partially written files.

The reduce phase consists of copy, sort, and reduce sub-phases, with reduce() called once per fully grouped key.
reduce() receives a single-pass Iterable of values, and Hadoop reuses the underlying Writable object across iterations.
The number of reducers is set explicitly and significantly affects parallelism and output file granularity.
Setting reducer count to zero produces a map-only job and skips the shuffle-and-sort phase entirely.
Task output is written to a temporary attempt directory and only promoted to the final path after successful commit.
The two-phase task/job commit protocol makes speculative execution and task retries safe from data corruption.
Bypassing the OutputCommitter with raw side-effect file writes can produce duplicated or corrupted output.