Spark Task Failed 4 Times — Read the First Trace

It's 2 AM, your Spark job has been running for three hours, and the driver log ends with the dreaded line: 'Job aborted due to stage failure: Task 42 in stage 7 failed 4 times.' Your first instinct is to restart the job and hope. Your second instinct is to bump spark.task.maxFailures and hope harder. Both are wrong, and both cost another three hours.

That final error message is a wrapper. It tells you Spark gave up, not why it gave up. The why lives somewhere else entirely — in the stderr of the first failed attempt, usually 500 lines above where you're looking. Engineers who don't know this burn entire nights re-running jobs that fail identically, because the root cause never changed between attempt 1 and attempt 4.

Three failure families cause nearly every 'failed 4 times' abort. Shuffle fetch failures mean a task couldn't read another executor's output — typically because that executor died or the network dropped. Executor losses with OOM signatures mean the box was too small for the partition it chewed on. And code bugs — nulls, bad casts, exploding UDFs — mean your logic met production data for the first time and lost.

This article teaches you to triage in minutes. You'll learn where Spark hides the first attempt's trace, how to read the signature of each failure family, and how to tune retries honestly instead of using them as a rug to sweep bugs under.

The Wrapper Problem: Why the Last Error Lies

When Spark aborts a job, the DAGScheduler throws whatever the fourth attempt reported — and by then the cluster is usually in a degraded state that colors the evidence. The first attempt died on your real bug; the fourth attempt often dies on the wreckage (missing shuffle blocks from executors killed during attempts 1-3, timed-out heartbeats on an overloaded driver). Reading only the final error is like diagnosing a car crash from the tow-truck receipt.

The mechanics are simple. spark.task.maxFailures (default 4) counts consecutive task failures; each failure is retried on (usually) a different executor. Separately, a stage is retried up to spark.stage.maxConsecutiveAttempts (default 4) when it fails wholesale. So 'failed 4 times' describes one unlucky task, while the job-level abort is the scheduler deciding the stage can't proceed. Both counters reset the moment anything succeeds, which is why flaky infrastructure sometimes squeaks through and real bugs never do.

Your first move is always the same: isolate attempt 0. In the Spark UI, the failed stage lists every task attempt with its own stderr link. In logs, search backward from the abort for the first TaskEnd/TaskFailed event. Enable event logging (spark.eventLog.enabled=true with a durable dir) on every production job so this archaeology is possible after the cluster is gone — without it, you're debugging from memory the next morning.

spark-submit \
  --conf spark.eventLog.enabled=true \
  --conf spark.eventLog.dir=s3://my-bucket/spark-events/ \
  --conf spark.task.maxFailures=4 \
  --conf spark.stage.maxConsecutiveAttempts=4 \
  --conf spark.blacklist.enabled=true \
  --conf spark.blacklist.task.maxTaskAttemptsPerExecutor=2 \
  --conf spark.network.timeout=300s \
  --conf spark.executor.heartbeatInterval=20s \
  checkout_job.py

# After the abort, find the FIRST failure (not the wrapper):
# grep -m1 'TaskEnd.*TaskFailed' application.log | head -c 2000

FetchFailedException: Blame the Map Side, Not the Task

FetchFailedException means a reduce task asked for shuffle blocks and couldn't get them. Beginners blame the reduce task and its memory; veterans check the executor that was supposed to serve those blocks. Nine times out of ten, that executor is already dead — killed by OOM, preempted by the cluster manager, or partitioned off the network — and the fetch failure is just the obituary.

The tell is timing. Open the driver log and line up timestamps: a 'Lost executor 7' message seconds before your FetchFailed means the map side died first. The shuffle data it held died with it (unless you run the external shuffle service, which keeps blocks alive across executor loss). Spark then resubmits the map tasks and retries your reduce — which is exactly why one dead executor can burn through all 4 attempts of hundreds of downstream tasks.

If no executor died, you're looking at genuine shuffle-service or network trouble: an overloaded external shuffle service, a bad NIC dropping packets on one rack, or shuffle files cleaned up too aggressively. Check spark.shuffle.service.enabled against your deploy mode (YARN needs it for decommissioning safety), and look for fetch failures clustering on one host — a single sick node producing most fetch errors is a hardware ticket, not a Spark ticket.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("fetch-triage").getOrCreate()
sc = spark.sparkContext

# After a FetchFailed abort, pull the failed task attempts via the REST API
# and count fetch failures per serving host to spot one sick node.
import json
import urllib.request

