How Spark SQL joins work
See how Spark brings matching keys together with broadcast and shuffle joins, local algorithms, statistics, skew handling, and adaptive execution.
The whole story in 6 lines
Follow strategy selection through broadcast hash joins, shuffle exchange, local join tasks, skew, and Adaptive Query Execution.
- A distributed join can compare equal keys only after both records meet in the same task.
- Broadcast avoids a two-sided shuffle when one input is small enough, while larger inputs usually require partitioned exchange.
- A broadcast hash join sends one compact input to every executor and builds a local hash table beside each large partition.
- A shuffle exchange serializes, transfers, and may spill both inputs so equal keys land in matching partitions.
- Each local join task must preserve duplicate matches and the null-padding rules of the selected join type.
- Adaptive Query Execution can split skewed partitions or change strategy after runtime sizes replace planning estimates.
Setup
Welcome. This explainer follows how Apache Spark executes a join, step by step. We will start with four ideas that anchor the whole lesson.
A join combines rows from two tables wherever their keys match. Think of it like matching student IDs between an enrollment list and a grade sheet.
A partition is a slice of data that lives on one machine. Spark splits large tables into many partitions so work can happen in parallel.
An exchange is a network shuffle. Spark moves rows between machines so that matching keys end up on the same executor.
Now let us preview the road ahead. We will cover six stages: why keys must meet, how Spark picks a strategy, broadcast joins, shuffle exchanges, local join tasks, and adaptive execution.
Now let us start with the most fundamental idea: why matching keys need to be on the same machine before any comparison can happen.
Why Keys Must Meet
Here is the core question: how do two huge tables, scattered across a cluster, find their matching rows. The answer starts with the fact table on the left.
The dimension table holds the rows we want to match against. Right now these keys live on completely different machines than the fact rows. No comparison is possible yet.
Spark inserts an exchange boundary. This is the moment where rows leave their original partitions and travel across the network, grouped by join key.
The stage has reached the decision that determines its next state. What makes equal join keys meet in the same task when neither side is broadcast?
Matching keys now share a task slot. Each bucket contains all the left rows and all the right rows for its key range.
Local join tasks can finally run. No more network traffic is needed. The key takeaway: the expensive part of a distributed join is moving data, not comparing it. Next, we will see how Spark decides which join strategy to use.
Physical Strategy Choice
★ If you remember one thing · With a tiny build side and fresh statistics, Spark can select broadcast instead of repartitioning the large relation.
Spark must choose a physical strategy that brings equal keys together. The decision starts from the logical join, table statistics, and the size of each input.
The planner reads the fact side first. It needs to know how many rows and how many bytes it is dealing with. These numbers set the baseline cost for every strategy.
The stage has reached the decision that determines its next state. When can Spark avoid shuffling the large relation for an equi-join?
With a tiny build side and fresh statistics, broadcast hash wins because Spark can leave the 920 GB fact relation in place. The two shuffle alternatives remain visible for larger inputs.
Switch the Dim Size control through every choice. Compare the downstream outcome while the earlier input and system boundary remain fixed.
Spark emits one concrete physical operator. That single decision determines how much network traffic, memory, and CPU the join will cost. Next, we will see the fastest option: the broadcast hash join.
Broadcast Hash Join
We just saw how the planner picks a strategy. When it chooses broadcast, here is what actually happens. The driver starts with a physical plan that says "broadcast the small side."
The dimension table is serialized into one compact payload. This is the data that every executor will receive.
The driver ships that payload to every executor in the cluster. This is the one network trip. After this, the large side never moves.
Each executor unpacks the broadcast and builds a local hash table keyed on the join column. This hash table fits entirely in memory.
Fact partitions stream past the hash table. For each fact row, the executor looks up the join key in the hash table. A match means an output row. No match means the row is skipped.
The joined output appears without ever shuffling the large side. That is the broadcast tradeoff: one small copy per executor versus zero network cost for the big table. Next, what happens when the dimension is too large to broadcast.
Shuffle Exchange
When neither input is small enough to broadcast, equal keys still need to meet on one executor. Spark therefore exchanges both sides using the same partitioning rule.
Spark attaches shuffle writers to the left side. Every row gets hashed by the join key, and the hash determines which output partition the row lands in.
The right side does the same using the identical hash function. This is critical. Both sides must agree on the partitioner or keys will end up in different buckets.
Rows travel across the network into their hash buckets. This is the most expensive step. Every row from both tables crosses a network boundary.
Each bucket now contains all matching keys from both sides. Notice how one bucket balloons while others stay small. That imbalance will slow down the entire stage.
The shuffle is done. Aligned partition pairs are ready for local join tasks. The key cost was network serialization and transfer. Next, we will see what happens inside one of these local tasks.
Local Join Task
We saw how the shuffle aligns partitions. Now let us zoom into one executor task that receives a left partition and a right partition. This is where actual row comparison happens.
The task prepares its local state. In sort-merge mode, both sides get sorted by key. In hash mode, the right side builds an in-memory hash table.
Keys are compared inside this single executor. Sort-merge walks two sorted streams with a cursor. Hash mode probes the hash table for each left row. Either way, no network traffic is involved.
Duplicate keys create a cross product. If key 42 appears twice on the left and twice on the right, that is four output rows.
A left join preserves unmatched rows from the left input and fills the missing right-side columns with nulls. An inner join skips those rows. Join semantics therefore determine the final shape of the output independently of the physical strategy.
The task emits its slice of the final result. Every partition produces output independently. Correctness depends on handling duplicates and nulls locally. Next, we will see how AQE repairs a skewed shuffle.
Skew + AQE
Remember the hot bucket from Stage 4. After the shuffle finishes, Spark now has real partition sizes instead of estimates. These runtime metrics are AQE's starting point.
The metrics reveal that partition 2 holds far more data than the others. Spark flags it as a skewed outlier. Without intervention, this one task will take longer than all the rest combined.
AQE now decides whether it can safely rewrite the physical plan. Without adaptive execution, Spark would stay stuck with this uneven partition map.
The hot partition is split into smaller chunks. Spark creates follow-up tasks and replicates the opposite side of the join into each chunk. More tasks, but each one finishes quickly.
The repaired tasks finish faster and more evenly. The wall-clock time for the stage drops because no single task dominates.
The final output is logically identical. AQE changed only the physical execution, not the result. Now let us step back and connect strategy selection, data movement, local execution, and adaptive repair.
Recap
We started with the most basic requirement: matching join keys must be on the same machine. Without co-location, no comparison is possible. That is the foundation everything else builds on.
Then we learned how the Catalyst planner chooses a physical strategy. It reads table statistics, checks the broadcast threshold, and picks the cheapest path. The choice depends on table sizes, join type, and stat freshness.
When one side is small, broadcast hash join avoids shuffling the large table entirely. One copy per executor, zero network cost for the big side.
When broadcast is not an option, both sides pay for a shuffle exchange. Rows travel across the network into hash buckets so matching keys land together.
Inside each executor, the local join task does the real work: sorting and merging, or hashing and probing. Duplicate keys fan out, outer joins add null-padded rows.
Finally, AQE watches runtime metrics and repairs the plan when skew shows up. It splits hot partitions without changing the logical result.
That is the full pipeline. Strategy selection, data movement, local execution, and adaptive repair. Every Spark join you write follows this path.
The whole story in 6 lines
Follow strategy selection through broadcast hash joins, shuffle exchange, local join tasks, skew, and Adaptive Query Execution.
- A distributed join can compare equal keys only after both records meet in the same task.
- Broadcast avoids a two-sided shuffle when one input is small enough, while larger inputs usually require partitioned exchange.
- A broadcast hash join sends one compact input to every executor and builds a local hash table beside each large partition.
- A shuffle exchange serializes, transfers, and may spill both inputs so equal keys land in matching partitions.
- Each local join task must preserve duplicate matches and the null-padding rules of the selected join type.
- Adaptive Query Execution can split skewed partitions or change strategy after runtime sizes replace planning estimates.