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

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 shuffleSpark ShuffleThe 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..
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 executionAdaptive Query ExecutionSpark'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. splits such a partition and copies the matching rows of the other side; the plan then shows SortMergeJoin(skew=true). More in data skewShuffle and Data SkewA 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..
• Spill. The Sort step holds a partition in memory; a partition too big for its share spillsSpark SpillWhen 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. 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 schemaStar SchemaA 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., avoids it.

Our Spark UI guide opens a real skewed sort-merge join in the SQL tab and on its stage page, before and after the fix.