Spark Core · Module 3 · Production issues

When lazy execution bites

Problems that come from how Spark plans and runs jobs, rather than from memory or data size. Each follows the same path: symptom, why, how to confirm, fix, prevention, interview answer.

Recomputed lineage Leftover debug actions Huge plans from loops Fetch failures Job aborted after retries Errors far from the cause Interview questions

The same data is read and computed again and again

Symptom

A pipeline takes two or three times longer than its data size suggests. The Spark UI shows the same file scan in several jobs, each with the full input size.

Why it happens

Every action re-runs its DataFrame's lineage from the source. A DataFrame used by three writes is computed three times.
Expensive steps (joins, wide aggregations, UDFs) upstream of several outputs are repeated each time.

How to confirm

SQL / DataFrame tab: identical scan and join sub-plans in several queries.
Input size summed across jobs is a multiple of the table size.

Fix

Cache the shared DataFrame before the first action, and unpersist() after the last.
For long or heavy shared steps, write an intermediate table once and read it in later steps; this also makes reruns cheaper.
Combine outputs: one partitionBy write instead of a loop of filtered writes.

enriched = orders.join(customers, "cust_id").filter("status = 'PAID'").cache()
enriched.groupBy("segment").sum("amount").write.saveAsTable("gold.by_segment")
enriched.groupBy("country").sum("amount").write.saveAsTable("gold.by_country")
enriched.unpersist()

Prevent

When one DataFrame feeds several outputs, decide on purpose between recompute, cache and an intermediate table.

Say it in an interview

"Spark is lazy per action, so every action recomputes its lineage. When one expensive DataFrame feeds several outputs I cache it, or write it once as an intermediate table, and I make sure to unpersist afterwards."

Debug actions left in production code

Symptom

Extra jobs labelled count, show or collect appear before the real write, each doing a full pass over the data. Runtime is far above what the final write needs.

Why it happens

print(df.count()), df.show() and similar lines added during development are real actions. Each one runs the whole lineage. Logging row counts at every step can double or triple a job's runtime.

Fix

