The Shuffle: When Spark Redistributes Data Between Executors and Why It Costs You

Shannon Lowder

Shannon Lowder

— 3 min read

If you want to understand why a Spark job is slow, you need to understand the shuffle. The shuffle is the most expensive operation in Spark, and most of the performance tuning work in any non-trivial Spark application comes back to minimizing, eliminating, or managing shuffles.

What a Shuffle Is

Recall the M&M factory: 10 workers, each with 50 bags of M&Ms to sort. They can sort their own bags independently. No coordination needed. Fast.

Now ask them to regroup all M&Ms of the same color together — worker 1 gets all reds, worker 2 gets all blues, and so on. To do this, every worker has to send their colored M&Ms to the right destination worker. M&Ms travel across the factory floor. This is a shuffle.

In Spark, a shuffle happens when data needs to be redistributed across partitions based on a key. Every row needs to be evaluated, a destination partition determined (based on a hash of the key), and rows sent to their destination partition — possibly on a different executor on a different machine. This involves:

  1. Each executor serializes its data and writes it to local disk (shuffle write)
  2. Each executor reads its incoming data from other executors over the network (shuffle read)
  3. Each executor sorts the received data by key

Disk I/O. Network I/O. Deserialization. Sorting. That's why it's expensive.

What Triggers a Shuffle

  • groupBy() — rows with the same key need to land on the same executor
  • join() — matching rows from both sides need to land on the same executor
  • orderBy() / sort() — global sort requires redistributing to produce ordered output
  • repartition(n) — explicitly redistributing data across n partitions
  • distinct() — requires grouping identical rows to deduplicate
  • Window functions with partitionBy — partition groups must co-locate

Identifying Shuffles in the Spark UI

In Databricks, click on a running or completed job. Go to the Stages tab. Each stage boundary represents a shuffle. A job with many stages has many shuffles. Within a stage's details, look at "Shuffle Read Size" and "Shuffle Write Size" — these numbers tell you how much data moved across the network for that shuffle.

High shuffle read/write on a join usually means the join keys aren't well-distributed (data skew) or that the join is larger than it needs to be (filtering before joining would help). High shuffle read/write on a groupBy after a large scan usually means pre-aggregation before the shuffle would help.

Strategies to Reduce Shuffle Cost

1. Broadcast small tables in joins

If one side of a join is small enough to fit in memory on every executor, Spark can broadcast it — send a copy to every executor — eliminating the need to shuffle the large table. No shuffle on the large side means dramatically lower cost.

from pyspark.sql import functions as F

# Hint Spark to broadcast the small dimension table
large_df.join(
    F.broadcast(small_dim_df),
    on="product_id",
    how="inner"
)

# Or let Spark decide via autoBroadcastJoinThreshold (default 10MB)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(50 * 1024 * 1024))  # 50MB

2. Filter before joining or grouping

# Less data in the shuffle = cheaper shuffle
filtered = large_df.filter(F.col("event_date") >= "2020-01-01")
result = filtered.groupBy("user_id").agg(F.count("*"))

3. Set shuffle partitions appropriately

# 200 is the default -- often too many for small/medium clusters
spark.conf.set("spark.sql.shuffle.partitions", "32")

# Or in Spark 3.x, enable Adaptive Query Execution to tune this automatically
spark.conf.set("spark.sql.adaptive.enabled", "true")

4. Avoid unnecessary re-shuffles

# If you need to groupBy twice on the same key, chain them
# instead of materializing an intermediate result
result = df.groupBy("customer_id", "month")            .agg(F.sum("amount").alias("monthly_revenue"))            .groupBy("customer_id")            .agg(F.avg("monthly_revenue").alias("avg_monthly_revenue"))

# Spark can sometimes optimize this into one pass -- check explain()
result.explain()

Data Skew: When One Executor Gets All the M&Ms

Shuffle skew is when one partition contains dramatically more data than others. If 80% of your rows have the same groupBy key, 80% of the shuffle lands on one executor. The other 9 workers finish quickly. That one worker takes 10x longer. Your job's total time is dictated by the slowest task.

Symptoms in the Spark UI: one task in a stage takes much longer than all the others. The Stage page's task list will show median task time of 5 seconds and a maximum of 300 seconds.

Common skew sources: null keys in joins (all nulls hash to the same partition), categorical values with extremely uneven distribution, or a join key that's not the natural key (like a status field with one value dominating).

Solutions: filter out nulls before joining, salt the skewed key (add a random prefix, join twice and union), or use Spark's built-in skew join hint (skewJoin in recent versions of Databricks Runtime). Skew handling is a post for its own day, but recognizing it from the Spark UI is the first step.

Read more