MapReduce Joins Visualization | Sort-Merge, Skew, Map-Side

The idea: a join in a batch job

A website logs what its users do: user 104 loaded /x, user 101 paid. The log holds only the user id. To ask "which pages do people of which age visit?" each event needs the viewer's date of birth from the users database. That is a join.

Activity events and users go through mappers that emit user_id as the key, the shuffle sends even ids to reducer 0 and odd ids to reducer 1, and each reducer joins one user with its events — A reduce-side join: the key user_id works like an address, so every record of one user meets in one reduce call.

In a batch job the join is not one lookup. It resolves every association in the data at once: all events, all users. MapReduce has no indexes, so a job reads both inputs in full. That is fine for analytics, which touches most of the data anyway, and the work is spread over many machines.

The page uses the example from Kleppmann's Designing Data-Intensive Applications (chapter 10): six users and a log of 12 events in three HDFS blocks. Each tab joins them a different way, and the counters keep one row per run so you can compare.

Why not query the database for every event?

The simplest join is to scan the events and send a query to the users database for each one. Every query waits for a network round trip, so the job runs at the speed of the network, not the disk. Thousands of tasks doing this at once can overload the database. And the result depends on when the job ran, because the users table changes. (Demo: query per event: 12 events, 12 round trips.)

So a batch job works on a copy of the users database, for example an export from a backup, stored in the same distributed filesystem as the log. Then all reads are local and the job is deterministic: run it twice, get the same output.

Reduce-side sort-merge join

Two kinds of mappers run. One reads the users file and emits user_id → date of birth. The others read the event blocks and emit user_id → event. The framework partitions the map output by key (here user_id mod 2: R0 gets even ids, R1 odd ids), copies each partition to its reducer (the shuffle) and sorts it. Now all records of one user sit next to each other.

The reducer is called once per user id. It reads the date of birth, keeps it in a local variable, then walks the user's events and writes one row {url, dob} for each. It holds one user in memory and makes no network calls. (Demo: sort-merge join: 18 records shuffled, 9 per reducer, 12 rows out.)

One way to see it: mappers "send messages" to reducers, and the key is the address. The framework does the networking and retries failed tasks; the join code only sees records that are already in the right place.

Secondary sort

For the reducer to stream, the user record must come before the user's events. The job sorts by a compound key (user_id, tag, time), with tag 0 for the user record and 1 for an event, but groups only by user_id. So reduce(104) sees dob 1989, t1 /x, t4 /z, t7 /y. This is called a secondary sort. (Demo: secondary sort.)

GROUP BY and sessionization

Grouping uses the same machinery: emit the grouping key, and the shuffle brings each group to one reduce call, which can count, sum or pick the top k. Collecting all events of one user session from many web servers' logs (sessionization) is the same pattern, keyed by session or user id.

Skew and linchpin objects

"All records with the same key go to the same place" breaks down when one key has far more records than the others: a celebrity with millions of followers, a bot, a default value. These are linchpin objects. One reducer gets most of the work, and the job ends only when its slowest reducer ends. (Demo: a hot key: R0 reads 14 records, R1 reads 7.)

Before: reducer 0 gets 14 records because user 104 has 9 events, reducer 1 gets 7. After the skew join: reducer 0 gets 10 and reducer 1 gets 12, with user 104 copied to both — A skew join spreads the hot key's events over all reducers and pays for it by copying the hot user to each of them.

A skew join (Pig and Hive "skewed join", Crunch "sharded join") first samples the input to find the hot keys. Records of a hot key from the big side go to a random reducer instead of the hashed one; the matching record from the other side is sent to every reducer, so wherever an event lands, its user is there. The page sends them round-robin instead of randomly so that undo replays exactly. (Demo: skew join: 10 and 12 records, the same 15 rows.)

For a skewed GROUP BY the fix is two stages: the first job sends records to random reducers and each pre-aggregates its share; a second job combines the partial results per key.

Map-side joins

Sorting, copying to reducers and merging is the expensive part of a reduce-side join, and data may hit disk several times on the way. If you know more about the inputs, a map-side join skips all of it: no reducers, no sort, each mapper reads its input and writes one output file. The catch is the assumption it needs.

Broadcast hash join (Pig "replicated join", Hive "MapJoin"): the small input fits in memory. Each mapper loads all of it into a hash table, then scans its block of the large input and looks each record up. (Demo: broadcast hash join: 0 shuffled, but every mapper read all 6 users.)
Partitioned hash join (Hive "bucketed map join"): both inputs were partitioned the same way, with the same key, the same hash function and the same number of partitions. Mapper p loads only partition p of the small input. (Demo: partitioned hash join: 3 users per mapper. Demo: partitions don't match: users mod 3 vs events mod 2, refused.)
Map-side merge join: the same partitioning, and both sides sorted by the key. A mapper walks both files with two cursors, like a reducer merging its input. Nothing has to fit in memory. (Demo: map-side merge join.)

Three map-side joins side by side: in a broadcast hash join every mapper reads the whole users file; in a partitioned hash join mapper p reads users part p; in a merge join a mapper merges two sorted files with two cursors — What each mapper reads: all of the small input, its own partition of it, or its own sorted partition merged in order.

Join	Assumes	Shuffle	In memory	Output is ordered
Reduce-side sort-merge	nothing	both inputs	one key's first record	by join key
Skew join	a sample of hot keys	both, hot side copied	one key's first record	by join key
Broadcast hash	small side fits in memory	none	all of the small side, per mapper	like the large input
Partitioned hash	same partitioning	none	one partition of the small side	by partition
Map-side merge	same partitioning, sorted	none	one record	by partition, sorted

Choosing a join

The choice changes the shape of the output. A reduce-side join writes files partitioned and sorted by the join key; a map-side join writes one file per block (or partition) of the large input, in its order. A later job may depend on that. To pick a map-side join, the planner must know how each dataset is laid out: how many partitions, by which key, sorted or not. In Hadoop this metadata lives in the Hive metastore and HCatalog. Engines such as Hive, Spark and Impala use it, plus table sizes, to choose a join on their own.

See also MapReduce: word count, shuffle and failures for the job mechanics, join algorithms inside a database (nested loop, hash and sort-merge join on one machine), how HDFS stores the blocks and range vs hash partitioning.

What the page leaves out

Real jobs have thousands of mappers and spill sorted runs to disk and merge them; the page keeps everything in a few records. It does not show combiners, task failures and retries (see the MapReduce page), how the hot keys are sampled, compression, or the read-only on-disk index some broadcast joins use when the small side does not fit in memory. Spark, Flink and Tez run the same join algorithms as one dataflow without writing every intermediate result to HDFS.