APP_ID = "application_1729000000000_0042"  # from the YARN / Spark master UI
HISTORY = "http://history-server:18080"

with urllib.request.urlopen(f"{HISTORY}/api/v1/applications/{APP_ID}/stages") as r:
    stages = json.load(r)

failed = [s for s in stages if s["status"] == "FAILED"]
print(f"Failed stages: {[s['stageId'] for s in failed]}")
for s in failed:
    sid = s["stageId"]
    with urllib.request.urlopen(
        f"{HISTORY}/api/v1/applications/{APP_ID}/stages/{sid}/0/taskList"
        f"?length=1000&sortBy=-launchTime"
    ) as r:
        tasks = json.load(r)
    bad = [t for t in tasks if t["status"] == "FAILED"]
    print(f"Stage {sid}: {len(bad)} failed task attempts out of {len(tasks)} listed")
    print("First error:", (bad[0].get("errorMessage") or "")[:400] if bad else "none")

ExecutorLostFailure and OOM: Prove It's Memory First

ExecutorLostFailure with exit code 143 (SIGTERM, usually YARN killing an over-limit container) or 137 (SIGKILL, OOM-killer) looks like a memory verdict, but it only proves the container died — not that your sizing was wrong. A skewed partition 50x the median will kill any reasonably sized executor, and no amount of fleet-wide memory fixes a single giant task. The proof step takes two minutes: compare the dead task's input + shuffle-read bytes against its siblings.

YARN's 'exceeding memory limits' message deserves a close read because it names two numbers: physical memory used versus requested. If the overage is small (a few hundred MB over), bump spark.executor.memoryOverhead — the default max(384M, 10% of heap) is too thin for shuffle-heavy or native-code workloads. If the message says 'Java heap space' instead, the JVM heap itself filled: your partitions are too fat, your cache is retained too long, or a UDF is materializing whole partitions in memory.

Off-heap and GC logs turn guesses into measurements. Enable spark.memory.offHeap.enabled with spark.memory.offHeap.size for shuffle-heavy stages, and read the executor stderr for 'GC overhead limit exceeded' (heap churn from object-heavy code — switch to SQL functions or Kryo) versus a flat climb to the ceiling (genuinely too much data per task — split partitions). Fix the shape before the size: one repartition that halves max-partition bytes beats doubling memory across 200 executors on both cost and reliability.

Code Bugs That Only Bite at Scale: Nulls, Casts, UDFs

The cruelest 'failed 4 times' aborts come from code that passed every test. Tests run on clean samples; production runs on 4.3 billion rows including the one malformed record an upstream producer started emitting Tuesday. Nulls where your UDF assumes strings, 'N/A' where you cast to int, timestamps in a format nobody documented — each kills the task deterministically on every attempt, and retries just re-read the same poison row four times.

Python UDFs are the classic carrier because they run outside the Catalyst optimizer's safety nets: no null-safe codegen, whole-row serialization, and exceptions that surface as inscrutable Py4JJavaError wrappers. A scalar UDF calling .strip() on None fails exactly the way our incident did. Prefer built-in SQL functions (they're null-safe by design and run 10-100x faster), and when you must use a UDF, guard every input and PNL-test it against a sample of real production data, not hand-written fixtures.

The durable fix is a quarantine path, not just a guard. Route rows that fail validation into a separate output with a reason column, and alert on quarantine volume. Guards stop the crash; quarantine tells you the upstream schema drifted before the next field breaks. Pair it with mapPartitionsWithIndex during triage to identify exactly which partition — and therefore which input files — harbor the poison rows.

from pyspark.sql import functions as F
from pyspark.sql import types as T

orders = spark.read.parquet("s3://lake/orders/dt=2026-09-22/")

# Null-safe parsing with built-ins: no UDF, no NPE, 10-100x faster.
parsed = orders.withColumn(
    "promo_clean",
    F.upper(F.trim(F.coalesce(F.col("promo_code"), F.lit("")))),
)

# Durable fix: quarantine rows that violate the contract instead of
# letting one poison row burn 4 task attempts and abort the job.
valid = parsed.filter(
    (F.col("order_id").isNotNull()) & (F.length("promo_clean") <= 24)
)
quarantine = parsed.filter(
    (F.col("order_id").isNull()) | (F.length("promo_clean") > 24)
).withColumn("quarantine_reason", F.lit("contract_violation"))

valid.write.mode("overwrite").parquet("s3://lake/orders_clean/dt=2026-09-22/")
quarantine.write.mode("overwrite").parquet("s3://lake/quarantine/dt=2026-09-22/")

