What is a Shuffle?

How it works

Narrow transformations such as filter and select work on each partition independently. Wide transformations need rows from many partitions at once, so Spark inserts a shuffle, shown as an Exchange in the physical plan, and starts a new stage.

On the map side, each task computes the target partition of every row (by hashing the key modulo the number of shuffle partitions), sorts its output by target partition, spills to local disk if the buffer fills, and writes one data file plus an index file. On the reduce side, each task fetches its slice from every map task’s output over the network, then aggregates, joins or sorts it. No reduce task can start until every map task has finished.

A shuffle pays for the same data several times: serialisation and compression, local disk writes, network transfer (M map tasks and R reduce partitions mean M times R blocks), then reads and deserialisation. After a DataFrame shuffle, the number of partitions is spark.sql.shuffle.partitions (200 by default), which Adaptive Query Execution can coalesce at run time.

Ways to make shuffles cheaper:

• Filter rows and select columns before the wide operation.
• Aggregate before joining, and rely on map-side partial aggregation in groupBy.
• Broadcast the small side of a join so the large side never moves.
• Reuse one partitioning for several operations on the same key, or bucket tables that are joined repeatedly.
• Remove needless orderBy, distinct and repartition calls.

Example

A groupBy adds an Exchange to the plan, with a partial aggregation before it so only one row per key per map task crosses the network (PySpark 4.2.0).

import re
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.master("local[2]").appName("glossary-shuffle")
         .config("spark.sql.adaptive.enabled", "false").getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

df = spark.range(1000).withColumn("key", F.col("id") % 3)
agg = df.groupBy("key").agg(F.sum("id").alias("total"))
plan = agg._jdf.queryExecution().executedPlan().toString()
for line in plan.splitlines():
    if "Exchange" in line or "HashAggregate" in line:
        line = re.sub(r"#\d+L?|\*\(\d+\) |, \[plan_id=\d+\]", "", line)  # drop run-specific ids
        print(line.split(", output")[0].strip(" +-:"))
spark.stop()

HashAggregate(keys=[key], functions=[sum(id)]
Exchange hashpartitioning(key, 200), ENSURE_REQUIREMENTS
HashAggregate(keys=[key], functions=[partial_sum(id)]

Common mistakes

• Leaving spark.sql.shuffle.partitions at 200 for every job size without AQE; tiny jobs get 200 near-empty tasks and huge ones get oversized partitions.
• Joining or grouping before filtering, so far more data crosses the network than needed.
• Treating a FetchFailedException as a network problem when it usually means an executor holding shuffle files died from memory pressure.
• Calling repartition “to be safe” before every write, which adds a full shuffle each time.