00 · How to read this guide
Localize the bottleneck, then pick the smallest lever
The FDE reflex
Before choosing a lever, localize the bottleneck in the Spark UI. Ask which one it is: CPU-bound compute (tasks busy, cores pinned), shuffle / spill (large Exchange, spill-to-disk metrics), data skew (one task's time and shuffle-read ≫ the median), driver-side (time in planning / a
collect(), not executor compute), or platform (autoscaling lag, cold start, spot reclamation). Only then apply the smallest lever that fits — and remember adding workers only helps CPU-bound, evenly-parallel stages. Verify every change in df.explain() and the Spark UI; never assume.
The four lever families below map to the four sections. Most real calls are choosing which family the symptom belongs to — the wrong lever (usually "add more workers") wastes DBUs and leaves the true bottleneck untouched.
01 · Worker / executor count
Scale out. Helps only parallelizable, CPU-bound work — never a skewed or driver-bound stage.
02 · Memory / node type
Scale up. For OOM, spill, and caching — pick memory- vs compute-optimized to match the pressure.
03 · Shuffle partitions
Right-size the shuffle. Let AQE coalesce and split; avoid hardcoded counts that override it.
04 · Platform choices
Photon, serverless vs classic, Delta layout, Auto Loader / DLT — the no-code and layout wins.
01 · Executor & worker count (scaling out)
When more workers help — and when they do nothing
Scaling out only pays off when work is splittable and CPU-bound. These calls are about spotting when it isn't.
| Scenario | What's really happening | The call (fix) |
|---|---|---|
| 2 TB daily job runs 6h on 20 workers; SLA just dropped to 3h — scale workers, switch node type, or enable Photon? | You don't yet know the bottleneck — adding workers only helps CPU-bound, evenly-parallel stages. | Localize in the Spark UI first; enable Photon (biggest no-code win for scan/join/agg-heavy SQL); raise max workers only if the stage parallelizes; pick compute-optimized nodes if CPU-bound, memory-optimized if spilling. |
| Autoscaling set to 2–20 workers, but the job plateaus at 8 on a skewed shuffle stage. | Autoscaling adds nodes only when tasks are pending; a skewed stage has a few huge tasks, so nothing parallelizes onto new workers. | Fix the skew — AQE skew join is on by default; salt the hot key or repartition (Lesson 08). Workers can't help until the work is splittable. |
| A small 5 GB job lands on a 100-node autoscaling cluster left over from another workload; DBUs burn while nodes sit idle. | The cluster is oversized for the work — parallelism is capped by partitions, not nodes, so most executors never get a task. | Run it on a right-sized job cluster (see cluster sizing); lower the autoscaling max, or use a shared pool so idle capacity isn't billed to this job. |
| Month-end volume is 10× the daily, but the job runs a config cloned from the daily pipeline. | A cluster sized for 1× data under-provisions at 10× — stages queue and spill instead of fanning out. | Parameterize the job's max workers / node type by run date, or use a separate month-end job config; let AQE re-coalesce the larger shuffle rather than hardcoding partitions (see cluster sizing). |
| Enhanced Autoscaling on DLT keeps adding then reclaiming nodes mid-stage — helping or thrashing? | DLT Enhanced Autoscaling scales on backlog and utilization; churn near a stage boundary can look like thrash but is often correct down-scaling. | Read the DLT event log / compute metrics, not node count alone — confirm backlog is draining and no task retries; only pin a floor of min workers if genuine thrash inflates runtime. |
| A serverless SQL warehouse scales out clusters for concurrency, but one monster query still bottlenecks on a single cluster. | Warehouse scaling adds clusters for more concurrent queries; a single query runs within one cluster, so extra clusters don't speed it up. | Size up the warehouse (larger t-shirt = more compute per query) for the heavy query; add clusters only for concurrency. Consider Photon (always on for SQL warehouses) and better data layout. |
| A cluster policy caps max workers at 10 for cost; a legitimately large backfill silently runs under-provisioned. | The policy ceiling is below what the backfill needs, so it never scales into the headroom it would use. | Run the backfill under a policy (or job config) with a higher max-workers ceiling; policies should bound routine jobs, not one-off large loads. |
| Two jobs share an all-purpose cluster; one grabs all executor slots and the other queues behind it. | Executor cores are a shared pool — one greedy job's tasks fill every slot, starving the other (no fair-share by default). | Give each job its own ephemeral job cluster (isolation + cost attribution), or enable scheduler-pool fair scheduling; don't co-tenant unrelated production jobs on one interactive cluster. |
02 · Memory per executor & node type (scaling up)
OOM, spill, and where the memory actually goes
Scaling up is the lever for memory pressure — but only after you know which heap is under pressure (executor vs driver) and why.
| Scenario | What's really happening | The call (fix) |
|---|---|---|
Executors die with ExecutorLostFailure / OOM on a wide aggregation of a 500 GB Delta table — memory-optimized vs general-purpose nodes? |
The shuffle/aggregation buffers can't fit in execution memory and spill, then a partition exceeds the container limit. | First raise parallelism so partitions shrink and let AQE coalesce (Lesson 05); if spill persists, move to memory-optimized nodes and lift executor memory/overhead (Lesson 04). |
A collect() / display() on a "small" result pulls several GB to the driver and it OOMs — worker sizing was never the issue. |
Every partition is serialized into the one fixed driver heap; this is a driver OOM, not an executor one. | Don't collect() large data — write it, or use limit(n)/take(n); size spark.driver.memory/maxResultSize deliberately (Lesson 03). |
| A pipeline runs fine for months, then a skewed key balloons one partition and only that Photon task hits a memory limit and spills. | One oversized partition exceeds what a single task can hold; more or bigger nodes don't help the one hot task. | Fix the skew, don't grow the cluster — AQE skew join splits it (on by default at 5× median & 256 MB); salt the hot key if a groupBy is involved (Lesson 08). |
You picked a 64 GB node, but Photon's off-heap footprint plus shuffle leaves little for caching, so cache() silently evicts. |
Cached blocks live in the storage region and are evictable by execution memory — under pressure they drop and recompute. | Right-size the node for cache + Photon + shuffle together; cache less or use MEMORY_AND_DISK, and cache only DataFrames reused across ≥2 actions (Lesson 06). |
| A broadcast join OOMs after the "dimension" table quietly grew past the broadcast threshold. | The small side is collected to the driver and built there before shipping; once it's genuinely large, the driver OOMs during the build. | Don't force broadcast() on a growing table — let AQE decide at runtime; cap autoBroadcastJoinThreshold (10 MB OSS default, 30 MB Databricks runtime switch) (Lesson 02, Lesson 03). |
| Caching a 300 GB DataFrame on too-small a cluster drops blocks and recomputes the DAG — memory looks fine but the job is slow. | The cache doesn't fit, so blocks evict and every reuse recomputes the lineage; the cost hides as recompute, not OOM. | Only cache what fits and is reused; use MEMORY_AND_DISK to spill instead of recompute, or don't cache and let Photon re-scan Delta (Lesson 06). |
The driver is undersized on a notebook doing heavy pandas / toPandas(), and the whole cluster dies with the driver. |
toPandas() pulls every row into the JVM heap and converts in the driver's Python process — a dead driver kills the application. |
Aggregate / filter / sample before pulling local; size the driver node up for pandas work, or use Arrow-optimized conversion on a bounded result (Lesson 03). |
| Choosing Delta caching (disk) vs Spark caching (memory) on a repeatedly-scanned table changes the memory-pressure profile. | Spark cache() pins the storage region in RAM; the Delta cache keeps decompressed data on local SSD, off the heap. |
For repeated reads of the same Parquet/Delta files, prefer the local-SSD Delta cache (no heap pressure, no eviction of execution memory); reserve Spark cache() for a computed DataFrame reused across actions (Lesson 06). |
03 · Shuffle partition decisions
Let AQE size the shuffle; stop hardcoding 200
Almost every shuffle-partition call comes down to one thing: don't fight AQE. A static count or an upstream repartition quietly defeats its runtime coalescing.
| Scenario | What's really happening | The call (fix) |
|---|---|---|
AQE is on by default, but a hardcoded spark.sql.shuffle.partitions=200 in the cluster config overrides its coalescing on a 3 GB job. |
A static 200 overrides AQE's post-shuffle coalescing, so a tiny job runs 200 near-empty tasks. | Remove the hardcoded value and let AQE coalesce (Lesson 05); on Databricks prefer shuffle.partitions=auto. |
A repartition(500) upstream defeats AQE's post-shuffle coalescing, producing 500 tiny files. |
An explicit repartition(n) forces exactly n partitions; AQE won't coalesce a user-requested count, so tiny files persist. |
Drop the explicit repartition and let AQE size the shuffle to the ~64 MB advisory target; use REBALANCE (AQE-only) if you need even sizing before a write (Lesson 05, Lesson 08). |
| The same shuffle config on a 2 TB join produces 10 GB partitions that spill to local SSD despite Photon. | 200 partitions across 2 TB means ~10 GB each — far above what execution memory holds, so every task spills. | Raise parallelism so partitions land near the 64 MB advisory size and let AQE coalesce from there; Photon speeds compute but doesn't remove an oversized-partition spill (Lesson 05, Lesson 04). |
A Delta write produces hundreds of small files per partition folder; you're running OPTIMIZE or relying on auto-compaction. |
Many small shuffle/output partitions write many small files, degrading downstream read planning. | Enable optimized writes + auto-compaction so Databricks bin-packs at write time; run OPTIMIZE to compact existing files, and let AQE coalesce upstream (Lesson 05). |
| A join between a 1 TB fact and a 2 GB dimension over-partitions the small side because shuffle count is global, not per-side. | shuffle.partitions applies to the whole stage, so the 2 GB side is split into the same many partitions as the fact — lots of near-empty tasks. |
Broadcast the 2 GB side if it fits the threshold so it isn't shuffled at all; otherwise let AQE coalesce the small side's post-shuffle partitions (Lesson 02, Lesson 05). |
| A shuffle count tuned for one source file size breaks when Auto Loader ingests files that doubled after an upstream schema change. | A hand-tuned count is right for one data volume only; when input doubles, fixed partitions become oversized and spill. | Stop hand-tuning — set shuffle.partitions=auto and let AQE adapt to the actual runtime shuffle size each run (Lesson 05). |
| Disabling AQE "to reproduce old behavior" reintroduces the classic 200-partition skew problem on a migrated legacy job. | With AQE off there's no skew-join split or coalescing, so the static 200 partitions surface the original skew straggler. | Keep AQE on (default since Spark 3.2 / DBR 7.3 LTS+); if a legacy behavior is truly needed, fix the underlying skew rather than turning off the engine that repairs it (Lesson 05, Lesson 08). |
04 · Databricks platform & data-layout scenarios
Photon, layout, and the cost/latency tradeoffs
The widest set of day-to-day calls: they're rarely about Spark memory and usually about layout, engine, and cluster shape. Read for the symptom, then the smallest platform lever.
| Scenario | What's really happening | The call (fix) |
|---|---|---|
One customer_id owns 40% of rows; AQE skew join helps, but a groupBy on that key still creates a straggler on one node. |
AQE skew join splits join partitions, but a single-key aggregation still funnels all its rows to one task. | Salt the aggregation — two-stage groupBy(key, salt) then re-aggregate by key — to spread the hot key across tasks (Lesson 08). |
| A query silently falls back to non-Photon for an unsupported expression / UDF, and runtime regresses without an obvious error. | Photon only runs supported operators; an unsupported node makes the engine fall back to Spark for that part, losing the speedup. | Check the query profile / explain for Photon coverage (see Spark UI debugging); replace the Python UDF with a built-in or SQL expression so Photon stays engaged (Lesson 09). |
| A table partitioned on a high-cardinality column creates a small-files problem; migrate to liquid clustering? | Hive-style partitioning on high cardinality makes thousands of tiny directories — read planning and file counts explode. | Yes — move to liquid clustering (cluster by the query keys, no directory explosion), or repartition onto a low-cardinality partition column (Lesson 07). |
| A query filters on a non-clustered, non-partition column, forcing a full scan of a 5 TB Delta table instead of using data skipping. | Data skipping / pruning only works on partition or clustered columns; a filter on anything else can't skip files. | Cluster or Z-order the table on the common filter columns so Delta stats prune files; confirm pruning in the plan / query profile (Lesson 07). |
| A Structured Streaming / DLT micro-batch writes millions of tiny files, degrading downstream reads until auto-compaction catches up. | Each micro-batch commits its own small output files; at high trigger frequency they accumulate faster than compaction runs. | Enable optimized writes + auto-compaction on the sink; tune the trigger / batch size so each batch writes fewer, larger files (Lesson 07). |
Post-OPTIMIZE + ANALYZE, the optimizer flips BroadcastHashJoin ↔ SortMergeJoin and runtime unexpectedly changes. |
Fresh statistics change the estimated table size, moving it across the broadcast threshold and switching the join strategy. | Expected behavior — read explain to confirm the new strategy is right; use a BROADCAST/MERGE hint only if the estimate is genuinely wrong (Lesson 02). |
| Input file volume/shape shifts on an Auto Loader backfill, and static shuffle/partition configs can't adapt to the 100× size change. | A config tuned for steady ingest is wrong at 100× backfill volume — fixed partitions oversize and spill. | Let Auto Loader infer/evolve schema and set shuffle.partitions=auto so AQE adapts per run; cap the backfill batch with cloudFiles.maxBytesPerTrigger (Lesson 05). |
| A bursty, short job vs a steady 6-hour batch — which is cheaper, serverless or classic? | Serverless has fast startup and per-second billing (great for short/bursty); a steady long batch amortizes a right-sized classic cluster better. | Put the bursty job on serverless (no idle cold-cluster cost); run the steady 6-hour batch on a right-sized classic job cluster (see cluster sizing). |
| Spot instance reclamation mid-shuffle causes fetch failures and stage retries, inflating runtime on a "cheap" cluster. | Losing a spot node loses its shuffle files; dependent tasks fail to fetch and the stage retries, erasing the savings. | Keep the driver and a baseline of workers on-demand; use spot only for burst capacity, or move shuffle-heavy stages off spot (see cluster sizing). |
| A pipeline of many small tasks pays repeated cold-start cost on ephemeral job clusters vs a warm pool. | Each ephemeral cluster acquires and boots nodes from scratch; for many small runs, startup dominates the actual work. | Back the job with an instance pool (warm, pre-acquired nodes) or use serverless so startup is near-instant (see cluster sizing). |
| Jobs wait on node acquisition because the pool's min idle instances are set too low for a burst of concurrent jobs. | The pool only keeps a few warm nodes; a concurrency burst exhausts them and later jobs wait on cloud provisioning. | Raise the pool's min idle instances to cover the expected concurrent burst (traded against idle cost), and set a sensible max (see cluster sizing). |
| A repeatedly-scanned table gets evicted from the local-SSD Delta cache under memory pressure, silently re-reading from cloud storage. | The Delta cache is bounded per node; under pressure it evicts, so reads quietly fall back to slower cloud storage. | Use storage-optimized nodes with larger local SSD (bigger Delta cache), or pin the hot dataset; confirm cache hit rate in the Spark UI storage metrics (Lesson 06). |
| A query over a table with tens of thousands of files spends most of its time in metadata/planning on the driver, not executor compute. | Per-file metadata and plan construction happen on the driver; with huge file counts, planning dominates and executors idle. | Compact with OPTIMIZE (fewer, larger files) and cluster/partition to reduce files scanned; this is a driver-side, not a worker, cost (Lesson 03, Lesson 07). |
| A single large gzip file read via Auto Loader lands on one core, ignoring all cluster parallelism. | Gzip is non-splittable, so the whole file is one partition on one task — more workers can't help. | Re-ingest as a splittable format (Parquet, or many smaller files); once split into many partitions, parallelism and workers apply (Lesson 01). |
| Multiple dashboard queries hit one SQL warehouse; queuing appears — scale clusters (concurrency) or size up (single-query speed)? | Queuing is a concurrency problem, not a per-query speed one; a bigger warehouse won't clear a queue of many small queries. | Add clusters (multi-cluster auto-scaling) to absorb concurrent load; size up only if individual queries are themselves slow (Lesson 02). |
| A Python UDF forces row-by-row JVM↔Python serialization, dominating task time versus a native expression. | A plain Python UDF serializes each row across the JVM/Python boundary and blocks Photon — the boundary, not the logic, is the cost. | Replace with a built-in / SQL expression (stays in Photon); if custom logic is unavoidable, use a vectorized pandas UDF to amortize the boundary (Lesson 09). |
| 199 of 200 tasks finish in 30s; one runs 8 minutes on a degraded spot node, dragging out the stage. | A stage ends only when its slowest task finishes; one slow/degraded node becomes a straggler that gates the whole stage. | Enable speculative execution to relaunch stragglers on healthy nodes; if it's data skew rather than a bad node, fix the skew instead (Lesson 08). |
| A shared cluster runs multiple workloads; you can't tell which job drives DBU spend without tags. | Cost is attributed at the cluster level; without job/cluster tags, per-workload DBU spend is invisible. | Apply cluster/job tags (team, pipeline, cost-center) and split unrelated workloads onto their own job clusters so billing is attributable (see cluster sizing). |
| An orchestrated ETL job runs on an expensive all-purpose interactive cluster instead of a cheaper ephemeral job cluster. | All-purpose clusters bill at the higher interactive DBU rate and stay warm; scheduled ETL doesn't need interactivity. | Move the job to an ephemeral job cluster (lower job-compute DBU rate, spins down after) — reserve all-purpose for interactive/dev work (see cluster sizing). |
Files at read (maxFilesPerTrigger / cloudFiles.maxBytesPerTrigger) are mis-sized, causing huge slow batches or tiny inefficient ones. |
The trigger limit sets each micro-batch's input size; too high gives oversized spilling batches, too low gives inefficient tiny ones. | Tune cloudFiles.maxBytesPerTrigger to a batch that lands near the ~64 MB-per-partition target after shuffle; verify batch duration and spill in the streaming metrics (Lesson 05). |
05 · References
Sources
- Apache Spark — SQL Performance Tuning (AQE, coalesce, skew join, broadcast, hints)
- Apache Spark — Tuning guide (memory management, spill, serialization)
- Apache Spark — Configuration (shuffle.partitions, autoBroadcastJoinThreshold, driver/executor memory)
- Azure Databricks — Photon (native engine, fallback behavior)
- Azure Databricks — Adaptive Query Execution
- Azure Databricks — Compute configuration, autoscaling & driver/worker node types
- Azure Databricks — Delta OPTIMIZE, optimized writes & auto-compaction
- Azure Databricks — Liquid clustering & Z-order / data skipping
- Azure Databricks — Auto Loader (schema evolution, maxBytesPerTrigger)
- Azure Databricks — Cluster policies & instance pools