Apache Spark Debug & Fix

The reasoning

A shuffle assigns rows to partitions by hashing the join key. Every row with the same key lands in the same partition, so if one key holds 80% of the rows, one partition holds 80% of the data regardless of how many partitions you configure. That partition is one task, and one task runs on one core.

The fingerprint in the Spark UI is unmistakable once you know it: open the stage, sort tasks by duration, and look at the summary metrics. Skew shows as a max shuffle-read size and a max duration that are orders of magnitude above the median — not a long tail, a single outlier. If instead every task is slow, that is a sizing problem and a completely different investigation.

The plan below shows the sort-merge join that produced it: both sides hash-partitioned on account_id, both sorted, and the merge happening per partition. Nothing in the plan is wrong — the plan is correct and the data is uneven, which is why this is a data problem wearing a performance costume and why reading the plan alone will never reveal it.

The reasoning

The mechanism is to break one hot key into N. Adding a random salt in 0..N-1 to each event's account_id turns acct_0 into acct_0#0 through acct_0#7, which hash to eight different partitions — so the 80% is spread across eight tasks instead of one.

The half people forget is that the other side has to match all of them. If events carry acct_0#3, the accounts row for acct_0 must also exist as acct_0#3 or the join loses it. So the small side is exploded across every salt value, which multiplies it by N. That is affordable for a dimension and it is why salting is only viable when one side is small — at which point broadcasting was often available anyway, which is exactly why this is the third option rather than the first.

The costs are worth naming plainly. The small side is N times bigger. The join key is now a synthetic string nobody reading the code will understand without the comment. If you aggregate afterwards you have to strip the salt and re-aggregate. And N is a tuning parameter that has to be revisited as the skew changes. It works and it is real complexity, which is why AQE's automatic skew handling is the better answer whenever it suffices.

The reasoning

An inner join keeps only matched rows. A left join keeps every row from the left and fills the right's columns with null where there is no match. Swapping one for the other is a one-word edit that changes the semantics of the result, and neither version errors.

The two runs below return different rows, which the harness asserts — that is the whole point of publishing the pair. The broken version simply has fewer rows, and the only signal available is the count itself. If nothing is watching the count, this ships.

There is a second, subtler version of the same bug worth raising: a left join followed by a filter on a right-side column silently becomes an inner join, because the null rows fail the predicate. WHERE tier = 'pro' after a left join removes every unmatched row exactly as an inner join would. The fix is to put the condition in the join's ON clause rather than in the filter, which keeps the preservation semantics intact — and it is worth checking for whenever a left join's row count looks like an inner join's.

The reasoning

A join emits one row per matching pair. If products has two rows for a product_id, every order for that product produces two rows, and a sum over amount counts it twice. Nothing is wrong with the join or with Spark; the dimension stopped being unique on its key.

The diagnostic is one query and worth having as a reflex: compare count(*) against count(distinct product_id) on the dimension. If they differ, any join to it will fan out. Running that as an assertion in the pipeline — rather than as an investigation after finance calls — is what makes this bug a one-time event instead of a recurring one.

The fix in the code below is dropDuplicates on the key, and it is worth being honest that it is a patch rather than a repair. It picks an arbitrary row among the duplicates, which is only safe when they are genuinely identical. If they differ, you have a modelling problem — the grain of the dimension is not what you thought — and choosing which row survives is a business decision rather than a technical one.

The reasoning

Both versions produce one output partition and both have one Exchange, so a shuffle count tells you nothing here — which is a useful lesson in itself. What differs is *where* the narrowing sits relative to the work.

repartition(1) before the groupBy sends all 2,000 rows into a single partition and the aggregation then runs with one task. On real data that is the entire dataset through one core, and adding executors does nothing at all. coalesce(1) after the aggregation lets the grouping run across all partitions and merges the three resulting rows, which costs nothing.

The general rule is that narrowing belongs after the expensive work, and the specific trap is that coalesce propagates upstream — coalesce(1) placed *before* expensive work has the same effect as repartition(1), silently, because it introduces no stage boundary. If you must reduce to one partition before something expensive, repartition is the one to use, precisely because its shuffle creates the boundary that keeps the upstream parallel.

The reasoning

Driver OOM comes from data arriving at the driver: collect() or toPandas() on something large, a broadcast whose build side exceeded the estimate, or an enormous plan built by thousands of iterative transformations. The fix is to stop pulling data — write to storage and read it back, or aggregate before collecting — not to raise driver memory, which only moves the threshold.

