🔥 Spark & Databricks · Chapter 4

Shuffles — the thing that makes jobs slow

Every Exchange in your plan is a full copy of the data across the network. Count them before you run anything.

Study this lesson free in the app →

The basics

A shuffle is Spark redistributing rows across the cluster so that all rows sharing a key end up in the same partition.

Why it exists: to sum fares per city, every Shillong ride must be in one place. But the rides were loaded in whatever order the files came — Shillong rides are spread across all 320 partitions. Spark has to move them. That movement is the shuffle: each task writes its rows into buckets by key on local disk, and the next stage's tasks fetch their bucket from every executor over the network.

Mental model: a post-office sorting room. Mail arrives in sacks from every town, mixed. Before delivery, every letter is sorted into the bin for its destination — every sack touches every bin. That sorting step is expensive no matter how fast the postmen walk.

Explained

Meghalaya Cabs' driver-earnings job reads 400GB, groups by driver, joins the driver table, and sorts by earnings.

explain() shows three Exchange nodes. That is three times the job writes data to disk, sends it across the network and reads it back.

Rewritten — broadcast the driver table, drop the sort nobody reads — the plan has one. Same output. A third of the runtime.

The post-office sorting room

Fifty sacks arrive, one from each town, each holding letters for every district. To deliver, the office must get all Mawlai letters into one bin.

Every sack is opened, every letter thrown into one of 200 bins, and each postman collects his bin from every sorting table. Every sack touches every bin. That is the shuffle.

Three ways to make it cheaper: throw away junk mail before sorting (filter first), bundle letters to the same house with a rubber band (pre-aggregate), or — if one district's list is tiny — photocopy it for every postman instead of sorting anything (broadcast).

What one shuffle actually does

Serialise, write, network, read, deserialise — for every row that crosses. That is why one shuffle can outweigh all the narrow work in a job.

What triggers one — and what doesn't

Shuffle (Exchange):

• groupBy().agg(), distinct(), dropDuplicates() → hashpartitioning on the keys
• join() (sort-merge or shuffle-hash) → hashpartitioning on the join keys, both sides
• orderBy() / sort() → rangepartitioning (Spark samples the data first to pick ranges)
• repartition(n) → round-robin; repartition('col') → hashpartitioning
• Window with partitionBy → hashpartitioning on the window keys
• intersect, except

No shuffle:

• filter, select, withColumn, union
• coalesce(n) — merges neighbouring partitions locally
• A join where one side is broadcast — only the small side moves
• sortWithinPartitions — sorts each partition in place

And one optimisation: if a child is already hash-partitioned on the needed key with the right partition count, Spark's planner (EnsureRequirements) inserts no Exchange.

Reading Exchange in a plan

rev = (rides.filter(col('status') == 'COMPLETED')
            .groupBy('city').agg(sum('fare_inr')))
rev.explain()

AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[city], functions=[sum(fare_inr)])        ← final
   +- Exchange hashpartitioning(city, 200), ENSURE_REQUIREMENTS ← SHUFFLE
      +- HashAggregate(keys=[city], functions=[partial_sum(fare_inr)])
         +- Project [city, fare_inr]
            +- Filter (isnotnull(status) AND (status = COMPLETED))
               +- FileScan parquet [status,fare_inr,city]

Read it bottom-up. The partial_sum before the Exchange is the map-side combine: each task sends one row per city, not one row per ride — 400M rows shrink to a few thousand before crossing the network.

A sort-merge join shows Exchange hashpartitioning(driver_id, 200) + Sort under both sides. A broadcast join shows BroadcastExchange on the small side and nothing above the big side's scan.

Deleting Exchanges

from pyspark.sql.functions import broadcast

# BEFORE: 3 shuffles
out = (rides.join(drivers, 'driver_id')            # SMJ: 2 Exchanges
            .groupBy('driver_id', 'name')             # 0: reuses the join's driver_id partitioning
            .agg(sum('fare_inr').alias('earned'))
            .orderBy(desc('earned')))                 # 1 Exchange (range)

# AFTER: 1 shuffle
earned = (rides.select('driver_id', 'fare_inr')     # prune columns first
               .groupBy('driver_id')                 # 1 Exchange, tiny after partial agg
               .agg(sum('fare_inr').alias('earned')))
out = earned.join(broadcast(drivers.select('driver_id', 'name')), 'driver_id')
# no orderBy: the dashboard sorts 40k rows itself

Three moves: aggregate before joining (shuffle 40k partial rows, not 400M full ones), broadcast the 30MB driver dimension (two Exchanges gone), drop a sort nobody needed (one more gone). The one shuffle left is tiny.

At 100×

Shuffle partition size is the dial. Default spark.sql.shuffle.partitions = 200. A 2TB shuffle into 200 partitions is 10GB per task — guaranteed spill. Target roughly 100–200MB per shuffle partition; AQE (on by default since Spark 3.2) coalesces small ones, but it cannot split upward if you started with too few, so set the initial number high (e.g. 2,000+) for big jobs.

Shuffle disk is local disk. Executors on nodes with small or slow local disks fail with No space left on device mid-shuffle.

Fetch failures (FetchFailedException) mean a reducer could not get a block — usually the executor holding it died. Spark reruns the map stage that produced it, which can turn one lost node into a 30-minute retry.

Skew is a shuffle problem: hashing puts every row of one hot key in one partition. That is its own topic (sp-006).

Gotchas

• repartition(10) before writing to shrink files — a full shuffle when coalesce(10) would merge locally (but beware: coalesce also reduces the parallelism of the work before it in the same stage)
• orderBy before a write nobody reads in order — a range shuffle for nothing
• distinct() on all columns to dedupe — shuffles every column; dedupe on the key with dropDuplicates(['ride_id']) after selecting what you need
• Joining before aggregating — moves 400M rows when 40k would do
• A UDF or cast on the join key — prevents Spark from recognising existing partitioning, adding an Exchange
• Reading the plan top-down — data flows from the bottom (scan) up

Words to know

In two minutes

• A shuffle repartitions data by key: map tasks write bucketed files to local disk, reduce tasks fetch their bucket from every executor.
• Triggers: groupBy/agg, distinct/dropDuplicates, most joins, orderBy, repartition, window functions with partitionBy, intersect/except.
• Not triggers: filter, select, withColumn, union, coalesce, and joins where one side is broadcast.
• In explain(), every shuffle is an Exchange — hashpartitioning (groupBy/join), rangepartitioning (orderBy), RoundRobinPartitioning (repartition(n)). BroadcastExchange is not a shuffle of the big side.
• Spark skips an Exchange when the child is already partitioned the way the next operator needs — so aggregating then joining on the same key shuffles once per side, not twice.
• Avoid shuffles by broadcasting small tables, filtering and pruning columns first, aggregating before joining, reusing existing partitioning, and using coalesce instead of repartition to shrink.
• In the UI, Shuffle Write, Shuffle Read and Spill (memory/disk) tell you how big each shuffle was and whether it fit in memory.

Test yourself

There are 8 questions, 3 puzzles and 5 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.