Broadcast joins (distributed)

Concepts covered: Join Cardinality

A single machine joins 2 tables by reading both from its own disk. A cluster holds each table in pieces spread across many workers, and a row from the events table can only be paired with a row from the regions table when both sit in the same worker's memory. Every distributed join starts with the same question: which rows have to move so that matching keys meet on one machine?

Two ways to bring keys together

A partitioned join hashes the join key on both sides and sends each row to the worker that owns that hash value. Both tables cross the network, so a 2TB events table costs 2TB of network traffic before a single pair is compared. A broadcast join leaves the large table where it already sits and copies the entire small table to every worker. Each worker then joins its own slice of events against a full local copy of regions.

Partitioned join

• Both sides hashed on the join key and shuffled
• Network cost grows with the size of both tables
• Each worker holds one slice of each side
• Works when both tables are large
• A hot key sends all of its rows to one worker

Broadcast join

• The large side never moves
• The small side is copied once per worker
• Each worker holds the entire small side
• Needs one side that fits in memory
• The large side keeps its even spread

Put numbers on it. The events table is 2TB, the regions table is 50MB, and the cluster has 100 workers. A partitioned join shuffles both tables, so the network carries all of events plus all of regions. A broadcast join sends the 50MB regions table to each of the 100 workers and moves nothing else.

• 2.05TB
• 5GB

The query does not change

The SQL is identical under both strategies. The planner chooses from table statistics. It broadcasts whichever side its statistics show is small enough to fit in each worker's memory, as long as the join type allows it: in a LEFT JOIN only the right side, the one not preserved, can be broadcast. Otherwise it partitions. Most engines expose a session setting or a hint to force one choice when statistics are stale. Run the join below. With 8 event rows and 3 region rows, regions is the side to broadcast, and the result is the same under either plan.

SELECT
  r.region_name,
  COUNT(*) AS plays,
  SUM(e.minutes) AS minutes
FROM viewing_events e
JOIN regions r
  ON e.region_code = r.region_code
GROUP BY r.region_name
ORDER BY minutes DESC

Three groups come back: Brazil with 108 minutes from 2 plays, the United States with 100 minutes from 4 plays, and Japan with 80 minutes from 2 plays. On a cluster each worker computes partial counts and sums for its own slice of events, and a final step adds the partials per region.

Where broadcast breaks

The small side has to be small on every worker at once. A 40GB accounts table broadcast to 100 workers puts 40GB into each worker's memory, 4TB of copies in total, and a worker with 8GB of memory fails the query. The second trap is a table that was small when the query was written: a lookup table that grew from 10,000 rows to 50 million turns a fast broadcast into an out of memory error months later. Statistics kept current are what let the planner notice the change.

A query joins a 2TB events table to a 40GB accounts table on account_id. The cluster has 100 workers with 8GB of memory each. Which plan runs?

• Broadcast accounts, because it is the smaller side and the large side never has to move
• Partition both sides on account_id, because neither table fits in a single worker's memory
• Broadcast events, because every worker already holds a copy of the accounts table
• Partition events only and leave accounts in place, because accounts is the smaller side

About this lesson

This section is part of the Joins: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.