DE Learning

Spark Core · Module 5 of 5

Finding and fixing slow Spark jobs

Interviewers rarely ask "what is skew?" on its own. They describe a slow or failing job and watch how you investigate. Learn the flow first, then the six scenarios it leads to.

Debug flowSix scenariosReducing shufflerepartition vs coalescePartition sizingExecutor sizingCache vs persistCheck yourself

The debug flow

Start in the Spark UI: Jobs → the slowest job → Stages → the slowest stage → its summary metrics (min / 25th / median / 75th / max per task). Most answers come from comparing median and max, and from the spill and shuffle columns. Click an outcome to open that scenario.

Decision flow for diagnosing a slow or failing Spark job Job is slow, or fails Did tasks fail with OutOfMemory or "container killed"? In the slowest stage, is max task time ≫ median? Does the stage show Spill (memory / disk)? Thousands of very short tasks, or thousands of files? Large shuffle read/write, or a big table in SortMergeJoin? Check GC time, Python UDFs, source I/O, cluster size nonononono yes yes yes yes yes OOM: executor or driver → Data skew → Disk spill → Small files / too many partitions → Heavy shuffle / slow join →

Where each signal lives in the Spark UI

SignalWhere to look
Which job/stage is slowJobs and Stages tabs, sort by duration
SkewStage → Summary Metrics: Duration and Shuffle Read Size, max vs median; Event Timeline with one long bar
SpillStage → Spill (Memory) and Spill (Disk) columns
GC pressureStage task table → GC Time; Executors tab → GC time vs task time (over ~10% is a warning)
FailuresStage → failed tasks → error column; Executors tab → dead executors and logs
Final plan, rows per operatorSQL / DataFrame tab: the plan after AQE, with row counts and sizes per node

Six scenarios

Each one: what you see, why it happens, how to confirm it, how to fix it, and a short interview answer. Metrics are representative examples.

Reducing shuffle: the checklist

  1. Read less. Filter and select columns as early as possible; make sure filters reach the scan (PushedFilters / PartitionFilters).
  2. Broadcast small sides. Raise the threshold or add a hint for dimension tables up to a few hundred MB.
  3. Pre-aggregate before joining. Aggregate the fact to the grain you need, then join the smaller result.
  4. Remove accidental shuffles. Unneeded distinct, orderBy before a write, repartition before a join that reshuffles anyway, window functions with a different partitionBy each time.
  5. Shuffle once, reuse. Several joins or aggregations on the same key in a row: Spark reuses an existing partitioning on that key. Bucketing (or co-located clustering) avoids the shuffle for repeated joins on large tables.
  6. Right-size what's left. Let AQE coalesce, or set spark.sql.shuffle.partitions so each shuffle partition is roughly 100–200 MB.

repartition vs coalesce

repartition shuffles into even partitions; coalesce merges neighbouring partitions without a shuffle repartition(3) · full shuffle even sizes, can go up or down, costs network coalesce(2) · no shuffle merges neighbours, cheap, sizes stay uneven uneven output
repartition(n) / repartition(col)coalesce(n)
ShuffleYes (wide)No (narrow)
DirectionIncrease or decreaseDecrease only
ResultEvenly sized; by column = same key in same partitionUneven if inputs were uneven
Use forFixing skew/too few partitions, increasing parallelism, partitioning before a partitionBy writeCutting the number of output files cheaply after a filter
The coalesce(1) trap
Because coalesce adds no stage boundary, Spark pushes the reduced parallelism upstream: df.filter(...).groupBy(...).agg(...).coalesce(1).write can run the whole final stage in a single task. If upstream work is heavy, repartition(1) is faster: the heavy work stays parallel and only the final write is single-threaded.

Partition sizing

Rule of thumb: aim for roughly 100–200 MB per partition in memory and at least 2–3 tasks per core.

Executor sizing

Classic YARN/Kubernetes approach: about 5 cores per executor (good I/O throughput without too many threads sharing a heap), one core and ~1 GB per node left for the OS and daemons, and ~10% of memory for overhead.

One executor slot is subtracted for the YARN application master in cluster mode. On Databricks there is one executor per worker, so you size by instance type and worker count; the same per-core memory reasoning applies.

How to choose, in an interview

  • Too-fat executors (1 executor × 15 cores × 60 GB): long GC pauses, HDFS/S3 client contention.
  • Too-thin executors (1 core each): no shared broadcast variables or cache, more JVM overhead, more executors to manage.
  • Memory-hungry stages (big joins, wide rows): fewer cores per executor or memory-optimised instances, so each task gets more memory.
  • CPU-bound stages (parsing, UDFs, compression): compute-optimised instances, more cores.
  • Variable load: dynamic allocation or autoscaling with sensible min/max.

Cache vs persist

  • cache() is persist() with the default level: memory-only for RDDs, memory-and-disk for DataFrames.
  • persist(StorageLevel.X) lets you choose: MEMORY_ONLY, MEMORY_AND_DISK, DISK_ONLY, serialised variants, and _2 replicated variants.
  • Caching is lazy: nothing is stored until an action runs. Check the Storage tab to see what's cached and what fraction fits.
  • Cache when a DataFrame is reused by several actions and is expensive to recompute. Don't cache something used once or cheap to re-read from columnar files.
  • Call unpersist() when done; cached data competes with execution memory.
  • Alternatives: checkpoint() or writing an intermediate table to truncate very long lineage in iterative jobs; on Databricks the disk cache speeds up repeated Parquet/Delta reads automatically.

Check yourself

"Our nightly job went from 1 hour to 4 hours. Nothing changed in the code." Walk me through it.
Compare the job in the Spark UI with a previous good run. Find which stage grew. Check input volume (did the source grow or a filter stop pruning?), task-time distribution (new hot key → skew), spill (data outgrew memory per task), file counts (small files accumulating), and the join strategy (a dimension crossed the broadcast threshold and became a sort-merge join). Fix the specific cause, then add monitoring on input size and stage duration.
199 tasks finish in 30 s, one takes 45 min. Fix it in three ways.
AQE skew join (if it's a sort-merge join and AQE is on), broadcast the smaller side, or salt the hot key. Also check whether the hot key is NULL or a default value that should be filtered out.
When would you increase spark.sql.shuffle.partitions, and when decrease it?
Increase when post-shuffle partitions are large (spill, OOM, long tasks); decrease when they're tiny (many short tasks, small output files). With AQE, set it high-ish and let coalescing reduce it.
Is spill always bad?
No. Spill means Spark survived memory pressure instead of failing. A little is fine; spill that is a large multiple of shuffle read, or that appears on every task, means partitions are too big for the memory per task.
Your write produces 20,000 small files. Options?
Repartition or coalesce before the write (by the partition column if using partitionBy), enable optimised writes or auto-compaction on Delta, run OPTIMIZE, and avoid partitioning on high-cardinality columns.

Mark this module's topics