Y

Apache Spark · Advanced

Joins, shuffles and skew

4 min read

Why it matters

When a Spark job is slow or runs out of memory, the cause is almost always a shuffle: a join or aggregation that moves data between partitions. Senior data engineering interviews give you a slow job and ask what you would look at. Reading the physical plan is the answer.

The concept

Spark picks a join strategy per join:

  • Sort-merge join: both sides are shuffled by the join key (an Exchange hashpartitioning in the plan), sorted and merged. Works for any size; costs a full shuffle of both sides.

  • Broadcast hash join: the small side is copied to every executor, and the big side is joined in place with no shuffle. Fast, but only when the small side fits in memory. Spark does this automatically below spark.sql.autoBroadcastJoinThreshold (10 MB by default), if it knows the size.

Other levers:

  • spark.sql.shuffle.partitions (default 200) sets how many partitions a shuffle produces. Too many for small data wastes time on tiny tasks; too few for big data creates huge tasks that spill to disk.

  • Adaptive Query Execution (AQE), on by default in Spark 3.2 and later, adjusts partitions and join strategies at runtime using real statistics.

  • Skew: when a few keys hold much of the data, one task gets far more work than the others and the whole stage waits for it.

Worked example: two plans for one join

spark_plan.py
import time
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.master("local[*]").appName("plans")
         .config("spark.sql.autoBroadcastJoinThreshold", -1)      # turn off automatic broadcasts to compare
         .getOrCreate())
trips = spark.read.parquet("yellow_tripdata_2025-01.parquet")
zones = spark.read.option("header", True).option("inferSchema", True).csv("taxi_zone_lookup.csv")

def timed(label, frame):
    start = time.time()
    rows = frame.groupBy("Borough").count().collect()
    print(f"{label}: {time.time() - start:.1f}s, {len(rows)} boroughs")

shuffle_join = trips.join(zones, trips.PULocationID == zones.LocationID)
broadcast_join = trips.join(F.broadcast(zones), trips.PULocationID == zones.LocationID)
shuffle_join.explain()
broadcast_join.explain()
timed("sort-merge join", shuffle_join)
timed("broadcast join", broadcast_join)
print("shuffle partitions:", spark.conf.get("spark.sql.shuffle.partitions"))
print("skew:", trips.groupBy("PULocationID").count().orderBy(F.desc("count")).limit(3).collect())
spark.stop()

The first plan (shortened):

text
SortMergeJoin [PULocationID], [LocationID], Inner
:- Sort [PULocationID]
:  +- Exchange hashpartitioning(PULocationID, 200)
:     +- FileScan parquet ... PushedFilters: [IsNotNull(PULocationID)]
+- Sort [LocationID]
   +- Exchange hashpartitioning(LocationID, 200)
      +- FileScan csv ...

Two Exchange operators: 3.5 million trips and the zone table are both shuffled into 200 partitions. The second plan:

text
BroadcastHashJoin [PULocationID], [LocationID], Inner, BuildRight
:- FileScan parquet ...
+- BroadcastExchange HashedRelationBroadcastMode(...)
   +- FileScan csv ...

No shuffle of the trips at all. On a laptop the sort-merge version took about 3.8 seconds and the broadcast version about 0.8 seconds. On a cluster with billions of rows, the difference is minutes of cluster time on every run.

The skew check shows the busiest pickup zones: 161 (169,977 trips), 237 (163,703) and 236 (155,647). Here the keys are fairly balanced; in real systems, keys like "unknown customer" or one huge merchant can hold 30% of rows. Remedies: broadcast the other side, let AQE split skewed partitions (spark.sql.adaptive.skewJoin.enabled), or "salt" the hot key into several sub-keys.

Common mistakes

  • Relying on automatic broadcasts after transformations. Spark may not know a filtered table is small. Add F.broadcast() explicitly when you know.

  • Broadcasting something big. Broadcasting a multi-gigabyte table can crash executors. Check sizes first.

  • Leaving 200 shuffle partitions for everything. Let AQE coalesce, or set it to fit the data size.

  • Fixing by adding machines. More executors do not help a single skewed task. Find the key.

Interview angle

"A daily Spark job went from 20 minutes to 3 hours. How do you investigate?" Answer: open the Spark UI, find the slowest stage, check whether one task takes far longer (skew) or spills to disk, read the plan for unexpected shuffles or a lost broadcast, check whether input size or key distribution changed. Then fix the cause, not the cluster size.

Practice task

  1. Join trips to zones twice (pickup and dropoff). Make both joins broadcast and verify in the plan.

  2. Set spark.sql.shuffle.partitions to 8 and to 2,000 for the sort-merge version and compare times.

  3. Create an artificial skew by setting 50% of PULocationID values to 1, and try salting.

Further reading