What is a shuffle in Spark, and why is it expensive?

Summary

• A shuffle is Spark's mechanism for re-distributing data so that it's grouped differently across partitions. The Spark docs call it "a complex and costly operation" and say it "involves disk I/O, data serialization, and network I/O."
• In a Spark SQL physical plan, a shuffle shows up as an Exchange node. Spark's ShuffleExchangeExec operator "performs a shuffle that will result in the desired partitioning."
• spark.sql.shuffle.partitions, "the number of partitions to use when shuffling data for joins or aggregations", defaults to 200. Adaptive query execution (AQE), on by default since Spark 3.2.0, can coalesce small shuffle partitions at runtime.
• A BROADCAST join hint broadcasts one join side, and the Spark source notes "The streamed relation is not shuffled."
• On Databricks, setting spark.sql.shuffle.partitions to auto enables auto-optimized shuffle.

What is a shuffle in Spark, and why is it expensive?

"The shuffle is Spark's mechanism for re-distributing data so that it's grouped differently across partitions. This typically involves copying data across executors and machines, making the shuffle a complex and costly operation." The RDD programming guide says it "involves disk I/O, data serialization, and network I/O". Shuffle operations also build in-memory data structures, and when data does not fit Spark spills them to disk, which adds disk I/O and garbage collection.

How do I spot a shuffle in a query plan?

EXPLAIN prints the physical plan by default. In Spark's source, the shuffle operator ShuffleExchangeExec is documented as "Performs a shuffle that will result in the desired partitioning", and its node name is Exchange. In the Spark docs' group-by example, an Exchange sits between the two aggregation steps:

EXPLAIN select k, sum(v) from values (1, 2), (1, 3) t(k, v) group by k;

== Physical Plan ==
*(2) HashAggregate(keys=[k#33], functions=[sum(cast(v#34 as bigint))])
+- Exchange hashpartitioning(k#33, 200), true, [id=#59]
   +- *(1) HashAggregate(keys=[k#33], functions=[partial_sum(cast(v#34 as bigint))])
      +- *(1) LocalTableScan [k#33, v#34]

The partial_sum aggregate sits below the Exchange and the final sum aggregate above it. In the Spark UI, stage DAG nodes carry names such as Exchange, and the SQL tab's Details link expands the full physical plan.

Which Spark SQL settings control shuffles?

Setting	Default	What the Spark docs say
spark.sql.shuffle.partitions	200	Number of partitions to use when shuffling data for joins or aggregations
spark.sql.adaptive.enabled	true	Re-optimizes the query plan in the middle of query execution, based on runtime statistics
spark.sql.adaptive.coalescePartitions.enabled	true	Coalesces contiguous shuffle partitions to the target size, to avoid too many small tasks
spark.sql.adaptive.advisoryPartitionSizeInBytes	64 MB	Advisory size of a shuffle partition during adaptive optimization
spark.sql.autoBroadcastJoinThreshold	10 MB	Maximum size of a table broadcast to all worker nodes in a join; -1 disables broadcasting

For Structured Streaming, spark.sql.shuffle.partitions cannot be changed between query restarts from the same checkpoint location.

How do I avoid shuffling a large table in a join?

A BROADCAST hint prioritizes a broadcast join with the hinted table as the build side, even when its statistics exceed spark.sql.autoBroadcastJoinThreshold. Spark may not choose the hinted strategy for every join type.

SELECT /*+ BROADCAST(r) */ * FROM src s JOIN records r ON s.key = r.key

The scaladoc for Spark's BroadcastHashJoinExec says the values are placed in a Spark broadcast variable and "The streamed relation is not shuffled." The other join hints request a shuffle sort merge join (MERGE), a shuffle hash join (SHUFFLE_HASH) or a shuffle-and-replicate nested loop join (SHUFFLE_REPLICATE_NL). AQE can also convert a sort-merge join to a broadcast hash join at runtime, which the Spark docs call "not as efficient as planning a broadcast hash join in the first place".

How do I reduce the cost of a shuffle in my code?

• RDD aggregations: the groupByKey docs say reduceByKey or aggregateByKey "will provide much better performance" for a sum or average per key. reduceByKey merges locally on each mapper before sending results to a reducer.
• Fewer partitions: the DataFrame coalesce docs say going from many partitions to fewer "results in a narrow dependency", so there is no shuffle. Calling repartition() instead "will add a shuffle step".
• Out-of-memory reduce tasks: shuffle operations "build a hash table within each task", and the tuning guide's simplest fix is to increase the level of parallelism so each task's input is smaller.

How does Databricks handle shuffles?

On Databricks, spark.sql.shuffle.partitions defaults to 200, and setting it to auto "enables auto-optimized shuffle, which automatically determines this number based on the query plan and the query input data size." The Databricks Spark UI guide recommends spark.sql.shuffle.partitions=auto for stages with high shuffle.
The Databricks AQE docs explain that Databricks "has the most up-to-date accurate statistics at the end of a shuffle and broadcast exchange" and can use them to pick a better physical strategy or an optimal post-shuffle partition size and number. AQE "dynamically coalesces partitions (combine small partitions into reasonably sized partitions) after shuffle exchange", and Photon "uses a redesigned columnar shuffle to increase throughput for large-scale joins."

FAQs

Does every join shuffle?

No. For a broadcast hash join, Spark's source says "The streamed relation is not shuffled." The other strategies in the hint docs, shuffle sort merge join, shuffle hash join and shuffle-and-replicate nested loop join, have shuffle in their names.

Which setting applies to RDD code instead of Spark SQL?

For RDD operations, pass numPartitions or set spark.default.parallelism. For reduceByKey and join the default is the largest number of partitions in a parent RDD.