Spark shuffle and skew on the DE interview

What shuffle actually is

If you have ever watched a Spark job sit at 95% complete for forty minutes while one executor refuses to die, you have met the two concepts this post is about: shuffle and skew. They are also the single most common Spark topic on Data Engineering interviews at companies like Databricks, Snowflake, Netflix, and Stripe.

A shuffle moves data across the network to regroup it by a new key. Every node has some rows, the operation needs all rows for a given key in the same place, so Spark redistributes everything. It is the most expensive thing the engine does — on real workloads, shuffle accounts for 70 to 90 percent of total job time.

The mechanics, in three stages:

1. Map side. Each task on an input partition hashes the keys and writes the resulting rows to local disk, pre-bucketed by target partition.
2. Network. Reduce tasks pull their assigned buckets from every executor that produced data. This is usually the bottleneck.
3. Reduce side. The reduce task reads its data, merges, and optionally sorts.

When shuffle happens

A useful mental model: any operation that needs to compare rows across partitions causes a shuffle. Anything that processes rows within a partition does not. Interviewers love asking you to classify operations on the spot.

The principle that falls out: filter early, project early, shuffle late. A repartition followed by a filter shuffles ten times more data than the reverse.

# bad: shuffle before filter
df.repartition(200, 'user_id').filter('event_date = "2026-05-01"')

# good: filter shrinks the dataset before the shuffle pays for it
df.filter('event_date = "2026-05-01"').repartition(200, 'user_id')

The same rule applies to projections — drop columns you do not need before any wide transformation.

Skew and how to spot it

Skew is what happens when shuffle does its job correctly but the data itself is lopsided. One partition ends up with 90% of the rows, the other 199 partitions get the remaining 10%, and your 200-way parallelism collapses into a single straggler task.

Skew comes from the business domain. One user_id belongs to a scraping bot and has 10 million events; the median user has thirty. One merchant_id is Amazon and owns half the orders on a marketplace. After hash partitioning, all rows for that hot key collide into one partition, and one task does all the work while the cluster sits idle.

Diagnostic signs you should be able to list from memory:

• Spark UI → Stages tab: one task runs for ten minutes while the rest finish in seconds. That is skew, full stop.
• Stage details → Shuffle Read Size: one partition is 50 GB, the rest are 100 MB. Same diagnosis.
• Executor metrics: memory pressure or OOM concentrated on one executor.
• Scaling test: adding executors does not reduce wall-clock time. The bottleneck is one task, and one task cannot be parallelized further without changing the data.

The classic interview prompt is some variant of: "You join two tables, one side is small, the job is still slow. Why?" The expected answer is skew on the join key — one key has millions of rows on the big side, after shuffle they all land in one partition, and one task carries the whole join.

Salting to fix skew

Salting is the manual fix: you append a random suffix to the hot key so the rows distribute across N partitions instead of one. The small side of the join is then duplicated N times so that every salted variant of the key still finds its match.

from pyspark.sql.functions import concat, lit, rand, floor, col
import pyspark.sql.functions as F

N_SALTS = 10

# big side: add a random salt suffix to user_id
events_salted = events.withColumn(
    'user_salt',
    concat(col('user_id'), lit('_'), floor(rand() * N_SALTS).cast('int'))
)

# small side: duplicate every row N times, one per salt value
salts = spark.range(N_SALTS).withColumnRenamed('id', 'salt_n')
users_salted = (
    users
    .crossJoin(salts)
    .withColumn('user_salt', concat(col('user_id'), lit('_'), col('salt_n')))
)

joined = events_salted.join(users_salted, 'user_salt')

The cost is real: the small side balloons by N×. If N=10 and users is 1 GB, your salted version is 10 GB on every executor's working set. The join itself parallelizes properly, but you spent memory to get there.

A smarter variant, sometimes called adaptive salting or selective salting, only salts the hot keys. You first compute the top-100 keys by row count, salt only those, do an ordinary join for everything else, and UNION the results. This keeps the small-side blowup proportional to the number of hot keys, not the whole table.

AQE: automatic skew handling

Adaptive Query Execution (AQE), generally available since Spark 3.0 and on by default since 3.2, watches the actual shuffle output and re-plans the query mid-flight. For shuffle and skew, three features matter:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

Skew join handling. AQE identifies skewed partitions by comparing each partition's size to the median (default: a partition is "skewed" if it is 5× the median and larger than 256 MB). It splits those partitions into smaller sub-partitions and duplicates the matching rows from the other side — the same idea as manual salting, executed automatically.

Coalesce shuffle partitions. After shuffle, AQE merges tiny partitions so you do not launch a task for every 1 MB chunk. The default spark.sql.shuffle.partitions=200 becomes much more forgiving — you can set it high and let AQE collapse downward.

Dynamic broadcast join. If a filter or aggregation shrinks one side of a join below the broadcast threshold at runtime, AQE will switch a planned SortMergeJoin to a BroadcastHashJoin without you touching the code.

AQE is not a silver bullet — on a single terabyte-scale hot key it will still struggle, because splitting one mega-partition into ten 100 GB partitions is still a bad situation. For genuine pathological skew, manual salting on top of AQE is the answer.

Broadcast joins

