Spark mechanics a data engineer must know

Lesson 2 of 5 in Transformation Engines.

You will not be asked to write Spark on this exam. You will be asked why a job is slow, and the answer is almost always one of three mechanisms. All three follow from a single distinction.

A narrow transformation — map, filter, select, withColumn — produces each output partition from exactly one input partition. Every row it needs is already on that executor, so Spark fuses a chain of them into a single stage and runs it as one pass per partition, with no coordination between executors at all.

A wide transformation — groupBy, join, distinct, orderBy, a window function with PARTITION BY — produces an output partition from many input partitions. To compute a total per customer, every row for that customer must end up on the same executor, and since those rows began scattered everywhere, Spark has to move them physically.

That movement is the shuffle, and it is worth knowing in mechanical detail because the detail is what the questions are made of. Each task in the upstream stage partitions its output rows by the hash of the shuffle key into as many buckets as there will be reduce partitions — spark.sql.shuffle.partitions, default 200 — and writes those buckets as shuffle files on the executor's local disk, with an index. Each task in the downstream stage then fetches its one bucket from every upstream task.

So the data is serialised, written to disk, transferred, and deserialised, and the number of fetches is writers x readers. Three consequences, and the first is the one people refuse to believe:

Adding executors makes a shuffle wider, not faster. More writers and more readers means more connections for the same bytes.
A shuffle is a stage boundary: nothing downstream starts until the whole exchange completes, so one slow task delays everything.
Shuffle files live on local disk, which is why No space left on device is a shuffle symptom and why the disk column in the worker table is a real constraint.

Three fixes follow from that model, in the order worth trying.

Move less data through the exchange. Filter and project before the wide operation, not after. A groupBy over a table you then filter down to 2% has shuffled fifty times more rows than it needed to, and reordering two lines of code is free.

Avoid the exchange entirely with a broadcast join. A sort-merge join shuffles both sides; if one side is small, sending a complete copy to every executor lets each one join its local partition of the big side in memory. No exchange, no stage boundary, no skew, and the large table never moves. Spark does this automatically when it can estimate the small side under spark.sql.autoBroadcastJoinThreshold, which defaults to 10,485,760 bytes (10 MiB) — and over S3 data with no collected statistics the estimate is often wrong, which is why an explicit broadcast() hint earns its place. The failure mode to respect: the driver collects the small side before distributing it, so broadcasting something that is not actually small gives you a driver out-of-memory error.

Fix the distribution. Data skew is the dominant cause of a slow Spark job in practice and it is invisible in aggregate metrics: total CPU looks fine, the cluster looks busy, and one task is doing almost all the work. In the Spark UI it is unmistakable — the max task duration in a stage is orders of magnitude above the median.

Nulls are the sharpest version. LEFT JOIN ... ON a.customer_id = b.customer_id where a fifth of rows have no customer id puts every one of those nulls in the same reduce partition, building one enormous partition out of rows that will not match anything anyway. Filtering them out before the join is free and is the first thing to try.

A grouped horizontal bar chart comparing task durations within one Spark reduce stage under three key distributions, in illustrative seconds. For an even key: median task 12 s, 95th percentile 15 s, maximum 18 s, so the stage takes about 18 seconds and every core finishes within a narrow band. For a moderately skewed key, such as grouping by country in a business with a dominant home market: median 9 s, 95th percentile 46 s, maximum 210 s — the median has gone down because most partitions now hold less, while the stage has become twelve times longer. For a null-heavy join key where a fifth of rows carry no value: median 8 s, 95th percentile 52 s, maximum 640 s, and the stage is thirty-five times its even-key duration. The figure makes three points. The median improving while the stage degrades is why average or total CPU metrics conceal skew completely. The stage duration tracks the maximum and nothing else, because a stage boundary waits for its last task. And the null case is the worst even though those rows will match nothing in the join, which is why filtering nulls before a join is the cheapest available fix.

A stage with 200 reduce tasks under three key distributions. The stage ends when its last task does, so the only number that matters is the maximum. (illustrative — source: AWS prescriptive guidance — tuning AWS Glue for Apache Spark: optimize shuffles)

The AQE trap: setting shuffle partitions turns off the thing that would have set them

Adaptive query execution re-plans a query mid-flight from real shuffle statistics, and it is on by default in AWS Glue 3.0 and later and in current EMR releases. Three optimisations:

Coalescing shuffle partitions merges contiguous small partitions after a stage completes, removing scheduling overhead. Here is the trap: this is enabled by default unless spark.sql.shuffle.partitions is explicitly set. So hand-tuning the partition count switches off the mechanism that would have tuned it at runtime, and a job "optimised" that way can get slower. If you set it, you own it.

Adaptive join conversion rewrites a sort-merge join as a broadcast hash join when the runtime size of one side comes in under spark.sql.autoBroadcastJoinThreshold — which is how a join over a heavily filtered table avoids a shuffle that the plan-time estimate said was necessary.

Skew join optimisation (spark.sql.adaptive.skewJoin.enabled) detects outlier partitions from shuffle file statistics, splits them into roughly even subpartitions and replicates the matching side. It is the automatic version of salting a key.

The master switch is spark.sql.adaptive.enabled; turning it off disables the framework and all three.

Symptom to mechanism to fix. Every row here is a question shape — the symptom is what the stem gives you, and the mechanism is what the answer has to name.
Symptom in the stem	Mechanism	The fix that works	The fix that does not
199 tasks finish in seconds, one runs for an hour	skew — one key’s rows are in one reduce partition	filter nulls and sentinels first, enable the AQE skew split, salt the key, or broadcast the small side	more executors: the problem is one task, and one task runs on one core
Shuffle Read is far larger than the input	the exchange amplified the data — every row written to disk, fetched, re-read	filter and project before the wide operation; replace a sort-merge join with a broadcast	a bigger cluster: the bytes are unchanged and the connection count grows
`No space left on device`	shuffle files plus spill exhausted the executor’s local disk	more shuffle partitions (smaller pieces) first, then a worker type with more disk	more workers of the same type, which divides nothing per-task
`Container killed by YARN for exceeding memory limits`	one partition is too large for its executor’s memory	raise shuffle partitions; then a larger or R-series worker	raising `spark.executor.memoryOverhead` alone, which treats a sizing problem as a configuration one
Driver `OutOfMemoryError`, executors idle	the driver’s file index, a `collect()`, or a broadcast that was not small	group input files, bound the run, or stop collecting to the driver — see the small-files problem	more executors: the driver is a single process and is not one of them
One task reads the entire input	a single non-splittable object, usually one large `.gz`	many moderate files, or a columnar format whose container splits	more cores, which cannot divide an indivisible stream
Thousands of tasks of a few milliseconds each	too many tiny input files, or too many shuffle partitions for the data	file grouping, compaction, or let AQE coalesce by not pinning the partition count	pinning `spark.sql.shuffle.partitions` higher, which makes it worse and disables coalescing

Key terms: Narrow transformation, Wide transformation, Shuffle, Data skew, Broadcast join, Adaptive query execution (AQE), Executor, Spill