BeginnerSpark · 13 min · 15 challenges
Joins: Beginner
Almost every pipeline you will ever ship joins something: facts to dimensions, events to users, orders to products. And in most slow Spark jobs, the line that costs the money is a join. The reason is physical, not logical: a join has to compare rows that live on different machines, and comparing rows means first getting them to the same place. Spark has exactly two ways to arrange that meeting. It can move both tables, shuffling every row so matching keys land together, or it can copy one small table to every executor and move nothing else. Which of the two happens decides whether your join is a footnote in the runtime or the whole bill. This lesson builds that two-strategy model: why the meeting is required, when the cheap copy applies, and the one way a join can quietly multiply your data on the way through.
Why a Join Moves Data
You can now explain why a join forces both tables through a shuffle to co-locate matching keys.
Start with what a join physically has to do. To match an order to its product, Spark must compare two rows, one from orders and one from products, sharing a product_id. But those rows live wherever the data happened to be read: the order sits in a partition on one machine, the product in a partition on another. No amount of cleverness lets two machines compare rows without at least one of them moving. A join is therefore a co-location problem before it is a matching problem: every pair of rows that might match has to be brought onto the same machine, at the same moment, before any matching logic can run.
The default way Spark arranges that meeting is the shuffle you already know from the narrow-and-wide family, applied to both tables at once. Every row of orders is hashed on product_id and sent to the partition that owns that hash; every row of products goes through the identical function and lands in the same place. Because both sides use the same partitioning, all rows for product 4017, from either table, converge on one partition on one executor. The join then runs partition by partition, each one holding everything it needs locally. Two full tables have crossed the network to make that true.
The bill is on the inputs
Be precise about what you pay for. The shuffle moves the input tables, not the join result. Join a 400 million row orders table to a 2 million row products table and roughly 402 million rows are hashed, staged to disk, and sent across the network before a single match is produced. A join that returns 10 rows can still shuffle terabytes to find them. When you profile a slow join, the number that predicts the pain is the combined size of the two inputs, and the shuffle family taught you where to read it: the shuffle write and shuffle read on the join's stages.
same key, same machine
the join contract
This shuffle-both-sides plan has one great virtue: it always works. It makes no assumptions about size, it spills to disk when memory runs short, and it scales to joining two tables of any size. That generality is why it is the engine's default posture. But generality has a price, and you now know exactly what it is: two full shuffle write and read cycles, the most expensive operation Spark performs, executed twice, plus a synchronisation barrier where every task waits for the exchange to finish before any matching begins.
So the beginner question about any join is not whether it will work; it will. The question is whether you have to pay for the full meeting. If both tables are genuinely large, the answer is yes, and your job is to make the shuffle move as little as possible by filtering both sides first. But a remarkable share of real joins have a second property that changes everything: one side is tiny. When that is true, Spark can skip the meeting entirely, and that trick is the next section.
Broadcasting the Small Side
You can now describe how a broadcast join ships the small table everywhere and deletes the shuffle.
Look at the orders-to-products join again with fresh eyes. Orders is 400 million rows; products is 2 million rows, maybe 80 megabytes on disk. The default plan shuffles both, which means 400 million order rows cross the network to meet a table that would fit in the memory of a phone. That is backwards. If products is small enough to hold in memory, the cheaper move is obvious: leave the giant table exactly where it is, and send a complete copy of the small table to every executor instead.
That is a broadcast join, and its mechanics are worth having precisely. The small table is collected into one place, serialised, and shipped whole to every executor in the cluster. Each executor builds it into an in-memory hash map keyed by the join key. Then the big table is joined without moving at all: each partition of orders streams its rows, in place, probing the local hash map for each product_id. Every lookup is answered from memory on the same machine. The big side behaves like a narrow transformation: per-partition work, no network, no staging to disk, no barrier.
Tally what just disappeared. No shuffle write of 400 million rows, no shuffle read, no disk staging for the big side, and no stage boundary where fast executors idle waiting for slow ones. The join folds into whatever narrow work surrounds it, often pipelining with the filters and column work before and after. This is why getting the right join to broadcast is one of the highest-leverage single-line changes in Spark: it does not shave the expensive thing, it deletes it.
The big table never moves
Each partition joins in place, like a narrow transformation: no shuffle, no barrier.
The small table travels once per executor
A full copy is shipped to every executor and held as an in-memory hash map.
Memory is the new constraint
Every executor holds the entire small table while the join runs. Small has to mean it.
The map-side join
You will hear this called a broadcast hash join, or in older MapReduce vocabulary a map-side join, and the old name is descriptive: the join happens on the map side, before any shuffle, because every worker already holds the whole lookup table. The hash map is built once per executor and shared by all of that executor's tasks, so the cost of the broadcast is paid per executor, not per task.
Nothing here is free, of course; the cost has moved rather than vanished. The cluster now holds one full copy of the small table per executor, in memory, in hash-map form. For a country lookup that is nothing. For a table that is merely smallish, it is a real bill, and judging that line, what small enough actually means, is exactly the next section's job.
Which Side Is Small Enough
You can now judge from a table's nature and size whether it is a broadcast candidate.
The broadcast trade is network for memory: you avoid shuffling the big table by making every executor hold the entire small one. So the operative question about the small side is not how many rows it has but whether a full copy sits comfortably in one executor's memory next to everything else that executor is doing: shuffle buffers, cached data, the tasks themselves. If the answer is comfortably yes, on every executor, with room to spare after the hash-map form inflates it, broadcast is on the table. If you have to squint, it is not.
The reliable instinct is to think in categories of tables, not bytes. Lookup and dimension tables describe a bounded world: countries, currencies, product catalogs, store locations, status codes. They are small because reality is small, and they grow slowly if at all. Fact and event tables record things happening: orders, clicks, payments, log lines. They grow with the business, without limit. Dimensions are broadcast candidates; facts never are. Most production joins attach facts to dimensions, which is why broadcast joins are everywhere once you start looking for them.
| Table | What it is | Broadcast it? |
|---|---|---|
| country_codes, 200 rows | A bounded lookup | Always |
| products, 2M rows | A dimension, tens of megabytes | Usually |
| users, 80M rows | A dimension that grew into a fact | Rarely; measure first |
| orders, 400M rows and growing | A fact table | Never |
Notice the asymmetry in the rule: only one side has to be small. Joining 10 terabytes to 5 megabytes is one of the cheapest operations in Spark, because the 10 terabytes never move. Joining 10 terabytes to 50 gigabytes is a genuinely heavy shuffle no matter how you arrange it. The size of the big side is almost irrelevant to whether broadcast applies; the whole question lives on the small side. So the habit worth building is to ask, of every join you write: is either side of this a lookup table?
Small today is not small forever
The subtle failure is the table that qualified when the code was written. A products table of 80 megabytes broadcasts beautifully; 3 years of catalog growth later it is 4 gigabytes, and the same line of code is quietly hurting every executor on the cluster. A broadcast decision is a bet about a table's future, not just its present, and the tables worth betting on are the ones whose size is bounded by the real world rather than by the growth of the business.
This section gave you the intuition on purpose and withheld the numbers on purpose. Spark draws its own automatic line for what it will broadcast, that line is a configuration value with a famous default, and you can overrule it in either direction. Those numbers, and the tools for forcing the decision, are where the intermediate tier begins. What you have now is the judgment the numbers serve: broadcast is for tables that describe the world; shuffles are for tables that record it.
TIP
In an interview, the crisp version is: "If one side of the join fits comfortably in each executor's memory, broadcast it: every executor gets a full copy and the big table joins in place with no shuffle. If both sides are large, both get shuffled by key." Sizing the small side is the whole decision.
Join Type vs Join Strategy
You can now separate the join type you wrote from the strategy Spark chose, and debug each on its own axis.
Two vocabularies collide in every conversation about joins, and keeping them apart is a mark of someone who understands the system. The first vocabulary is the join type: inner, left outer, right outer, full outer, semi, anti. That is logic. It answers one question: which rows appear in the result. The second vocabulary is the join strategy: broadcast the small side, or shuffle both sides. That is physics. It answers a different question: how do matching rows physically meet. The type is in your code and changes the answer; the strategy is in the engine and changes the runtime.
The two are almost fully orthogonal. The same inner join of orders to products produces identical results whether Spark broadcasts products or shuffles both tables; only the cost differs. And a left outer join can ride a broadcast just as an inner join can. When you tune a join you are moving along the strategy axis with the type held fixed; when you fix a correctness bug you are moving along the type axis, and the strategy has nothing to say about it.
✓Join type (the WHAT)
- inner, left, right, full, semi, anti
- Decides which rows survive
- Written by you, in the code
- Changes the answer
•Join strategy (the HOW)
- Broadcast the small side, or shuffle both
- Decides how rows physically meet
- Chosen by the engine, from sizes
- Changes the runtime and the bill
The practical payoff of the separation is diagnostic. Wrong numbers in the output? That is the WHAT: check the join type and the join keys, because no strategy can corrupt a result, only reshape its cost. Right numbers, terrible runtime? That is the HOW: check which strategy ran, because the logic does not care how the rows met. Engineers who blur the two end up switching a left join to an inner join to make a job faster, which changes the answer, or adding a broadcast to fix missing rows, which cannot possibly help.
Where type does constrain strategy
One asymmetry does couple them, and it is worth knowing early: an outer join constrains which side may be broadcast. In a left outer join, every left row must appear in the result even if it matches nothing, and an executor holding a full copy of the right table can safely decide that for its local left rows. Flip it around and it breaks: an executor holding only a copy of the left table sees just a slice of the right side, so it can never be sure a left row is truly unmatched. The rule of thumb: the side whose rows must all survive stays put and streams; only the other side is a broadcast candidate.
Multiple Choice
A left outer join attaches a small regions lookup to a huge events table and runs slowly. Which change speeds it up without changing the result?
Carry the separation forward, because the next two tiers live almost entirely on the strategy axis. The join types you already know from SQL transfer to Spark unchanged; nothing distributed alters what an inner or left join means. What is new in Spark is the HOW, and the vocabulary of this section lets you say precisely what the rest of the family is about: choosing, forcing, and eventually eliminating the physical meeting, while the logical answer never moves.
The Duplicate-Key Blowup
You can now predict and prevent the row multiplication a duplicated join key causes.
There is one way a join hurts you that has nothing to do with strategy, and it belongs in the beginner tier because it produces wrong bills and wrong dashboards, not just slow jobs. A join matches every qualifying pair. If a key appears m times on the left and n times on the right, the result contains m times n rows for that key. One-to-one and one-to-many joins behave the way intuition expects. Many-to-many joins multiply, and they usually multiply by accident.
| product_id | Rows in orders | Rows in products | Rows out |
|---|---|---|---|
| 4017 | 3 | 1 | 3 |
| 4018 | 5 | 1 | 5 |
| 4019 | 4 | 3 (duplicated!) | 12 |
| All keys | 12 | 5 | 20 |
Where do the duplicates come from? Almost never from malice. A dimension table that keeps history holds one row per product per version, and joining on product_id alone matches every version. A feed loaded twice inserts every row twice. A join key that is not actually unique, an order_id in a table that is secretly one row per order item, quietly turns a lookup into a multiplier. In each case the author believed the right side held one row per key, and the belief was stale or wrong.
The damage lands twice. First in the results: every duplicated match double-counts whatever you aggregate downstream, and revenue that is suddenly 12 percent high because 12 percent of products carried two catalog versions is a genuinely evil bug to find. Second in the runtime: the blowup happens at the join, so every stage after it processes the multiplied data. A 400 million row table joined against doubled keys becomes 800 million rows, and every shuffle after the join pays for all of them. The shuffles you learned to fear move the inputs; a blowup inflates everything downstream too.
Guard the grain
The defence is to know the grain of both sides before joining: what does one row mean in this table? If the join plan assumes products is one row per product_id, verify it cheaply, a count of rows against a count of distinct keys, and make the pipeline fail loudly when the assumption breaks rather than multiply silently. Deduplicate deliberately, choosing which version of a duplicated key is correct, never reflexively; a dropDuplicates bolted on after a blowup picks an arbitrary row and converts an obvious wrong answer into a subtle one.
You now have the whole beginner model of a Spark join. Matching keys must be co-located, which costs a double shuffle unless one side is small enough to broadcast instead. The join type is logic, the strategy is physics, and they are tuned separately. And the row count out of a join is a product, not a sum, so the grain of each side is part of the contract. The intermediate tier turns this model quantitative: the full menu of strategies Spark chooses from, the exact threshold that triggers a broadcast, and how to check and overrule the engine's choice.
✓Do
- Estimate both sides of every join before you write it; the small side decides the strategy.
- Ask of every join whether one side is a bounded lookup or dimension table.
- Verify the join key's uniqueness on the one-per-key side before trusting the grain.
- Filter both tables before the join so whatever movement happens moves less.
✗Don't
- Don't change the join type to fix performance; type is logic and changes the answer.
- Don't broadcast a table that grows with the business, no matter how small it is today.
- Don't patch a duplicate-key blowup with dropDuplicates after the join; fix the grain.
- Don't judge a join by its output size; the shuffle bill is set by the inputs.
❯❯❯PUTTING IT ALL TOGETHER
> A nightly job that joins a 600 million row orders table to a small store locations table has started taking 4 hours, and finance also reports that this month's revenue dashboard is mysteriously 15 percent high.
The runtime smells like strategy: a small locations table on the non-preserved side is a textbook broadcast candidate, and 4 hours suggests both sides are being shuffled instead.
The wrong revenue is a different axis entirely: strategy cannot corrupt results, so you check the grain of the join instead of the physics.
A distinct-key count on locations shows duplicated store rows from a re-run load: every duplicated key multiplied its orders, inflating revenue downstream.
You fix the duplicate load and broadcast the small side: the type-versus-strategy split turned one confusing incident into two clean, separate fixes.
KEY TAKEAWAYS
A join must co-locate matching keys, and the default plan shuffles both tables by the join key to do it.
A broadcast join ships a full copy of a small table to every executor so the big table joins in place with no shuffle.
Broadcast candidates are bounded lookups and dimensions; tables that grow with the business are not.
Join type decides which rows survive; join strategy decides how they meet; tune them separately.
A key appearing m times on one side and n on the other emits m x n rows; guard the grain of both sides.
A join is a meeting. Someone has to pay for the travel.
- Category
- Spark
- Difficulty
- beginner
- Duration
- 13 minutes
- Challenges
- 15 hands-on challenges
Topics covered: Why a Join Moves Data, Broadcasting the Small Side, Which Side Is Small Enough, Join Type vs Join Strategy, The Duplicate-Key Blowup
Lesson Sections
- Why a Join Moves Data (concepts: paDistributedPrimitives)
Start with what a join physically has to do. To match an order to its product, Spark must compare two rows, one from orders and one from products, sharing a product_id. But those rows live wherever the data happened to be read: the order sits in a partition on one machine, the product in a partition on another. No amount of cleverness lets two machines compare rows without at least one of them moving. A join is therefore a co-location problem before it is a matching problem: every pair of rows t
- Broadcasting the Small Side (concepts: paBroadcastJoin)
Look at the orders-to-products join again with fresh eyes. Orders is 400 million rows; products is 2 million rows, maybe 80 megabytes on disk. The default plan shuffles both, which means 400 million order rows cross the network to meet a table that would fit in the memory of a phone. That is backwards. If products is small enough to hold in memory, the cheaper move is obvious: leave the giant table exactly where it is, and send a complete copy of the small table to every executor instead. That i
- Which Side Is Small Enough (concepts: paBroadcastJoin)
The broadcast trade is network for memory: you avoid shuffling the big table by making every executor hold the entire small one. So the operative question about the small side is not how many rows it has but whether a full copy sits comfortably in one executor's memory next to everything else that executor is doing: shuffle buffers, cached data, the tasks themselves. If the answer is comfortably yes, on every executor, with room to spare after the hash-map form inflates it, broadcast is on the t
- Join Type vs Join Strategy (concepts: paDistributedPrimitives)
Two vocabularies collide in every conversation about joins, and keeping them apart is a mark of someone who understands the system. The first vocabulary is the join type: inner, left outer, right outer, full outer, semi, anti. That is logic. It answers one question: which rows appear in the result. The second vocabulary is the join strategy: broadcast the small side, or shuffle both sides. That is physics. It answers a different question: how do matching rows physically meet. The type is in your
- The Duplicate-Key Blowup (concepts: paDistributedPrimitives)
There is one way a join hurts you that has nothing to do with strategy, and it belongs in the beginner tier because it produces wrong bills and wrong dashboards, not just slow jobs. A join matches every qualifying pair. If a key appears m times on the left and n times on the right, the result contains m times n rows for that key. One-to-one and one-to-many joins behave the way intuition expects. Many-to-many joins multiply, and they usually multiply by accident. Where do the duplicates come from