Databricks Spark Developer Associate: Twelve Tasks That Take 55 Minutes: Skew, AQE, and a Tempting Broadcast

Twelve Tasks That Take 55 Minutes: Skew, AQE, and a Tempting Broadcast

A daily query joins orders (2.1 TB, 40,000 files) to customers (6 GB, 41 million rows) on customer_id. The join shuffles into 192 partitions. In stage 14, 180 tasks finish in about 4 minutes each; 12 tasks run for 55 minutes and together spill roughly 300 GB to disk. The single largest customer_id accounts for 9% of order rows. The cluster is 32 workers x 16 cores x 64 GB, spark.executor.memory=20g, memoryFraction at default. Adaptive Query Execution is enabled, but spark.sql.adaptive.skewJoin.enabled was set to false by an earlier engineer. spark.sql.autoBroadcastJoinThreshold is 10 MB. Re-running the job produces the same 12 slow task IDs in the same stage.

1. Is the tail caused by data skew or by uneven partitions/hardware? What does the stability of the slow task IDs tell you, and what would you look at to confirm?

Read the worked answer

Data skew on the join key, not stragglers.

Stable task IDs across runs rules out hardware noise, noisy neighbours and speculative-execution lottery. Random stragglers move around; a deterministic partition-ID set points at a specific key distribution.
Key concentration corroborates it: 9% of a 2.1 TB side implies ~190 GB landing in a single shuffle partition, versus ~11 GB average (2.1 TB / 192). A ~17x imbalance explains a 55-minute task next to 4-minute tasks and 300 GB of aggregate spill.
Confirm in the UI: on the stage's task table, compare shuffle read size and spill (memory/disk) for the max task against the median. If shuffle-read bytes are wildly asymmetric with equal record counts, the skew is in row width; if record counts are asymmetric, it is key skew.
Confirm in data: orders.groupBy('customer_id').count().orderBy(F.desc('count')).show(20) - one key at ~9% is enough to make AQE's default skewedPartitionFactor (5x median) and skewedPartitionThresholdInBytes (256 MB) both trivially satisfied.
Root cause of the non-fix: skew join handling is explicitly disabled, so AQE is enabled but blind to this. With it enabled, AQE would split the oversized partition into sub-partitions - but only for a sort-merge join query stage.

2. A colleague proposes raising `spark.sql.autoBroadcastJoinThreshold` to 8 GB so `customers` is broadcast and the shuffle disappears. Evaluate this.

Read the worked answer

It would remove the skew, and it is still the wrong first move.

Why it is attractive: a broadcast hash join eliminates the shuffle entirely, so there is no 190 GB partition to be skewed - the tail vanishes by construction. That is a real argument, not a naive one.

Why it is risky here:

The driver must build and ship the relation. Collecting ~6 GB (compressed Parquet stats may under-report; the in-memory hash table is typically larger than the file size) risks driver OOM and long GC before any executor does work.
Per-executor footprint. With 64 GB nodes and 20 GB executor heaps, a multi-GB broadcast block plus the build hash table plus an 11 GB sort-merge-side partition competes for the same heap. Broadcast blocks are shared per executor, so you may survive - but you now have almost no headroom for the spill-heavy side.
The threshold compares estimated sizes, not measured ones. Raising it to 8 GB is a global setting that silently changes plans for every other query in the session/warehouse - a blast-radius decision for one query's problem.
Fragility. Any growth of customers turns a working plan into an OOM at 3 a.m., with no graceful degradation.

Verdict: acceptable as an experiment behind a query-scoped hint (F.broadcast(customers)) with raised executor memory, but not by globally raising the threshold. The cheap, reversible fix comes first.

3. Give a fix that neither broadcasts the 6 GB table nor abandons AQE, with rough cost arithmetic, and say when you would escalate to manual salting.

Read the worked answer

Step 1 - enable what is already paid for. Set spark.sql.adaptive.skewJoin.enabled=true. AQE will split any partition above both skewedPartitionFactor x median and skewedPartitionThresholdInBytes, turning a 190 GB partition into N sub-partitions. Cost: none beyond an extra shuffle-map read. Prerequisite to verify: the join must resolve to a sort-merge join (AQE cannot split a broadcast hash join), and the skew must appear in a shuffle query stage, which it does.

Step 2 - if the tail persists, isolate the hot key.

hot    = orders.filter(col('customer_id') == hot_id)
rest   = orders.filter(col('customer_id') != hot_id)
result = rest.join(customers, 'customer_id') \
             .unionByName(hot.join(F.broadcast(customers.filter(col('customer_id') == hot_id)), 'customer_id'))

Cost: one extra pass over orders (cheap, both filters are pushed down), and the hot branch broadcasts a 1-row relation. The 190 GB partition is dissolved.

Step 3 - salting, when the hot set is not a single key. Add salt = pmod(hash(...), N) to the big side and explode the small side over 0..N-1, joining on (customer_id, salt). Cost: the small side is replicated N times into the shuffle - 6 GB x 10 = 60 GB of extra shuffle traffic and 10x more partitions on one input - which is a good trade against a 190 GB tail only if the hot key(s) genuinely cannot be isolated. If you salt, also raise spark.sql.shuffle.partitions so the salted partitions are not artificially huge.

Escalation rule: enable AQE skew handling -> isolate hot keys -> salt only when the skew is diffuse across many keys.