Executor OOM means a single task could not fit its partition's working set. That is almost always skew: 199 partitions fit and one did not. It can also be a wide aggregation with high cardinality, or an explode multiplying rows inside a task. The fix is to make partitions smaller or more even, not to buy bigger executors — raising memory to accommodate the largest partition means paying for it on every executor for the sake of one task.

The reason the distinction matters so much is that the instinctive fix — raise spark.executor.memory — helps neither case properly. For a driver OOM it is the wrong process entirely. For an executor OOM it treats the symptom and leaves you with an expensive cluster that fails again when the skew worsens. Reading which process threw is thirty seconds of work and it redirects the entire investigation.

The reasoning

When a task's working set outgrows its share of execution memory, Spark writes part of it to local disk and reads it back — that is spill, and it is the graceful degradation that stops an OOM. Succeeding with spill is better than failing without it, and it is much slower than not spilling.

The two columns measure the same event twice. Spill (memory) is how large that data was in memory before it was written; spill (disk) is how many bytes hit the disk after serialisation and compression. The ratio is a compression ratio, not two separate spills, and people frequently read the pair as if 52 GB moved.

Fixing it means either giving tasks less to hold or more room to hold it, and less is usually right. More partitions makes each task's share smaller; fixing skew makes them even; and aggregating or projecting earlier means less is carried into the operator that spilled. Raising spark.memory.fraction is the lever people reach for and it steals from storage memory, so it makes caching worse to make execution better — occasionally correct, rarely the first move.

The reasoning

A broadcast is collect-then-distribute: the driver pulls the whole build side, builds a relation from it, and ships it to every executor. The timeout covers that whole sequence, so exceeding it means the data is big, the driver is slow, or the cluster is wide enough that distribution takes real time.

The usual root cause is a size estimate that was wrong. Without table statistics Spark estimates from compressed Parquet file sizes, which systematically under-reports the in-memory size — a table that looked like 8 MB can be several times that once deserialised. It is also common after a chain of filters and joins, where the optimizer's estimate of an intermediate has drifted far from reality.

Three options, in order. Stop broadcasting it: remove the hint or lower the threshold, and accept a sort-merge join, which is slower per query and does not fail. Make the estimate right: run ANALYZE TABLE so the decision is based on real statistics rather than file size. Or enable AQE, which decides from the *actual* shuffled size at runtime and will simply not attempt a broadcast that does not fit. Raising spark.sql.broadcastTimeout is fourth, because it gives a too-large broadcast more time to exhaust the driver.

The reasoning

The structured API keeps rows in Tungsten's off-heap-style binary format precisely to avoid this: no per-row Java objects means very little for the collector to trace. So high GC time usually means something has put ordinary objects back on the heap — an RDD of case classes or tuples, a UDF materialising collections per row, or a cache using a deserialised storage level.

Executor size is the second cause and it is structural. A very large heap takes longer to collect, so 64 GB executors can show worse pause behaviour than several 16 GB ones even at the same total memory. That is one of the concrete reasons for the conventional advice of moderate executors rather than a few enormous ones.

The fixes follow the causes: stay in the DataFrame API so rows stay binary, use MEMORY_ONLY_SER or MEMORY_AND_DISK_SER when caching large data so cached blocks are one byte array rather than millions of objects, and prefer moderate executor sizes. Tuning the collector itself — switching to G1, adjusting region sizes — is a real lever and it is the last one, because it manages the symptom rather than reducing the allocation.

The reasoning

A Python UDF is registered with a Spark type, and Spark trusts that declaration. If the function returns a Python float and the UDF declares IntegerType, or returns a string where a DoubleType was declared, the conversion fails and the result is null rather than an exception. Every row, silently.

The reason it is so confusing is the asymmetry: the function is correct, the test in Python passes, and the failure only exists at the boundary. numpy types are a frequent culprit — a numpy.int64 is not a Python int, and the conversion can produce null depending on the declared type. Returning a dict where a StructType was declared with different field names does the same thing.

The way to find it is to check the declared type against the function's actual return with a single row, and the way to avoid it is to not be there — a native expression has no declaration to get wrong, and a pandas UDF with type hints makes the mismatch far more visible. When a UDF is genuinely necessary, being explicit about the return type and converting inside the function is what keeps it honest.

The reasoning