q = quarantine.count()
if q > 0:
    print(f"QUARANTINED {q} rows — check upstream schema before next run")

Tuning Retries Honestly: Blacklists, Timeouts, Speculation

Retry knobs exist for genuinely transient failures — a preempted container, a blipped rack, a straggler disk — and each knob has a price tag. Raising spark.task.maxFailures from 4 to 8 on a stage whose tasks run 30 minutes each means one sick task can burn 4 extra cluster-hours before aborting; across 200 tasks that's a potential 800 paid hours per run. Do that math in the incident channel before touching the knob, because 'just bump retries' is how a 2-hour outage becomes a $4,000 invoice.

Blacklisting is the cheapest honest retry: with spark.blacklist.enabled=true, executors (or whole nodes) that fail repeatedly are excluded from future attempts, so your 4 tries spread across healthy hardware instead of hammering the same sick box. Pair it with spark.blacklist.task.maxTaskAttemptsPerExecutor=1 for fetch-failure storms — one strike per executor forces the scheduler to move on immediately.

Timeouts and speculation handle the slow rather than the dead. If long GC pauses starve heartbeats, raise spark.network.timeout (which also lifts heartbeat and fetch timeouts) to 300s or more rather than watching healthy executors get declared dead. Enable spark.speculation=true only for short-task stages with stragglers — it launches duplicate copies of slow tasks and keeps whichever finishes first, which wastes work on long tasks but saves whole jobs when one disk goes sluggish.

# Price an extra retry before you buy it.
# Inputs: task minutes, task count, $/executor-hour, executors per task.
TASK_MIN=30        # p50 task duration in this stage
TASKS=200          # tasks in the stage
DOLLARS_PER_HR=0.9 # blended executor-hour rate
EXTRA_ATTEMPTS=4   # raising maxFailures 4 -> 8

worst=$(python3 -c "print(${TASKS} * ${EXTRA_ATTEMPTS} * ${TASK_MIN} / 60 * ${DOLLARS_PER_HR})")
echo "Worst-case extra spend per run: \$$worst"
# 200 tasks x 4 extra attempts x 0.5h x $0.90 = $360 per run, worst case.
# If the bug is deterministic (attempt 0 = attempt 3), that money buys zero information.

# Honest transient-only retry profile:
# --conf spark.blacklist.enabled=true
# --conf spark.blacklist.task.maxTaskAttemptsPerExecutor=1
# --conf spark.network.timeout=300s
# --conf spark.speculation=true   # short-task stages with stragglers only

Reading the Spark UI Like a Paramedic

The Spark UI rewards a fixed reading order the way an ER rewards triage order: vitals first, history second, labs last. Start at Stages — the failed stage is red, and its description names the operation (often a wholeStage codegen range like 'parquet at X'). Click through to the task table and sort by duration or shuffle-read: one task 20x slower or fatter than the median is skew; all tasks uniformly slow is sizing; a handful of instant failures is poison data or a missing file.

Next, open the Timeline view. Tasks that die in waves right after a cluster event (autoscaling, decommissioning, a deploy) point at infrastructure, not code. The Event Timeline also shows when executors were added and removed — if your failures start exactly when spot nodes were reclaimed, you've found the killer without reading a single stack trace.

Finally, read the SQL tab for the failed stage's query plan. It shows the exact join order, broadcast decisions, and partition counts the optimizer chose. A SortMergeJoin on two terabyte tables where you expected a broadcast tells you autoBroadcastJoinThreshold was missed (check for type mismatches killing the broadcast). The plan plus the task histogram together answer 80% of 'failed 4 times' pages before you ever SSH anywhere.

-- After the abort, reproduce the stage's data shape in SQL to confirm
-- skew vs sizing before you change any Spark setting.
-- Run against the stage's input; replace with your table + key.

-- 1. Is one key eating the stage? (skew if top share > ~10%)
SELECT promo_code, COUNT(*) AS n,
       ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 1) AS pct
FROM lake.orders
WHERE dt = '2026-09-22'
GROUP BY promo_code
ORDER BY n DESC
LIMIT 20;

-- 2. Are partitions uniformly fat? (sizing if p99 ~ p50 and both huge)
-- Approximate with a hash bucket histogram over the join key.
SELECT bucket, COUNT(*) AS n
FROM (
  SELECT ABS(HASH(order_id)) % 200 AS bucket FROM lake.orders
  WHERE dt = '2026-09-22'
) t
GROUP BY bucket
ORDER BY n DESC
LIMIT 10;