Hadoop jobs often look simple at the API level: map, shuffle, reduce. The expensive part is usually not the map or reduce function itself, but the movement of intermediate data and the resource decisions around it. Shuffle is the bridge between the map and reduce stages, and it is where sorting, spilling, merging, network transfer, and memory pressure all meet.
Map-side Shuffle
A map task writes output into a memory buffer, commonly a circular buffer. When the buffer reaches its threshold—80% by default—the task spills buffered records to local disk. During the spill, records are sorted by key. If a Combiner is configured, it runs first to perform a local partial aggregation. After the map task finishes, all spill files are merged through a multi-way merge into the final output. The map-side intermediate result is then served to reducers over HTTP.
Reduce-side Shuffle
A reduce task starts by copying the partitions that belong to it from the various map outputs. Once copying is complete, it performs merge and sort, then calls the reduce function for the final aggregation. The output is written to HDFS.
Tuning the map side
<table> <thead> <tr> <th>Lever</th> <th>Setting and effect</th> </tr> </thead> <tbody> <tr> <td>Ring buffer size</td> <td>mapreduce.task.io.sort.mb is 100MB by default; raising it to 200MB gives spills more room.</td>
</tr>
<tr>
<td>Spill threshold</td>
<td>mapreduce.map.sort.spill.percent can move from 0.8 to 0.9.</td>
</tr>
<tr>
<td>Merge frequency</td>
<td>Control the number of spill files; merge 20 files per pass.</td>
</tr>
<tr>
<td>Combiner</td>
<td>Aggregate early on the map side to reduce I/O and network traffic.</td>
</tr>
</tbody>
</table>
Tuning the reduce side
<table> <thead> <tr> <th>Lever</th> <th>Setting and effect</th> </tr> </thead> <tbody> <tr> <td>Map/Reduce counts</td> <td>Too few tasks cause waiting; too many tasks cause resource contention.</td> </tr> <tr> <td>Overlap</td> <td>mapreduce.job.reduce.slowstart.completedmaps allows reducers to start after some maps finish.</td>
</tr>
<tr>
<td>Skip unnecessary reduce</td>
<td>Simple processing jobs can bypass reduce and save network transfer.</td>
</tr>
<tr>
<td>Parallel copy</td>
<td>Increase reducer threads for fetching map data.</td>
</tr>
<tr>
<td>Reduce buffer</td>
<td>mapreduce.reduce.shuffle.input.buffer.percent controls the reduce-side shuffle buffer.</td>
</tr>
</tbody>
</table>
Compression by stage
<table> <thead> <tr> <th>Stage</th> <th>Guidance</th> </tr> </thead> <tbody> <tr> <td>Wire compression</td> <td>Reduce network I/O; Snappy and LZO are recommended.</td> </tr> <tr> <td>Map input</td> <td>Use split-friendly formats such as Bzip2 or LZO with an index.</td> </tr> <tr> <td>Map output</td> <td>Prioritize speed with Snappy or LZO.</td> </tr> <tr> <td>Reduce output</td> <td>Gzip for long-term storage; if the next job will read it, use a split-friendly format.</td> </tr> </tbody> </table>YARN resources and memory settings
YARN is Hadoop's resource management and scheduling framework. It allocates and monitors resources for applications. The key components are:
<table> <thead> <tr> <th>Component</th> <th>Role</th> </tr> </thead> <tbody> <tr> <td>ResourceManager (RM)</td> <td>Cluster-wide resource management and scheduling center.</td> </tr> <tr> <td>NodeManager (NM)</td> <td>Per-node agent that monitors resources and executes tasks.</td> </tr> <tr> <td>ApplicationMaster (AM)</td> <td>Per-application coordinator that manages the map/reduce task lifecycle.</td> </tr> <tr> <td>Container</td> <td>Resource unit of CPU and memory that carries task execution.</td> </tr> </tbody> </table>
The execution flow is straightforward: a client submits a job to the ResourceManager; the RM allocates an ApplicationMaster; the AM requests Containers from the RM; NodeManagers start the tasks; completed tasks report status, and the RM releases resources.
<table> <thead> <tr> <th>Direction</th> <th>Parameter</th> <th>Suggestion</th> </tr> </thead> <tbody> <tr> <td>NodeManager memory</td> <td>yarn.nodemanager.resource.memory-mb</td>
<td>On a 128G machine, configure 100G.</td>
</tr>
<tr>
<td>Single-task memory cap</td>
<td>yarn.scheduler.maximum-allocation-mb</td>
<td>Controls the maximum memory a task can receive.</td>
</tr>
<tr>
<td>Map memory</td>
<td>mapreduce.map.memory.mb</td>
<td>Default 1G; can be raised to 4–5G.</td>
</tr>
<tr>
<td>Reduce memory</td>
<td>mapreduce.reduce.memory.mb</td>
<td>Default 1G; can be raised to 4–5G.</td>
</tr>
<tr>
<td>Map heap</td>
<td>mapreduce.map.java.opts</td>
<td>Helps prevent OutOfMemoryError.</td>
</tr>
<tr>
<td>Reduce heap</td>
<td>mapreduce.reduce.java.opts</td>
<td>Same purpose as the map heap setting.</td>
</tr>
<tr>
<td>CPU parallelism</td>
<td>Container core count</td>
<td>Increase map/reduce parallel capacity.</td>
</tr>
<tr>
<td>Multiple directories</td>
<td>dfs.datanode.data.dir</td>
<td>Spread I/O pressure.</td>
</tr>
<tr>
<td>NameNode thread pool</td>
<td>dfs.namenode.handler.count</td>
<td>Recommended: 20 * log2(Cluster Size).</td>
</tr>
</tbody>
</table>
Scheduler choices
<table> <thead> <tr> <th>Scheduler</th> <th>Characteristics</th> <th>Multiple queues</th> <th>Production use</th> </tr> </thead> <tbody> <tr> <td>FIFO</td> <td>Single queue, first-in-first-out</td> <td>No</td> <td>Not recommended</td> </tr> <tr> <td>Capacity Scheduler</td> <td>Multi-queue, capacity-based allocation</td> <td>Yes</td> <td>Common</td> </tr> <tr> <td>Fair Scheduler</td> <td>Multi-queue, fair sharing</td> <td>Yes</td> <td>Common at large companies</td> </tr> </tbody> </table>Apache's default is the Capacity Scheduler; CDH's default is the Fair Scheduler.
Queues can be organized by framework, such as Hive, Spark, and Flink, or by business module, such as registration, orders, and recommendations. This prevents one job from consuming all resources, protects core business priority, supports resource downgrade scheduling when resources are tight, and improves task isolation.
Benchmark the cluster
After a cluster is built, run benchmarks to verify HDFS read/write performance and MapReduce compute performance. The test JAR is under Hadoop's share directory:
hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-tests.jar TestDFSIO
Failure modes worth watching
<table> <thead> <tr> <th>Scenario</th> <th>Cause</th> <th>Fix</th> </tr> </thead> <tbody> <tr> <td>MR jobs bring down the system</td> <td>Too many concurrent YARN tasks exhaust memory</td> <td>Adjustyarn.scheduler.maximum-allocation-mb to limit per-task memory.</td>
</tr>
<tr>
<td>Fast writes bring down NameNode</td>
<td>Excessive Kafka→HDFS write pressure</td>
<td>Limit Flume batch size (batchsize) or Kafka buffer size.</td>
</tr>
</tbody>
</table>
Data skew
Data skew means some keys carry far more data than others, so a few reducers take much longer. Causes include uneven key distribution, uneven map output key counts, and a poorly designed custom partitioner.
<table> <thead> <tr> <th>Method</th> <th>Principle</th> <th>Use case</th> </tr> </thead> <tbody> <tr> <td>Combiner</td> <td>Local aggregation on the map side reduces the amount of data sent through Shuffle.</td> <td>Aggregation operations such as sum and count.</td> </tr> <tr> <td>Random prefix</td> <td>Add a random prefix to keys so they spread across reducers, then aggregate a second time.</td> <td>Many concentrated keys.</td> </tr> <tr> <td>More reducers</td> <td>Increase parallelism and reduce pressure on individual nodes.</td> <td>Large data volumes.</td> </tr> <tr> <td>Custom partitioner</td> <td>Design a partition function based on key characteristics.</td> <td>Uneven key distribution.</td> </tr> <tr> <td>Two-stage MR aggregation</td> <td>First stage performs local random hashing; second stage performs global aggregation.</td> <td>Very large-scale skew.</td> </tr> </tbody> </table>When tasks pile onto one node
If a 30-machine cluster sends all map tasks to the same machine, YARN is not balancing task instances as intended. The default allows multiple task instances on a single node without spreading them evenly. Turn off that behavior:
yarn.scheduler.fair.assignmultiple = false
With this parameter disabled, tasks are distributed more evenly across nodes.