Spark's TimestampType is an instant, and rendering it as a date requires a timezone. That timezone is spark.sql.session.timeZone, which defaults to the JVM's — so the same job produces different daily buckets on a laptop in Europe and a cluster in UTC. Only rows within the offset of midnight move, which is why the symptom looks so specific.

The fix is to set the session timezone explicitly rather than inheriting it. Being explicit means the job produces the same answer everywhere, and it forces the question the bug was really about: which timezone does the business mean by 'day'? For a global product that is usually UTC; for a retailer it may be store-local, which is a per-row conversion rather than a session setting.

The related trap is TimestampNTZ and string timestamps. A timestamp stored as a string and cast on read is interpreted in the session timezone too, so the same shift applies at parse time. And if the source wrote local times with no offset, there is no timezone information to recover — the correct handling has to be agreed with the producer rather than guessed, which makes it a data-contract question as much as a Spark one.

The reasoning

Ties in an ordering are the most common. row_number over a window whose ORDER BY does not uniquely determine an order picks arbitrarily among tied rows, and which one wins depends on partitioning and task completion. Adding a unique tiebreaker makes it reproducible, and it is the fix most top-N code is missing.

Functions that are non-deterministic by construction are the second: rand() without a seed, and monotonically_increasing_id, whose values depend on partition ids and therefore change if the partitioning changes. Using the latter as a stable key is a specific and common mistake — it is unique within a run and not stable across runs.

Third is any operator that picks arbitrarily: dropDuplicates on a subset of columns, first() and last() in an aggregation without an ordering. All of them return *a* row rather than a defined one. And the reason all three tend to appear late is that they are usually stable on a single-partition test dataset and only diverge once the data is large enough to be partitioned differently between runs.

The reasoning

partitionBy chooses the directory layout and has no opinion about file count. Every in-memory partition that contains rows for a date opens a file in that date's directory, so the upper bound is tasks multiplied by distinct values — and with a default of 200 shuffle partitions that number is large before anyone has done anything unusual.

The fix is to align the in-memory partitioning with the write layout: repartition on the same column before writing, so all rows for a date live in one task and one file is produced. The shuffle that costs is far cheaper than the read-side penalty of thousands of small files, which is paid by every consumer on every query.

If one date is too large for a single file, repartition on the date plus a deterministic bucket so it splits a controlled number of ways rather than however many the previous stage happened to leave. And if the table is Delta or Iceberg, a scheduled OPTIMIZE handles compaction after the fact — which is the right answer when you cannot control the writer, and not a substitute for writing sensibly when you can.

The reasoning

The number of tasks in the first stage comes from the number of input splits. For Parquet that is driven by file count and file size against spark.sql.files.maxPartitionBytes — so two large files give two splits, two tasks and two busy cores while thirty-eight idle.

A large *splittable* file can be divided, so the picture depends on the format. Parquet is splittable by row group; a gzipped CSV is not splittable at all, which means one 20 GB .csv.gz is exactly one task no matter what you do. That is the single strongest practical argument against gzip for large inputs, and it surprises people who assume compression is only a size decision.

The right fix is at the source: write more, appropriately-sized files, or use a splittable format and codec. If you cannot change the producer, lowering maxPartitionBytes produces more splits from the same files, and repartition after the read does work — at the cost of a full shuffle to fix a layout problem, which is why it is the last option rather than the first.

The reasoning

Shuffle output is written to the local disk of the executor that produced it. If that executor dies — OOM, spot reclamation, node loss — its shuffle files become unreachable, and the downstream task reports FetchFailedException. The error surfaces in the *reading* stage while the cause is in the *writing* one, which is why people investigate the wrong place.

Spark's response is to mark the map output as lost and re-run the upstream tasks that produced it. That is why you see a stage retry rather than a straightforward failure. If the reason the executor died is still present — a partition that does not fit, a node under memory pressure, aggressive spot reclamation — the recomputation dies the same way and the cycle repeats until the retry limit.

So treat it as a symptom and look upstream. Check for executor OOM or lost workers around the same time; the real fix is whatever killed the executor. And enable the external shuffle service, which holds shuffle files outside the executor process so losing one no longer loses its output — that alone turns many of these from a cascading failure into a rescheduled task, and it is close to mandatory on spot capacity.

The reasoning

