A Spark job that used to finish in ten minutes now takes two hours. How do you find out why?

The action is where you noticed it, not where it happened

Transformations in Spark build a plan and do nothing else. filter, select, join and withColumn return a new DataFrame describing an intention. Only an action - count, write, show, collect, foreach - submits a job, at which point Catalyst optimises the logical plan, produces a physical plan, and the DAG scheduler cuts that plan into stages at every shuffle boundary. Each stage becomes a set of tasks, one per partition, and a stage finishes only when its slowest task finishes.

The practical consequence is that "the count() is slow" is almost never a true statement about count(). It is a statement about everything upstream of it that had never run before. So the first move is not to read the code top to bottom but to open the SQL tab in the Spark UI, find the physical plan for that query, and see which stage consumed the time. df.explain(mode="formatted") gives you the same plan without running it, which is how you check whether your filter got pushed into the scan or is being applied after a join.

Two honest caveats on laziness, because interviewers probe them. Reading with schema inference is not lazy: spark.read.json or a CSV read with inferSchema scans data immediately to work out types, which is why supplying an explicit schema sometimes removes a mystery job. And a DataFrame reused by two actions is recomputed from its source both times unless you cache or persist it, so an action inside a Python loop can multiply the whole lineage by the loop count. That pattern is a common cause of a job that quietly got slower as someone added iterations.

Skew: the cluster is idle and one task is not

If the stage detail page shows a max task duration far above the median while most executors sit idle, you have skew rather than a capacity problem. Partitions are the unit of parallelism, so a single key holding a disproportionate share of the rows means one reduce task does most of the work, and the whole stage waits for it. The UI columns that tell you this are the task duration distribution and shuffle read size per task; the diagnosis is that both are lopsided, not merely large.

Adaptive query execution handles part of this for you. AQE has been on by default since Spark 3.2, and with spark.sql.adaptive.skewJoin.enabled it detects outsized post-shuffle partitions and splits them into smaller ones during a sort-merge join, using statistics collected at runtime rather than the estimates in the original plan. It does not help when the skew is in a groupBy aggregation rather than a join, and it does not help when the skew comes from a single key that cannot be split further because it is one key.

For those cases you salt. Add a random bucket to the skewed key on the large side, replicate the small side once per bucket so the composite keys still match, join on the pair, and aggregate afterwards.

from pyspark.sql import functions as F

# A handful of merchant_ids own most of the rows, so a handful of reduce
# tasks own most of the work.
salts = spark.range(16).withColumnRenamed("id", "salt")

orders_salted = orders.withColumn("salt", (F.rand() * 16).cast("long"))
merchants_salted = merchants.crossJoin(salts)   # 16 copies of the small side

joined = orders_salted.join(merchants_salted, ["merchant_id", "salt"])

The cost is explicit and you should name it: you have multiplied the small side by the salt factor, and you have made the code harder to read for the benefit of one skewed key. Salting is a targeted fix for a measured problem, not a default.

What a shuffle actually costs

A shuffle is where map-side tasks serialise their output, write it to local disk, and reduce-side tasks fetch the blocks they need over the network. So the cost is serialisation plus disk plus network plus the sort or hash on arrival, and it is the only part of a Spark job that is fundamentally not parallel-friendly - every reducer may need bytes from every mapper. That is why plan-level advice ("remove the shuffle") beats tuning advice ("give it more memory") almost every time.

The parameter worth knowing is spark.sql.shuffle.partitions, which defaults to 200 - wrong in both directions depending on your data. With a small result it produces 200 tiny tasks and 200 tiny files; with a large one it produces partitions too big for the executor to sort in memory, which spills to disk. AQE coalesces post-shuffle partitions when it can, removing much of the small-file half, but the target partition size is still worth checking against your volume.

repartition and coalesce are not variations on one idea. repartition performs a full shuffle and gives you evenly-sized partitions; coalesce merges existing partitions without a shuffle, which is cheap, but the reduced partition count propagates upstream into the stage that computes the data, so coalesce(1) before a write can collapse your entire computation to one task.

Broadcast turns a join into a lookup

If one side of a join is small, ship it whole to every executor and the join becomes map-side: each task looks up against its local copy and no shuffle of the large side is needed. Spark does this automatically when it estimates a side under spark.sql.autoBroadcastJoinThreshold, which defaults to 10 MB, and you can force it with a hint when the estimate is wrong - which it often is after a chain of filters the optimiser cannot see through, or when statistics have never been collected.

from pyspark.sql.functions import broadcast

# Forces a broadcast hash join. Correct only if dim_currency really is small.
enriched = orders.join(broadcast(dim_currency), "currency_code")

Two failure modes come with it. The small side is collected to the driver before being distributed, so a "small" table that is not small pressures driver memory or trips the broadcast timeout. And a broadcast copy lives on every executor, so the memory cost is per-executor, not once.

Why collect() on a large result is the mistake it is

collect() fetches every partition into the driver JVM heap as a local array. The cluster's whole point is that no single machine holds the dataset, and collect reverses that in one call. Spark guards it with spark.driver.maxResultSize, which defaults to 1g and aborts the job when the serialised results exceed it - so the visible symptom is often an abort rather than the slowness you expected, or the driver stalling in garbage collection just before it dies.

The correct replacements depend on what you wanted. If you wanted to look at the data, show() or take(n) fetches from the first partitions only. If you wanted a summary, aggregate in the cluster and collect the handful of resulting rows. If you wanted the whole result somewhere, write it to storage and let the consumer read it. And if you genuinely must iterate the full result on the driver, toLocalIterator fetches one partition at a time, trading many small transfers for bounded driver memory.

Where the diagnosis usually goes wrong

The instinct that separates an adequate answer from a strong one is resisting the executor-memory reflex. More memory and more cores make an under-provisioned job faster, and they do nothing at all for a job where one task out of two hundred holds most of the rows - you have simply bought idle capacity for the other 199. Ask for the task duration distribution before asking for hardware. The same mistake in reverse is naming skew immediately without saying how you would confirm it, which reads as a guess; one follow-up about where you saw the number will expose it.

Almost every real answer to "why is this Spark job slow" is a statement about the physical plan - a shuffle that did not need to exist, a partition that is not the size of its siblings, or a result being funnelled through one JVM. Configuration tuning is what you do after the plan is right.