Joins in Batch Processing

How a batch job joins two datasets that live on many machines: shuffle both sides, copy the small side everywhere, or partition both the same way in advance, and what to do about hot keys.

Overview

A join matches records from two datasets on a shared key, for example page views and user profiles on user_id. On one machine a database does this with an index. In a batch job both inputs are large files split across many machines, and there is no index. The records that must meet usually sit on different machines.

So every distributed join answers one question: how do the records with the same key end up in the same place? There are three answers, and they differ in what crosses the network.

Sort-merge join (shuffle both sides) — Both datasets, once — Nothing special; works at any size
Broadcast hash join — Only the small dataset, copied to every worker — The small side fits in each worker's memory
Partitioned join (map-side) — Nothing at join time — Both inputs already split the same way on the join key

Sort-merge join: shuffle both sides

This is the default when both sides are large. It is also called a reduce-side join or repartition join. Each mapper reads a piece of either input and emits the join key with the record, tagged with the side it came from. The framework then partitions the output by key, so every record with the same key goes to the same reducer. The MapReduce paper describes the default partition function as hash(key) mod R, where R is the number of reducers, and guarantees that each reducer sees its keys in sorted order.

The reducer's job is simple because the keys arrive sorted: it takes all records for one key, pairs every profile with every view, and moves to the next key. One detail matters for memory. If the reducer must hold all records for a key before pairing them, a key with millions of views can exhaust memory. The fix is a secondary sort: order records so that the small side (the one profile) arrives first for each key. The reducer then keeps only that one record in memory and streams the views past it. Blanas and others call this the improved repartition join.

The cost is that both datasets cross the network and are sorted. Shipping the data is linear in its size; sorting adds a log factor on each machine.

Broadcast hash join: copy the small side everywhere

Often one side is small: a table of countries, products or ad campaigns. Then there is no need to move the big side at all. Every worker loads a full copy of the small table into an in-memory hash table, reads its own piece of the big table, and probes the hash table for each record. The job needs no reducers.

The limit is memory: every worker must hold the whole small side. Spark broadcasts a table automatically when its estimated size is below spark.sql.autoBroadcastJoinThreshold, which defaults to 10 MB. Setting it to -1 turns automatic broadcasting off. A broadcast also does not fit every join type. In an outer join, the side whose unmatched rows must be kept cannot be the broadcast side: each worker sees only a slice of the other input, so no single worker can tell that a broadcast row matched nothing anywhere. That is why a left outer join broadcasts the right side, a right outer join broadcasts the left, and a full outer join, which keeps both sides, cannot use a broadcast hash join at all.

Partitioned joins: arrange the data in advance

If both inputs were written split the same way, with the same partition function on the join key and the same number of partitions, then partition 3 of one input can only match partition 3 of the other. Each worker joins one pair of partitions with no shuffle. Hadoop calls this a map-side join; Blanas and others call it a directed join; Hive and Spark offer it through bucketed tables.

The saving is real only when the same layout serves many jobs, because someone paid for the shuffle once when the data was written. The layout is also fragile: change the number of partitions or the hash function on one side and the join is wrong or impossible without a reshuffle.

Skew: one hot key slows the whole job

A sort-merge join sends every record for a key to one reducer. If one key is far more common than the rest (a celebrity account, a default value, a null), that reducer gets far more work than the others. The job finishes only when its slowest task does, so one hot key can hold up hundreds of idle machines.

Salting — On the big side, add a random number 0 to N-1 to the hot key. Copy the matching small-side record N times, once per salt value. — N copies of the small-side rows; you must know which keys are hot
Broadcast the hot keys — Join the hot keys with a broadcast join and the rest with a normal shuffle join, then combine — Two joins; needs a list of hot keys
Let the engine split it — Spark's adaptive query execution (AQE) splits a skewed partition into smaller tasks at run time — Only for sort-merge joins, and only past its size thresholds

Spark treats a shuffle partition as skewed when it is more than 5 times the median partition size and larger than 256 MB (the defaults of skewedPartitionFactor and skewedPartitionThresholdInBytes). It then splits that partition and copies the matching partition of the other side, if needed, to each piece.

The twelve views of the hot key now spread across four reducers instead of one.

How an engine picks a strategy

Spark SQL chooses from table statistics, and you can override it with hints. The hints are BROADCAST, MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL; when both sides carry different hints, Spark prefers them in that order. A hint is a request: Spark ignores it when the strategy cannot run that join type.

Adaptive query execution, on by default in current Spark, can also change the plan after a shuffle stage has run: if the real size of one side turns out to be under the broadcast threshold, it switches a planned sort-merge join to a broadcast hash join.

What you will be able to do

Explain why a distributed join must bring equal keys to one machine, and the three ways to do it.
Walk through a sort-merge join in MapReduce, including the secondary sort that keeps reducer memory small.
Say when a broadcast join is safe and what limits it.
Diagnose a join held up by one slow task and fix it with salting or engine skew handling.

Key concepts

Sort-merge join: Also called a reduce-side or repartition join. Both inputs are partitioned by the join key and sorted, then each reducer merges the records for its keys. Works at any size; both sides cross the network.
Broadcast hash join: Every worker receives the whole small input as an in-memory hash table and probes it with its slice of the large input. The large input never moves.
Partitioned (map-side) join: Both inputs were written with the same partition function and number of partitions on the join key, so partition i of one side only meets partition i of the other. No shuffle at join time.
Shuffle: The step that moves every record to the worker responsible for its key, usually by hash of the key modulo the number of reducers. Often the most expensive part of a batch job.
Secondary sort: Sorting within each key so that records from one side arrive before the other. It lets a reducer keep only the small side in memory and stream the large side.
Join skew: A key with far more records than the rest. Its reducer gets most of the work, and the whole stage waits for it.
Salting: Adding a random suffix to a hot key on the large side and copying the matching small-side rows once per suffix, so the hot key spreads over several tasks.

When to use it

Designing a batch pipeline that enriches a large event log with user, product or campaign data
Explaining why a Spark job spends most of its time in the shuffle, and how a broadcast would avoid it
Debugging a join where one task runs far longer than the others
Deciding whether to store tables bucketed on a join key that many jobs share

Interview questions

How does Spark decide which join strategy to use?
Walk me through a sort-merge join in MapReduce. Where does the cost go?
One task in your join stage runs for an hour while the other 199 finish in two minutes. What is going on and how do you fix it?
Why do batch jobs join against a copy of a database table instead of querying the database?
When is it worth storing tables pre-partitioned (bucketed) on a join key?