AQE's skew handling targets sort-merge joins specifically: it detects oversized partitions after a shuffle and splits them across tasks, replicating the matching side. If your skew is in a groupBy, a window partitionBy, or a distinct, none of that machinery applies and the hot partition is processed by one task exactly as before.

The thresholds are the second reason. A partition has to exceed skewedPartitionThresholdInBytes *and* be several times the median partition size before it is treated as skewed. A stage where every partition is large has no outlier relative to the median, so nothing triggers even though every task is slow — and that is a sizing problem rather than a skew one, which is worth diagnosing differently.

Two more that catch people: AQE needs a shuffle to observe, so a skewed stage with no Exchange gives it nothing to work with; and the split relies on statistics from the completed map stage, so a very small number of partitions can leave it too little to act on. When AQE genuinely does not apply, you are back to the manual options — broadcast if the other side fits, salt if it does not, or change the key so the aggregation is not on the skewed column.

The reasoning

Start at Jobs to see which action is slow — a script with four writes has four jobs, and narrowing to one immediately halves the search. Then Stages, sorted by duration, to find the stage inside it. Everything before this is triage.

The stage detail page is where the diagnosis happens, and the summary metrics table is the part to read. Compare the max against the median for duration, shuffle read size and spill: a max far above the median is skew; every task slow with high shuffle read is an undersized stage or too much data; high spill in most tasks is a memory problem; high GC time points at object churn. Those four readings cover most incidents and take about a minute.

Then the SQL tab, which shows the query plan annotated with actual row counts and the AQE final plan. That is where you find out that the optimizer estimated a thousand rows and got ten million, or that the broadcast you expected never happened. Going there first is tempting and usually wasteful — the plan tells you what Spark intended, and the stage metrics tell you what actually hurt.

The reasoning

explode is a narrow transformation, which is exactly the problem here. There is no shuffle, so the output rows for a given input row all stay in the task that held it. One record with a million-element array becomes a million rows in one partition, and no partition count anywhere changes that.

The average being four is what makes it hard to spot. Array length distributions in event data are usually long-tailed — a bot session, a bulk import, a retry loop that appended repeatedly — and the maximum is what determines whether a task survives. Checking max(size(col)) rather than avg is the diagnostic, and it takes one query.

The fixes depend on what the giant arrays mean. If they are legitimate, repartition so those rows are spread out before exploding — though since explode is per-row that only helps if several large rows share a partition — or process them separately from the rest. If they are pathological, filter or cap them at ingestion, which is usually the right answer. And if you are exploding only to filter and re-aggregate, the higher-order functions do it without exploding at all, which removes the problem rather than managing it.

The reasoning

cache() marks a DataFrame; it stores nothing until an action runs, and only for the partitions that action touched. So df.cache() followed by three branches means the first branch computes and caches while the second and third may hit a partially populated cache. The idiom is to follow the cache with a count() to materialise all of it deliberately.

The second reason is plan matching. Spark reuses a cached result when a subtree of the plan matches the cached plan exactly. Add a filter or a column *before* the point you cached, or cache after a transformation and then reference the pre-transformation DataFrame, and the subtrees no longer match — so the cache exists, sits in memory, and is never read. The Storage tab shows it cached while the SQL plan shows a scan, which is the confirming pair of observations.

There is a third cause worth checking: eviction. With MEMORY_ONLY, partitions that do not fit are silently dropped and recomputed on access, so a cache larger than available storage memory behaves like no cache at all while still reporting as cached. MEMORY_AND_DISK spills them instead, which is why it is the sane default.

The reasoning

Spark gives you task-level fault tolerance, not job-level idempotency. A failed job can leave committed output behind — whole partitions the write already finished — and rerunning with append adds everything a second time. The job is green both times.

The fix is the write mode. Dynamic partition overwrite replaces only the partitions the batch contains, so rerunning a day produces the same table rather than a larger one. Static overwrite would also be idempotent and it replaces the *entire* table, which is a far more destructive default and a bug of its own when someone reruns one day.

On a table format the story is better and worth naming. Delta and Iceberg writes are atomic — a failed write commits nothing, so there is no partial state to reason about — and MERGE or replaceWhere gives idempotence on a key or a predicate. That combination is the reason 'use a table format' is the standard advice for anything that will be retried, which in practice is everything an orchestrator runs.

The reasoning

