# How does a sort-merge join work in Spark, and when does Spark choose it?

Spark's default join for 2 large tables: it shuffles both by the join key, sorts each partition, then walks the 2 sorted sides together to find matching rows.

- Page: https://tecyfy.com/concepts/sort-merge-join
- Subject: Data engineering in the AI era (https://tecyfy.com/concepts)
- Level: intermediate
- Also called: SortMergeJoin, Shuffle sort merge join, MERGE join hint, Broadcast hash join vs sort-merge join

A join has to bring together the rows of 2 tables that share a key. When both tables are large, no executor can hold either one whole, so Spark splits the work by key. A sort-merge join does it in 3 steps, and each step is a box you can find in the plan.

## The 3 steps

1. **Shuffle both sides.** Each table is redistributed by a hash of the join key into the same number of partitions, so partition 7 of the orders and partition 7 of the customers hold the same keys. In a plan this is an `Exchange hashpartitioning(customer_id, 200)` on each side. See [shuffle](https://tecyfy.com/concepts/spark-shuffle).
2. **Sort each partition** by the key, a `Sort` box on each side.
3. **Merge.** With both sides in key order, one pass down each finds the matches, the way you would match 2 sorted lists by hand. Neither side needs to fit in memory as a lookup table.

Running it ourselves on Spark 4.2, with broadcasting turned off, an orders-to-customers join planned exactly this way:

```
SortMergeJoin [customer_id], [customer_id], Inner
:- Sort [customer_id ASC NULLS FIRST]
:  +- Exchange hashpartitioning(customer_id, 200)
+- Sort [customer_id ASC NULLS FIRST]
   +- Exchange hashpartitioning(customer_id, 200)
```

## When Spark picks it

Spark prefers cheaper joins when it can. If one side is under `spark.sql.autoBroadcastJoinThreshold`, 10 MB by default, it plans a broadcast hash join instead: the small table goes to every executor and the big one is never shuffled. With the default threshold, the same join above became a `BroadcastHashJoin`, because the customers table was 10.9 KiB. Sort-merge is the choice when both sides are too big to broadcast, since `spark.sql.join.preferSortMergeJoin` is true. A shuffled hash join, which builds a hash table per partition instead of sorting, is used only when settings or a hint ask for it.

You can ask for a strategy with a hint: `MERGE` for sort-merge, `BROADCAST`, or `SHUFFLE_HASH`. When both sides carry different hints, BROADCAST wins over MERGE, which wins over SHUFFLE_HASH, and Spark ignores a hint the join type cannot use.

## What goes wrong

- **Skew.** All rows of one key land in one partition, so one task sorts and merges far more than the others. [Adaptive query execution](https://tecyfy.com/concepts/adaptive-query-execution) splits such a partition and copies the matching rows of the other side; the plan then shows `SortMergeJoin(skew=true)`. More in [data skew](https://tecyfy.com/concepts/data-skew).
- **Spill.** The Sort step holds a partition in memory; a partition too big for its share [spills](https://tecyfy.com/concepts/spark-spill) to disk. The Sort box in the SQL tab shows `spill size` and `peak memory`.
- **A side that was small after all.** If a filter leaves one side tiny, AQE can switch to a broadcast join at runtime, but the shuffle has already been paid for. Filtering early and joining small tables to big ones, as in a [star schema](https://tecyfy.com/concepts/star-schema), avoids it.

Our [Spark UI guide](/data-engineering/spark-ui) opens a real skewed sort-merge join in the SQL tab and on its stage page, before and after the fix.

## Related

- [Spark Shuffle](https://tecyfy.com/concepts/spark-shuffle): The step that moves rows between executors so rows with the same key end up together, for a join, group by or repartition; it splits a job into stages.
- [Adaptive Query Execution](https://tecyfy.com/concepts/adaptive-query-execution): Spark's way of re-planning a query while it runs: after each shuffle it looks at the real sizes, then merges small partitions, splits skewed ones and can switch a join to a broadcast.
- [Shuffle and Data Skew](https://tecyfy.com/concepts/data-skew): A shuffle moves rows between machines so rows with the same key meet, for a join or group by; skew is when one key has far more rows, so one task does most of the work.
- [Spark Spill](https://tecyfy.com/concepts/spark-spill): When a task's sort, join or aggregation no longer fits in execution memory, Spark writes part of it to local disk and reads it back later, so the task finishes, only slower.
- [Star Schema](https://tecyfy.com/concepts/star-schema): A way to arrange warehouse tables for reporting: a fact table of events and their numbers, joined by keys to dimension tables that describe them, such as customer, product and date.

## Sources

- [Hints: Join hints (Spark 4.2.0)](https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html)
- [Performance Tuning (Spark 4.2.0)](https://spark.apache.org/docs/latest/sql-performance-tuning.html)
- [Web UI (Spark 4.2.0)](https://spark.apache.org/docs/latest/web-ui.html)