Shuffle and sort

1 The story

The sorting room between two teams

Riya runs the same item-revenue job three times on the 140 MB orders file. Same input, same code, same answer each time. Only one setting changes: the size of the map task's sort buffer. With 16 MB the map task writes its output to disk 18 times and then has to merge those 18 files in two passes. With 200 MB it writes twice. Kabir asks the obvious question: why does a map task write anything to disk at all, if the answer is just six lines?

Arjun Sir

In Lesson 8.1 we said “Hadoop groups by key for you”. Today we open that box. It is called the shuffle and sort, and it is usually the most expensive part of a job: the map output is sorted, written to the worker's local disk, copied over the network to the reducers, and merged again.

Riya

So the map writes to disk because the buffer is too small to hold everything?

Arjun Sir

Partly. Even with a huge buffer the output must land on disk, because the reducers run later, maybe on other machines, and the map task's JVM exits when it is done. The NodeManager keeps serving that file to whichever reducer asks for it.

Kabir

And if the buffer is small, the map writes many little sorted files and has to merge them. That's our 18 spills and the extra 4.35 million records in “Spilled Records”.

Arjun Sir

Exactly. Let's follow one pair, (“chai”, 15), from context.write() to reduce().

2 Definition

Formal definition (write this in exams)

The shuffle and sort is the phase of a MapReduce job, carried out by the framework, that moves intermediate key-value pairs from map tasks to reduce tasks so that each reducer receives all values of its keys, sorted by key. On the map side, output pairs are assigned a partition by the Partitioner (by default HashPartitioner: (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks), collected in an in-memory circular buffer (mapreduce.task.io.sort.mb), sorted by partition and key, optionally passed through a combiner, and spilled to local disk whenever the buffer reaches mapreduce.map.sort.spill.percent (0.80); spill files are merged (mapreduce.task.io.sort.factor streams at a time) into one partitioned, sorted output file per map task. On the reduce side, fetcher threads copy the reducer's partition from every map over HTTP from the NodeManager's ShuffleHandler, merge the segments in memory and on disk, and present each key with an iterator over its values to the reduce() method.

In simple words

Each map sorts its output into labelled piles, one pile per reducer, and leaves them on its worker's disk. Each reducer collects its pile from every map, merges them into one sorted stream, and then calls your reduce once per key.

Words you'll meet in this lesson

PartitionWhich reducer a key goes to; decided by the Partitioner.

Sort bufferCircular in-memory buffer for map output, io.sort.mb (100 MB by default).

SpillWriting the sorted buffer contents to a file on local disk.

Merge factorio.sort.factor: how many sorted files are merged in one pass (10).

ShuffleHandlerThe NodeManager service (mapreduce_shuffle, port 13562) that serves map output.

FetcherA reducer thread that copies map output; 5 in parallel by default.

SlowstartReducers start copying after 5% of maps finish (reduce.slowstart.completedmaps).

Spilled RecordsCounter: how many records were written to local disk, map and reduce side together.

From our lab (Hadoop 3.5.0, ItemRevenue on the 140 MB orders file)

Map output	9,000,000 records; 107,999,586 materialized bytes = 107,999,586 Reduce shuffle bytes (everything the map wrote, the reducer fetched)
io.sort.mb = 16	“soft limit at 13421773” (80% of 16 MB) · 18 spills · “Merging 18 sorted segments” → “Merging 9 intermediate segments out of a total of 18” → “Down to the last merge-pass, with 10 segments left” · Spilled Records 31,354,091 · map 18.6 s
io.sort.mb = 64 / 200	5 spills / 2 spills · Spilled Records 27,000,000 both · map 16.3 s / 16.4 s
3 reducers	HashPartitioner sent chai, dosa, samosa, vada pav to part 1 and idli, poha to part 2; `part-r-00000` stayed empty (0 bytes)
422 MB job (Lesson 8.1)	Shuffled Maps=4, Merged Map outputs=4, 323,998,764 bytes shuffled; the reduce attempt started while maps were still running (slowstart)

Shuffle and sort at three levels

Pick your level. New to this? Start at Beginner and climb.

Beginner The post office sorting room

Every map is a post office collecting letters. It doesn't send them one by one: it drops them into bags, one bag per destination city (a partition per reducer), sorted by street (key).
When the counter table is full, the clerk ties up a bundle and puts it in the back room (a spill). At closing time all bundles for a city are combined into one sorted bag (the merge).
Each city's van (a reducer) visits every post office and picks up only its own bag (the fetch), then merges the bags so the postman walks each street once (reduce).

Intermediate The steps and the settings

Partition: every context.write(k, v) gets a partition number from the Partitioner, immediately.
Buffer: pairs go into the io.sort.mb buffer. At 80% full a background thread sorts it by (partition, key), runs the combiner if any, and spills one file. The map keeps writing into the remaining 20% meanwhile.
Merge: at the end the spills are merged, at most io.sort.factor (10) at a time, into file.out plus an index of where each partition starts. 18 spills needed an extra pass; 5 or 2 did not.
Fetch: after 5% of maps finish, reducers start; 5 fetchers per reducer download their partition from each finished map over HTTP (port 13562).
Merge and reduce: fetched segments are merged (in memory first, spilling to disk when needed); the final merge feeds reduce() directly, one key at a time.

Advanced Reading the counters like an engineer

Spilled Records counts records written to local disk anywhere in the job. Ideal is 2 × map output (map writes once, reducer once): 18,000,000 here, only possible when the map spills exactly once.
With 2–10 spills the map writes each record twice (spill + final merge) and the reducer once: 3 × 9,000,000 = 27,000,000, our 64 MB and 200 MB runs.
With 18 spills the merge needed an intermediate pass that rewrote 9 of the 18 segments: 31,354,091 − 27,000,000 = 4,354,091 extra records, about half the data written a third time.
Materialized bytes (108 MB) exceed raw map output bytes (90 MB for this file) because each record carries key and value lengths in the intermediate file format.
Fixes in order of effect: emit less (filter, combiner, compact types), compress map output (mapreduce.map.output.compress, Module 10), then tune io.sort.mb so a map spills once or twice.

Exam tip: For “Explain shuffle and sort in MapReduce”, draw two boxes: map side (partition → buffer → sort → spill → merge, local disk) and reduce side (fetch over HTTP → merge → reduce, keys in sorted order). Name three settings: io.sort.mb, sort.spill.percent 0.80, io.sort.factor 10.

3 Types and ways to use it

The settings that shape the shuffle

Setting	Default	Side	What it changes
`mapreduce.task.io.sort.mb`	100 MB (our lab: 64)	map	Sort buffer size; bigger means fewer spills. Must fit in the map's heap.
`mapreduce.map.sort.spill.percent`	0.80	map	When a spill starts; “soft limit at 13421773” for 16 MB
`mapreduce.task.io.sort.factor`	10	both	Files merged per pass; 18 spills needed two passes
`mapreduce.map.output.compress`	false	map	Compress spills and shuffle traffic (e.g. Snappy, LZ4)
`mapreduce.reduce.shuffle.parallelcopies`	5	reduce	Fetcher threads per reducer
`mapreduce.job.reduce.slowstart.completedmaps`	0.05	reduce	Fraction of maps finished before reducers launch
`mapreduce.reduce.shuffle.input.buffer.percent`	0.70	reduce	Share of reducer heap used to hold fetched map output
`mapreduce.job.reduces`	1	job	Number of partitions = number of reducers = output files

The hooks you can plug into the shuffle

Partitioner

Decides which reducer gets a key. Default: hash of the key modulo reducers. Custom ones in Lesson 9.2.

which reducer

Sort comparator

Decides the order of keys. Default: the key type's natural order (Text compares bytes).

which order

Combiner

A mini-reduce run on each spill on the map side, to shrink the data before it travels. Lesson 9.1.

less data

Grouping comparator

Decides which consecutive keys share one reduce() call. Used for secondary sort, Lesson 9.7.

which group

The shuffle is all-to-all: with M maps and R reducers there are M × R pieces to copy. 1,000 maps and 200 reducers mean 2 lakh small HTTP fetches. That is why the number of reducers and the size of map output matter so much for big jobs.

4 Step by step

Watching the shuffle on a real cluster

Run on the Module 8 lab with shuffle_lab.sh (in “Run it on your PC”). It runs ItemRevenue (Lesson 8.1) three times on orders_x30.csv (140 MB, one split) with different sort buffers and reads the task logs with yarn logs.

Run the same job with three buffer sizes
```
for mb in 16 64 200; do
  hadoop jar mr.jar ItemRevenue -D mapreduce.task.io.sort.mb=$mb \
      /user/hadoop/data/orders_x30.csv /out/shuf_$mb
done
```
== io.sort.mb=16 job took 39 s == io.sort.mb=64 job took 37 s == io.sort.mb=200 job took 36 s
-D sets a property for this job only (the driver uses GenericOptionsParser). Each map task now gets a different buffer.
The buffer fills and spills
```
yarn logs -applicationId <app> -log_files syslog | grep MapTask
```
MapTask: mapreduce.task.io.sort.mb: 16 MapTask: soft limit at 13421773 MapTask: Spilling map output MapTask: Finished spill 0
13,421,773 bytes is 80% of 16 MB (16,777,216 × 0.8). At that point a SpillThread sorts the buffer by partition and key and writes it to the worker's local disk, while the map keeps filling the rest.

Many spills, merged in passes

MapTask: Finished spill 17
Merger: Merging 18 sorted segments
Merger: Merging 9 intermediate segments out of a total of 18
Merger: Down to the last merge-pass, with 10 segments left of total size: 107999642 bytes

18 spill files (0 to 17), but only 10 may be merged at once. So Hadoop first merges 9 of them into one file, leaving 9 + 1 = 10, then does the final pass into one sorted file.out.

Compare the counters
```
io.sort.mb  spills  Spilled Records  map time
16          18      31,354,091       18.6 s
64           5      27,000,000       16.3 s
200          2      27,000,000       16.4 s
(Map output records = 9,000,000 and Reduce shuffle bytes = 107,999,586 in all three)
```
Map output and shuffle size never change: the data is the same. What changes is how many times it is written to local disk. 27,000,000 = 3 × 9,000,000 (spill, merge, reducer); the 16 MB run wrote 4,354,091 records a third time in the extra pass.
Where each key goes
```
hadoop jar mr.jar ItemRevenue -D mapreduce.job.reduces=3 canteen_orders.csv /out/red_3
```
part-r-00000 (0 bytes): part-r-00001 (79 bytes): chai dosa samosa vada pav part-r-00002 (38 bytes): idli poha
HashPartitioner: (hash("chai") & MAX) % 3 = 1. With only 6 keys, hashing happened to give nothing to reducer 0 and four items to reducer 1. The Python runner below reproduces these numbers.
The reducer sees sorted keys
```
hdfs dfs -cat /out/shuf_16/part-r-00000
```
chai 1505100 22576500 dosa 1498440 67429800 idli 1499520 44985600 poha 1496730 37418250 samosa 1500630 27011340 vada pav 1499580 29991600
The input was in random item order; the output is alphabetical because the reducer reads one merged stream sorted by key and calls reduce() each time the key changes. Keys are sorted within each reducer, not across reducers.

A bigger io.sort.mb is not free: the buffer lives inside the map task's heap. Our maps had -Xmx300m; setting io.sort.mb close to or above that makes the task die with an OutOfMemoryError (values above 2047 MB are rejected outright as invalid). Raise the container memory and the heap together.

5 See it happen

From context.write() to reduce()

Six steps of the shuffle for the 16 MB run: fill, spill, merge, serve, fetch, reduce.

Shuffle and sort 6 steps

What you see

Behind the screen

6 Real-world use

The shuffle at work

1. Tuning slow nightly jobs

Platform teams compare Spilled Records with map output records. A ratio well above 3 means extra merge passes; raising the sort buffer (with container memory) or compressing map output often cuts minutes from big jobs.

Counters→io.sort.mb, compression→Fewer disk writes

2. Diagnosing skew

One reducer runs for hours while the others finish in minutes: one key (a popular product, a null customer ID) got most of the shuffle. Our 3-reducer run showed the small version: 4 items to one reducer, none to another.

Hot key→one huge partition→Salting / custom partitioner

3. Sorting benchmarks

TeraSort, the classic Hadoop benchmark, is almost pure shuffle: map reads, the framework sorts and shuffles terabytes, reduce writes. Clusters are tuned and compared on it.

teragen→terasort (shuffle)→teravalidate

4. The same shuffle in Spark

Spark's groupBy and join also write sorted, partitioned shuffle files to local disk, and on YARN they can be served by an external shuffle service in the NodeManager, just like mapreduce_shuffle.

Stage 1 tasks→shuffle files→Stage 2 tasks

Simplified; organisations differ in tools and policies.

7 Run the code

Simulate the map-side sort buffer, spills and merge

A tiny map task: the buffer holds only 4 pairs, so it spills often. Each spill is sorted by (partition, key). At the end the spills are merged with at most FACTOR files per pass, just like io.sort.factor.

import heapq
BUFFER, FACTOR, REDUCERS = 4, 3, 2
pairs = [("samosa", 18), ("chai", 15), ("poha", 25), ("dosa", 45), ("chai", 15),
         ("idli", 30), ("samosa", 18), ("chai", 15), ("vada pav", 20), ("poha", 25)]

def part(key):                                  # same formula as Hadoop's HashPartitioner on Text
    h = 1
    for b in key.encode():
        h = (31 * h + b) & 0xFFFFFFFF
    return (h & 0x7FFFFFFF) % REDUCERS

spills, buf, written = [], [], 0
for k, v in pairs:
    buf.append((part(k), k, v))
    if len(buf) == BUFFER:                      # buffer full: sort and spill
        spills.append(sorted(buf)); written += len(buf); buf = []
if buf:
    spills.append(sorted(buf)); written += len(buf)
print(len(spills), "spills:", [[k for p, k, v in s] for s in spills])

passes = 0
while len(spills) > 1:                          # merge at most FACTOR files per pass
    group, spills = spills[:FACTOR], spills[FACTOR:]
    merged = list(heapq.merge(*group))
    written += len(merged); passes += 1
    spills.append(merged)
print("merge passes:", passes, "| records written to disk on the map side:", written)
for p, k, v in spills[0]:
    print("partition", p, k, v)

3 spills: [['chai', 'dosa', 'poha', 'samosa'], ['chai', 'chai', 'idli', 'samosa'], ['vada pav', 'poha']] merge passes: 1 | records written to disk on the map side: 20 partition 0 chai 15 partition 0 chai 15 partition 0 chai 15 partition 0 dosa 45 partition 0 vada pav 20 partition 1 idli 30 partition 1 poha 25 partition 1 poha 25 partition 1 samosa 18 partition 1 samosa 18

def part(key): # same formula as Hadoop's HashPartitioner on TextText.hashCode() is WritableComparator.hashBytes: start at 1, multiply by 31, add each (signed) byte. ASCII bytes are positive, so this matches exactly.
if len(buf) == BUFFER: # buffer full: sort and spillHadoop spills at 80% (spill.percent), in a background thread, while the map keeps writing.
merged = list(heapq.merge(*group))Merging sorted files never needs to load them fully: it reads the smallest head of each, like heapq.merge.

Run it on your PC: three sort buffers, one job

Run this inside the namenode container of the Module 8 lab (Lesson 8.1), in /tmp/mr where mr.jar and orders_x30.csv already are. It runs ItemRevenue three times with a different mapreduce.task.io.sort.mb, prints the shuffle counters, counts the spills in the map task log with yarn logs, and shows the merge passes. It takes about 3 minutes.

Shuffle lab

# shuffle_lab.sh: watch the map-side sort buffer spill to disk, and the reducer fetch map outputs
cd /tmp/mr
for mb in 16 64 200; do
  hdfs dfs -rm -r -f -skipTrash /out/shuf_$mb > /dev/null 2>&1
  start=$(date +%s)
  hadoop jar mr.jar ItemRevenue -D mapreduce.task.io.sort.mb=$mb \
      /user/hadoop/data/orders_x30.csv /out/shuf_$mb > /tmp/shuf_$mb.log 2>&1
  app=$(grep -o "application_[0-9_]*" /tmp/shuf_$mb.log | head -1)
  echo "== io.sort.mb=$mb   job took $(( $(date +%s) - start )) s"
  grep -E "Map output records|Map output materialized bytes|Spilled Records|Reduce shuffle bytes|Total time spent by all map tasks" /tmp/shuf_$mb.log | sed 's/^\s*/  /'
  sleep 8
  yarn logs -applicationId $app -log_files syslog 2>/dev/null > /tmp/shuf_$mb.syslog
  echo "  spills written by the map task: $(grep -c 'Finished spill' /tmp/shuf_$mb.syslog)"
done
echo "== map task log (io.sort.mb=16), first lines about the sort buffer"
grep -E "mapreduce.task.io.sort.mb|soft limit|Spilling map output|Finished spill 0" /tmp/shuf_16.syslog | head -4 | cut -c25-150
echo "== map task log (io.sort.mb=16): merging the spill files into one sorted output"
grep -E "Finished spill 17|Merging|merge-pass" /tmp/shuf_16.syslog | cut -c25-170
echo "== reducer input is sorted by key"
hdfs dfs -cat /out/shuf_16/part-r-00000

# In the mr-lab folder:
#   docker cp shuffle_lab.sh lab8-namenode-1:/tmp/mr/
#   docker exec lab8-namenode-1 bash /tmp/mr/shuffle_lab.sh

== io.sort.mb=16 job took 39 s Total time spent by all map tasks (ms)=18644 Map output records=9000000 Map output materialized bytes=107999586 Reduce shuffle bytes=107999586 Spilled Records=31354091 spills written by the map task: 18 == io.sort.mb=64 job took 37 s Total time spent by all map tasks (ms)=16286 Map output records=9000000 Map output materialized bytes=107999586 Reduce shuffle bytes=107999586 Spilled Records=27000000 spills written by the map task: 5 == io.sort.mb=200 job took 36 s Total time spent by all map tasks (ms)=16440 Map output records=9000000 Map output materialized bytes=107999586 Reduce shuffle bytes=107999586 Spilled Records=27000000 spills written by the map task: 2 == map task log (io.sort.mb=16), first lines about the sort buffer INFO [main] org.apache.hadoop.mapred.MapTask: mapreduce.task.io.sort.mb: 16 INFO [main] org.apache.hadoop.mapred.MapTask: soft limit at 13421773 INFO [main] org.apache.hadoop.mapred.MapTask: Spilling map output INFO [SpillThread] org.apache.hadoop.mapred.MapTask: Finished spill 0 == map task log (io.sort.mb=16): merging the spill files into one sorted output INFO [main] org.apache.hadoop.mapred.MapTask: Finished spill 17 INFO [main] org.apache.hadoop.mapred.Merger: Merging 18 sorted segments INFO [main] org.apache.hadoop.mapred.Merger: Merging 9 intermediate segments out of a total of 18 INFO [main] org.apache.hadoop.mapred.Merger: Down to the last merge-pass, with 10 segments left of total size: 107999642 bytes == reducer input is sorted by key chai 1505100 22576500 dosa 1498440 67429800 idli 1499520 44985600 poha 1496730 37418250 samosa 1500630 27011340 vada pav 1499580 29991600

Real output. Log aggregation must be on (yarn.log-aggregation-enable=true in the lab config) for yarn logs to work after the job. Times vary; the spill counts and Spilled Records do not, for the same data and buffer.

8 Line by line

Merging three sorted spill files

The merger never re-sorts: it repeatedly takes the smallest head among the sorted files. Watch the output grow in key order.

Worked examples: real problems, solved and traced in memory

Three shuffle calculations: traffic, buffer sizing and skew.

Example 1BeginnerTraffic · all-to-all

How many pieces does the shuffle copy?

Problem statement

A job has 800 map tasks and 50 reducers. Each map writes one partition per reducer. How many map-output pieces are fetched, and how many fetch threads work at once if every reducer uses the default 5?

Input

800 maps · 50 reducers · 5 fetchers

Expected output

40000 pieces; 250 fetches in parallel; 160 rounds

Approach

Pieces = maps × reducers.
Parallel fetches = reducers × parallelcopies.
Each fetcher handles pieces one after another.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 2IntermediateTuning · buffer

Pick io.sort.mb so a map spills once

Problem statement

Each map task emits 150 MB of map output (2 crore records). The buffer also needs 16 bytes of metadata per record, and a spill starts at 80%. What io.sort.mb avoids a second spill, and what container size does that suggest?

Input

150 MB data · 2 crore records · 16 B metadata · 0.80

Expected output

io.sort.mb >= 569 MB; heap about 769 MB; container about 962 MB

Approach

Buffer need = data + records × 16 B.
Divide by 0.8 for the spill threshold.
The heap must hold the buffer plus the task itself.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 3AdvancedSkew · partitions

When one reducer gets everything

Problem statement

Orders per city are partitioned by city over 4 reducers. Mumbai has 70% of all orders. Using a simple partitioner (city index % 4), how much does each reducer receive out of 10 crore records, and how long does the slowest take at 50 lakh records per minute?

Input

city shares · 4 reducers · 50 lakh rec/min

Expected output

reducer 0: 7.4 crore records, 15 min
reducer 1: 1.2 crore records, 2 min
reducer 2: 0.8 crore records, 2 min
reducer 3: 0.6 crore records, 1 min

Approach

Sum the shares per partition.
Time = records / speed for each reducer.
The job waits for the slowest.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

9 Quiz and revision

Check yourself

Quick revision

Lesson 8.4 in one screen

Map side: partition → circular buffer (io.sort.mb) → sort by (partition, key) → [combiner] → spill at 80% → merge (io.sort.factor 10) → one sorted file per map on local disk.
Reduce side: after slowstart (5%), fetchers (5) copy the reducer's partition from every map via the NodeManager ShuffleHandler (13562) → merge → reduce() per key, keys sorted.
Lab: io.sort.mb 16/64/200 → 18/5/2 spills; Spilled Records 31,354,091 / 27,000,000 / 27,000,000; shuffle bytes 107,999,586 every time.
Spilled Records ideal = 2 × map output; 3× with several spills; more with extra merge passes.
HashPartitioner: (hash & MAX_VALUE) % R. 3 reducers: 4 items in part 1, 2 in part 2, part 0 empty.
Keys are sorted within each reducer, not globally. Number of reducers = number of output files.
Reduce shuffle cost: emit less, combiner, compress map output; then tune the buffer.

Interview and viva questions

What is shuffle and sort in MapReduce?

It is the framework phase between map and reduce. Map output pairs are partitioned by key, sorted and written to the map node's local disk; each reducer then copies its partition from every map over HTTP, merges the copies into one sorted stream and calls reduce once per key with all its values. The programmer does not write it, but can customise it with a partitioner, sort and grouping comparators and a combiner.

What is a spill in Hadoop MapReduce?

Map output is first collected in an in-memory circular buffer of mapreduce.task.io.sort.mb. When it reaches mapreduce.map.sort.spill.percent (80% by default), a background thread sorts the contents by partition and key, runs the combiner if one is set, and writes them to a spill file on local disk. At the end of the map, all spill files are merged into one output file.

Where is the intermediate output of a map task stored?

On the local disk of the node that ran the map task (in the NodeManager's local directories), not in HDFS. It is temporary and served to reducers by the NodeManager's shuffle service. If that node fails before the reducers have fetched it, the map task is re-executed elsewhere.

What is the role of the partitioner?

It decides which reducer receives each intermediate key, so that all values of a key reach the same reducer. The default HashPartitioner computes (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks. A custom partitioner can balance load or route ranges of keys to reducers (for example for a total sort).

Are keys sorted in the final output of a MapReduce job?

Within each reducer's output file, yes: the reducer processes keys in sorted order. Across output files there is no global order with the default hash partitioner, because keys are spread by hash. A global order needs one reducer or a range partitioner such as TotalOrderPartitioner.

Practice problems

Shuffle exercises using the real counters and log lines from the lab.

P1Easy

Soft limit

Compute the spill threshold in bytes for io.sort.mb = 16, 64 and 100 with spill.percent 0.80.

Input

MB × 1,048,576 × 0.8

Expected output

16 13421772
64 53687091
100 83886080

Show solution and explanation

for mb in (16, 64, 100):
    print(mb, int(mb * 1024 * 1024 * 0.8))

16 13421772 64 53687091 100 83886080

How it works: 13,421,772 for 16 MB; the log printed 13421773 because Hadoop rounds the float.

P2Easy

Spills to merge passes

With io.sort.factor 10, how many merge passes do 2, 5, 10, 18 and 40 spills need (roughly: keep merging up to 10 at a time until one file is left)?

Input

spills, factor 10

Expected output

2 spills -> 1 pass(es)
5 spills -> 1 pass(es)
10 spills -> 1 pass(es)
18 spills -> 2 pass(es)
40 spills -> 5 pass(es)

Show solution and explanation

for spills in (2, 5, 10, 18, 40):
    n, passes = spills, 0
    while n > 1:
        n = n - min(10, n) + 1
        passes += 1
    print(spills, "spills ->", passes, "pass(es)")

2 spills -> 1 pass(es) 5 spills -> 1 pass(es) 10 spills -> 1 pass(es) 18 spills -> 2 pass(es) 40 spills -> 5 pass(es)

How it works: 1 pass up to 10 spills, 2 for 18 (our log: 9 merged first, then the last 10). Hadoop chooses the first pass size cleverly so the final pass has exactly 10.

P3Easy

Pieces to fetch

Our 422 MB job had 4 maps and 1 reducer: “Shuffled Maps =4”. How many pieces would 4 maps and 3 reducers create?

Input

maps × reducers

Expected output

4 and 12

Show solution and explanation

print(4 * 1, "and", 4 * 3)

4 and 12

How it works: 4 and 12: every reducer fetches one piece from every map.

P4Medium

Partition a new item

The canteen adds “pav bhaji” and “masala chai”. Which of 3 reducers will each go to?

Input

HashPartitioner on Text

Expected output

pav bhaji 2
masala chai 0

Show solution and explanation

def p(s, r):
    h = 1
    for b in s.encode():
        h = (31 * h + b) & 0xFFFFFFFF
    return (h & 0x7FFFFFFF) % r
for item in ["pav bhaji", "masala chai"]:
    print(item, p(item, 3))

pav bhaji 2 masala chai 0

How it works: The partition depends only on the bytes of the key and the number of reducers, so you can predict it before running the job.

P5Medium

Materialized bytes

Map output bytes were 89,999,580 and materialized bytes 107,999,586 for 9,000,000 records. How many extra bytes per record does the intermediate format add?

Input

two counters

Expected output

2.0000006666666668 bytes per record

Show solution and explanation

print((107_999_586 - 89_999_580) / 9_000_000, "bytes per record")

2.0000006666666668 bytes per record

How it works: About 2 bytes: the key length and value length, each stored as a 1-byte variable-length integer for these small records (plus a few bytes of checksum per file).

P6Medium

Merge sorted runs

Merge three sorted lists of keys without sorting the combined list.

Input

3 sorted runs

Expected output

['chai', 'chai', 'dosa', 'dosa', 'idli', 'poha', 'samosa', 'vada pav']

Show solution and explanation

import heapq
runs = [["chai", "dosa", "poha"], ["chai", "idli"], ["dosa", "samosa", "vada pav"]]
print(list(heapq.merge(*runs)))

['chai', 'chai', 'dosa', 'dosa', 'idli', 'poha', 'samosa', 'vada pav']

How it works: This is exactly what the map-side merger and the reducer's merger do with spill files and fetched segments.

P7Business problem

Is a bigger buffer worth it?

A nightly job runs 2,000 maps. Raising io.sort.mb from 100 to 300 MB saves 3 s per map (fewer spills) but each map container grows by 256 MB. The cluster has spare memory only for 1,800 maps at once at the bigger size (vs 2,000 now). Which is faster overall if each map takes 60 s?

Input

2,000 maps · 60 s vs 57 s · 2,000 vs 1,800 slots

Expected output

now 60 s, bigger buffer 114 s

Show solution and explanation

import math
now = math.ceil(2000 / 2000) * 60
big = math.ceil(2000 / 1800) * 57
print(f"now {now} s, bigger buffer {big} s")

now 60 s, bigger buffer 114 s

How it works: 60 s vs 114 s: the saving per task is lost because the maps no longer fit in one wave. Tuning must consider the whole cluster.

P8Advanced

Simulate slowstart

10 maps finish at seconds 20, 22, 25, 30, 31, 33, 40, 41, 45, 60. Reducers start when slowstart × maps have finished. When do they start for 0.05, 0.5 and 1.0, and how many seconds of copying can overlap the maps?

Input

finish times, slowstart

Expected output

slowstart 0.05: reducers start at 20 s, can copy for 40 s while maps run
slowstart 0.5: reducers start at 31 s, can copy for 29 s while maps run
slowstart 1.0: reducers start at 60 s, can copy for 0 s while maps run

Show solution and explanation

import math
done = [20, 22, 25, 30, 31, 33, 40, 41, 45, 60]
for s in (0.05, 0.5, 1.0):
    need = max(1, math.ceil(s * len(done)))
    start = sorted(done)[need - 1]
    print(f"slowstart {s}: reducers start at {start} s, can copy for {60 - start} s while maps run")

slowstart 0.05: reducers start at 20 s, can copy for 40 s while maps run slowstart 0.5: reducers start at 31 s, can copy for 29 s while maps run slowstart 1.0: reducers start at 60 s, can copy for 0 s while maps run

How it works: Early start overlaps copying with slow maps, but the reducer holds its container while waiting; busy clusters often raise slowstart to 0.8 or so.

Real-world scenario questions (asked in company interviews)

The shuffle is where performance and skew questions come from, at every level of experience.

Concepts · basicsFresher

“What happens between the mapper and the reducer?”

A standard follow-up to word count.

The interviewer is checking: partition, sort, spill, fetch, merge.

Show a strong answer

Each pair gets a partition (HashPartitioner), goes into the sort buffer, is sorted by partition and key and spilled to local disk; spills are merged into one file per map.
Reducers fetch their partition from every map over HTTP (NodeManager ShuffleHandler), merge them and call reduce once per key, in sorted order.
A combiner can shrink data on the map side before it travels.

Performance · countersExperienced

Spilled Records is 5 times Map output records

A long-running aggregation job; the team asks what to tune.

The interviewer is checking: reading counters, buffer and merge settings.

Show a strong answer

Ideal is 2×; 3× means several spills; 5× means extra merge passes (more spills than io.sort.factor) on map or reduce side.
Reduce map output first: combiner, filter early, compact keys, compress map output (Snappy/LZ4).
Then raise io.sort.mb (and the map heap/container) and io.sort.factor; check the task logs for “Merging N intermediate segments”.

Skew · partitioningExperienced

99 reducers finish in 3 minutes, one runs for 2 hours

A join on customer_id; many records have customer_id = NULL.

The interviewer is checking: hot keys and how the partitioner routes them.

Show a strong answer

All values of one key go to one reducer: the NULL key is one giant partition. Check the reducer's “Reduce input records” and shuffle bytes.
Filter or handle NULLs separately; salt hot keys (key + random suffix) and aggregate in two steps; or a custom partitioner.
More reducers do not help: the hot key still lands on one of them.

Failures · shuffleExperienced

Reducers fail with “Too many fetch failures” after a node dies

The job eventually succeeds but takes twice as long.

The interviewer is checking: where map output lives.

Show a strong answer

Map output is on the dead node's local disk, not HDFS, so reducers can't fetch it.
After repeated fetch failures the MRAppMaster marks those maps failed and reruns them elsewhere; reducers then fetch the new output.
The delay is the rerun of finished maps; healthy disks, NodeManager monitoring and blacklisting bad nodes reduce it.

Shuffle and sort

The sorting room between two teams

Shuffle and sort

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, ItemRevenue on the 140 MB orders file)

Shuffle and sort at three levels

Beginner The post office sorting room

Intermediate The steps and the settings

Advanced Reading the counters like an engineer

The settings that shape the shuffle

The hooks you can plug into the shuffle

Partitioner

Sort comparator

Combiner

Grouping comparator

Watching the shuffle on a real cluster

Run the same job with three buffer sizes

The buffer fills and spills

Many spills, merged in passes

Compare the counters

Where each key goes

The reducer sees sorted keys

From context.write() to reduce()

The shuffle at work

1. Tuning slow nightly jobs

2. Diagnosing skew

3. Sorting benchmarks

4. The same shuffle in Spark

Simulate the map-side sort buffer, spills and merge

Run it on your PC: three sort buffers, one job

Shuffle lab

Merging three sorted spill files

Worked examples: real problems, solved and traced in memory

How many pieces does the shuffle copy?

Pick io.sort.mb so a map spills once

When one reducer gets everything

Check yourself

Quick revision

Lesson 8.4 in one screen

Interview and viva questions

Practice problems

Soft limit

Spills to merge passes

Pieces to fetch

Partition a new item

Materialized bytes

Merge sorted runs

Is a bigger buffer worth it?

Simulate slowstart

Real-world scenario questions (asked in company interviews)

“What happens between the mapper and the reducer?”

Spilled Records is 5 times Map output records

99 reducers finish in 3 minutes, one runs for 2 hours

Reducers fail with “Too many fetch failures” after a node dies