Remove them, or guard them behind a debug flag.
Get counts from the write itself where possible (Delta table history, the SQL tab's "rows output", or write metrics) instead of an extra pass.
If you need a count and the data, cache first so the count and the write share one computation.

Say it in an interview

"Things like count and show are actions, so each one runs the full lineage. In production I remove debug actions or put them behind a flag, and take row counts from write metrics or table history instead."

Huge plans from loops

Symptom

java.lang.StackOverflowError
    at org.apache.spark.sql.catalyst.trees.TreeNode...

Or the driver spends minutes "planning" before any task runs, with the Spark UI idle.

Why it happens

Calling withColumn hundreds of times in a loop, or unioning DataFrames in a loop, builds a very deep plan. Each call also triggers re-analysis of the whole plan, so planning time grows quickly.
Iterative algorithms that reuse the previous iteration's DataFrame grow lineage every round.

Fix

Add many columns in one call: select with a list, or withColumns with a dict (Spark 3.3+).
Union a list at once with functools.reduce over a small number of inputs, or read them as one source.
In iterative jobs, cut lineage every few rounds with checkpoint(), localCheckpoint(), or by writing and re-reading.

# before: 300 withColumn calls, 300 plan re-analyses
for c in cols:
    df = df.withColumn(c, F.trim(F.col(c)))

# after: one projection
df = df.select(*[F.trim(F.col(c)).alias(c) if c in cols else F.col(c) for c in df.columns])

Say it in an interview

"Building a DataFrame in a long loop, like hundreds of withColumn calls, creates a huge plan that's slow to analyse and can overflow the stack. I build the columns in a single select, and in iterative jobs I truncate lineage with checkpoints."

Fetch failures and stage retries

Symptom

org.apache.spark.shuffle.FetchFailedException: Failed to connect to host-12/10.0.4.12:7337
Stage 7 (shuffle map stage) resubmitted due to fetch failure

Stages show as retried; earlier stages run again; total runtime balloons.

Why it happens

A stage's tasks fetch shuffle files written by the previous stage's executors. If one of those executors died (OOM, spot reclaim, node failure, long GC), its shuffle files are gone. Spark must rerun the parent stage's tasks that produced them, using lineage, before continuing.

How to confirm

Stages tab: retried stage attempts; failed tasks mention FetchFailed.
Executors tab: lost or removed executors around the same time, and why.

Fix

Fix why executors die: memory (see Module 1 failures), GC, or preemption.
Keep shuffle files available after executor loss: external shuffle service, or graceful decommissioning with block migration on spot or Kubernetes.
Use more reliable nodes for shuffle-heavy jobs.

Say it in an interview

"A FetchFailedException means shuffle output from an earlier stage is no longer reachable, usually because its executor died. Spark recomputes the missing parent tasks from lineage. I fix the executor loss itself, and use a shuffle service or decommissioning so shuffle files survive."

Job aborted after four task failures

Symptom

Job aborted due to stage failure: Task 112 in stage 3.0 failed 4 times,
most recent failure: ... java.lang.NumberFormatException: For input string: "N/A"

Why it happens

A task is retried up to spark.task.maxFailures (4) times. If the cause is deterministic (a bad record, a parsing bug, a failing UDF), every retry hits the same row and fails, so the job aborts. Retries only help with transient causes such as a flaky node or network blip.

How to confirm

Read the most recent failure's exception, not just "failed 4 times".
Same task index failing each attempt on different executors means it's the data, not the machine.

Fix

Find and handle the bad records: try_cast in SQL (Spark 3.2+) or casts that return NULL, explicit validation, or quarantine rows that fail rules.
For file sources, read in PERMISSIVE mode with a corrupt-record column (or badRecordsPath on Databricks) instead of failing the job.
Make UDFs defensive about NULLs and unexpected values.

Say it in an interview

"Task retries help only with transient failures. If the same task fails four times on different executors it's deterministic, usually a bad record, so I read the root exception, find the row, and handle it with safe casts, permissive parsing or a quarantine table."

Errors that surface far from their cause

Symptom

A failure is reported at the final write on line 300, but the cause is a cast, UDF or division written on line 40. The stack trace points into generated code, not your line.

Why it happens

Because execution is lazy, data-dependent errors only happen when an action actually processes rows, so they're reported at that action. Everything in the lineage runs inside the same stage, so the trace doesn't say which of your lines produced it. With Spark Connect, even analysis errors such as a wrong column name move to the action.

Fix

Bisect: add a temporary action (.limit(1000).collect()) after suspected steps, in a notebook or test, to find where it breaks.
Unit-test transformation functions on small DataFrames so data errors show up near their cause.
With ANSI mode (spark.sql.ansi.enabled, on by default since Spark 4.0), invalid casts and overflows raise errors that name the expression instead of silently producing NULLs.

Say it in an interview

"Lazy execution means data errors appear at the action, not where they're written. I keep transformations in small, unit-tested functions, and when debugging I bisect with temporary limited actions until I find the failing step."

Common interview questions

Explain the Spark execution flow from spark-submit to job completion.

The driver starts, creates the SparkSession and gets executors from the cluster manager. Transformations build a logical plan. An action creates a job; Catalyst optimises the plan; the DAG scheduler cuts it into stages at shuffles; the task scheduler sends one task per partition to executor cores; stages run in dependency order, with shuffle files passed between them; results return to the driver, and failed tasks are retried using lineage.

How are jobs, stages and tasks created?

Each action creates a job. The DAG scheduler walks back from the result and creates a new stage at each shuffle dependency. Each stage has one task per partition of its data.

What's lazy evaluation and why does Spark use it?

Transformations only record a plan; work happens when an action needs a result. That lets Catalyst optimise the whole chain, fuse narrow steps into one pass, and recover lost partitions from lineage. It's per action, so each action recomputes its lineage unless cached.

What are narrow and wide transformations, and how do they relate to stages?

Narrow transformations build each output partition from one input partition (filter, select, map), so they're pipelined within a stage. Wide ones need data from many partitions (groupBy, distinct, most joins), so they shuffle, and each shuffle is a stage boundary.

Why are RDDs immutable?

So lineage-based recovery is safe (a recomputed partition equals the original) and parallel tasks need no locking. Every transformation returns a new RDD or DataFrame.

Your job runs much longer than expected and the UI shows the same scan in several jobs. Why?

Several actions are re-running the same lineage. Cache the shared DataFrame, write an intermediate table, or remove debug actions.