**Resource.** `OutOfMemoryError: Java heap space`, `GC overhead limit exceeded`, `ExecutorLostFailure`, "Container killed by YARN for exceeding memory limits", exit code 143. The cause is almost always a partition too large for the task, an oversized broadcast, or too much cached. Look at partition sizes, not at the total data volume.

**Shuffle and skew.** `FetchFailedException`, shuffle block not found, "Failed to send RPC to executor". These usually mean an executor died while others were fetching from it, so the real failure is one class up — but a genuinely enormous shuffle produces them on its own. Sort the stage's tasks by shuffle-read size before anything else.

**Serialization.** `Task not serializable`, `NotSerializableException`, Kryo failures. Caused by a closure dragging something non-serializable to the executors — a database connection, a logger, `self` in a class method. Nothing to do with data volume, and it fails immediately rather than at scale.

**Data and schema.** `AnalysisException` for a missing or ambiguous column, `NumberFormatException`, schema mismatch on read, corrupt Parquet or JSON. These fail on a specific file or a specific column, which is what distinguishes them from the resource class.

**Job logic.** `NullPointerException`, index errors, a UDF raising, division by zero, a join that explodes the row count. The job is doing exactly what you wrote.

**I/O and connectivity.** S3 404 or 403, `BlockMissingException`, "too many open files", cloud throttling. Permissions, missing paths, or rate limits — all environmental.

**Timeout and network.** Heartbeat timeouts, lost executors, `RPC message size exceeded`, fetch timeouts. Frequently a symptom of a long GC pause, which puts you back in the resource class — an executor that stops answering for 120 seconds is declared dead even though it was only collecting garbage.

The value of the taxonomy is what it rules out. "Out of memory" narrows to two classes and tells you to look at partition sizes; `Task not serializable` narrows to one and tells you to look at a closure. Reading the message for its *class* before its detail is what makes the next step obvious.

The reasoning

**Symptom.** A stage where 197 of 200 tasks finish in seconds and three run for twenty minutes. In the Spark UI, open the stage, sort tasks by duration and read the shuffle-read column: one task reading 8 GB where the median reads 200 MB is skew, and no amount of extra memory will change that. The distinction matters — a stage that is uniformly slow is a different problem entirely.

**Root cause.** Find the offending key by aggregating counts on the join key on the larger side and looking at the top few. It is usually one real value — a dominant country, a default merchant, or a placeholder like `unknown` or `-1` that stands in for every missing foreign key. That last case is worth separating out, because the fix is different: nulls and sentinels are often not join-worthy at all, and filtering them before the join removes the skew instead of managing it.

**Cheapest fix first.** If the other side of the join fits comfortably in executor memory, broadcast it — the shuffle disappears and the skew disappears with it. This is the fix to try first because it is one hint and no restructuring.

**Then AQE.** With `spark.sql.adaptive.enabled` and `spark.sql.adaptive.skewJoin.enabled`, Spark detects an outsized partition at runtime and splits it across several tasks, replicating the matching side. It handles the common cases without a code change. It does not help every case — it works on sort-merge joins, and if the skew is in an aggregation rather than a join it will not fire.

**Salting last.** Add a random suffix to the skewed key on the large side, explode the small side across the same suffix range, join on the salted key, then drop the salt and aggregate. It always works and it is the most invasive: the code gets harder to read and the small side gets multiplied.

**Verify.** Rerun and check the same number you diagnosed with — max task duration, not total runtime. Going from a 25-minute straggler to a 3-minute maximum is the result; a wall-clock number on its own does not tell you whether you fixed skew or just got a quieter cluster.

The reasoning

Every function you hand to a transformation is shipped to the executors, and shipping means serializing. The function comes with everything it references — and a reference to any field of the enclosing object drags **the whole object** along, which is why touching one string field of a class can fail on a database handle three fields away.

The usual culprits are things that are inherently local: an open connection, an HTTP client, a logger, a file handle. None of them can be meaningfully sent to another machine, and Spark tells you so before running a single task, which is at least a fast failure.

Three fixes, in the order they apply. **Create it on the executor** — `mapPartitions` opens the connection once per partition inside the worker, which is both the correct fix and the efficient one. **Broadcast the data** if what you needed was a lookup table rather than a live object. **Copy the field to a local variable** before the closure when the only problem is that a method reference is pulling in `self`.

The wrong instinct is to make the class serializable so the error goes away. That either fails at runtime on the non-serializable field anyway, or it succeeds and quietly ships a large object to every task, which is a performance problem where you previously had a clear error.