A BroadcastHashJoin sends the entire small side to every executor and joins in memory without shuffling the big side. Zero shuffle on the big table — which, when applicable, is the cheapest join Spark can do.

from pyspark.sql.functions import broadcast

result = big_df.join(broadcast(small_df), on='user_id', how='left')

When to reach for it:

• Small side fits comfortably in executor memory. Under 100 MB is safe; up to 1 GB works with the right spark.driver.memory and spark.sql.autoBroadcastJoinThreshold.
• Star-schema joins of a fact table against several dimension tables.
• After a heavy filter that you know will shrink one side, but AQE has not picked it up.

The auto-broadcast threshold is 10 MB by default — almost always too low. Raising it is a cheap win:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100m")

Limits to remember. Both sides must fit on every executor, not just the driver. Outer joins where the large side dominates with NULLs do not benefit. And broadcasting something genuinely large OOMs the driver during the collect step before the join ever starts.

Common pitfalls

When a candidate gets these wrong, it usually means they have copied Spark code without ever opening the UI. The interviewer will dig in immediately.

coalesce(1) to write a single output file. The most common foot-gun in production code. coalesce(1) forces the entire dataframe through a single task no matter how large, because coalesce does not shuffle and cannot redistribute work. On anything over a few GB you OOM the lone executor or wait hours. Either write N files and merge downstream with a sane coalesce(N), or write partitioned and let a separate process compact.

repartition with the default partition count. df.repartition(col) uses spark.sql.shuffle.partitions, default 200. If post-shuffle data is 100 GB, that is 200 partitions of 500 MB each — too coarse for a cluster with hundreds of cores. Rule of thumb: total_shuffle_size / 128MB.

Ignoring AQE because the codebase predates Spark 3. AQE typically delivers a 20 to 50 percent speedup on shuffle-heavy workloads, for the cost of three config lines. No good reason to leave it off in 2026 unless you are pinned to Spark 2.x — and if you are, that is the bigger problem.

Salting blindly. Picking N out of the air is worse than not salting. N=2 on a 1000× skew does almost nothing; N=100 on uniform data wastes 100× the small-side memory. Always profile the hot keys first, then size N to the worst offender.

Hash join that spills to disk. When shuffle memory is exhausted, Spark spills to a temporary file. On a skewed key that spill can be several GB on one executor, and disk-bound hash join is catastrophically slow. Symptom: large Spill (Memory) and Spill (Disk) in the Stage UI. Fix: salt the key, or raise spark.executor.memory if you genuinely need more headroom.

Filtering after the join. df1.join(df2).filter(...) joins the whole world and filters the result. Push filters into each side before the join. With AQE this matters less than it used to — Catalyst pushes filters down well — but it cannot push through a UDF or a join condition that depends on both sides.

Not opening the Spark UI. Every diagnosis above is visible in the UI. Engineers who tune Spark by editing configs without reading the UI are guessing. On the interview, mentioning specific UI panels (Stages, SQL, Executors) signals that you have shipped Spark code, not just read about it.

Related reading

• Spark fundamentals — DE interview deep dive
• Spark broadcast joins — DE interview deep dive
• Spark Catalyst and AQE — DE interview deep dive
• Database sharding for Data Engineers

If you want to drill Spark and Data Engineering questions like this every day, NAILDD is launching with 500+ interview problems across exactly this pattern.

FAQ

How many shuffle partitions should I set?

The default of 200 is rarely optimal. A working heuristic is total_shuffle_size / 128MB, so a job that shuffles 100 GB wants roughly 800 partitions. With AQE coalesce enabled you can set the number higher than you think and let Spark merge the small ones — safer than guessing low.

repartition or coalesce — when to use which?

repartition(N) performs a full shuffle and can grow or shrink partitions, redistributing rows evenly. coalesce(N) only shrinks, does not shuffle, and is much cheaper — but it cannot fix skew because it just merges adjacent partitions as-is. Use coalesce to reduce output file count at the end of a job; use repartition when you need an even distribution by key.

What is spill and when should I worry about it?

Spill is what happens when shuffle data does not fit in the executor's shuffle memory and Spark writes the overflow to local disk. You will see it in the Stage UI as Spill (Memory) and Spill (Disk) columns. A little is fine on big jobs. Hundreds of MB per task, concentrated on a few executors, means either insufficient memory or skew, and it slows the stage by an order of magnitude.

Can I avoid shuffle entirely?

Not if your operation requires regrouping by a new key. You can avoid it for joins by broadcasting the small side, and across multiple jobs by writing data already partitioned by the join key — the next read joins without shuffle. Bucketing tables on write does the same for Hive-style storage.

What is a bucket join?

If two tables are written with identical bucketing on the join key (df.write.bucketBy(N, 'user_id').saveAsTable(...)), Spark knows the rows for a given key live in matching bucket files on both sides and joins without shuffling. Common in Hive and managed lakehouse tables, less so on raw Parquet on S3 because Spark cannot trust the file layout the same way.

Is this official documentation?

No. This post draws on the Apache Spark 3.x docs, Databricks talks, and practitioner scar tissue. Specific tuning numbers depend on cluster shape, Spark version, and storage backend — starting points to measure against, not gospel.