Apache Spark Common Errors and Fixes with Practical Examples

Spark jobs can fail in ways that look mysterious when production is on fire and the clock is ticking. This guide explains common Apache Spark error messages, why they happen, and safe ways to fix them with practical, reproducible examples. You will:

Inventory your environment so you do not chase the wrong fix.
Choose safe, scoped configuration changes first.
Verify the outcome using Spark UI, logs, and targeted checks.
Understand failure modes and rollback cleanly if needed.
Keep a concise operations checklist you can follow under pressure.

The examples use constructed sample code and hypothetical numbers that you can adapt to your stack. Apply each change to a narrow, measurable pilot and inspect results locally or on a dedicated dev workspace before rollout.

Version and Environment Inventory

Many Spark errors are version- or environment-sensitive. Capture these facts first and attach them to any incident or change record.

Spark version and build info

CLI:

spark-submit --version
spark-shell --version

From REPL:

// Scala
spark.version

# Python
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
print(spark.version)

Runtime languages and Java

Java: java -version
Scala: scala -version (if relevant)
Python: python --version or python3 --version

Cluster manager and resource shape

Determine master:
Scala: spark.sparkContext.master
Python: spark.sparkContext.master
Show current conf (sanitized):
Scala: spark.sparkContext.getConf.getAll.foreach(println)
Python:

    for k, v in spark.sparkContext.getConf().getAll():
        print(k, v)

Record executor count, cores, memory. If you submit jobs, log the flags you used, for example:
--num-executors 8 --executor-cores 4 --executor-memory 6g --driver-memory 2g (constructed example)

Storage and file system endpoints

HDFS: hdfs dfsadmin -report and a sample hdfs dfs -ls /path of your input and output roots.
Object stores: the bucket/endpoint names (do not log secrets), and the Hadoop or Spark configs you use to authenticate (redacted).

Libraries and connectors

List extra JARs and packages provided with --packages or --jars.
Note any serialization settings (Java vs Kryo) and relevant SQL configs.

Having this inventory shortens the path from symptom to root cause and makes fixes reproducible across environments.

Safe Configuration Path

Treat configuration as a scalpel, not a hammer. Change the smallest scope first, for the shortest time, and verify. Below is a practical sequence you can follow for most Spark errors.

Reproduce on a narrow, measurable pilot

Filter to a small but representative partition or date range.
Keep input and output paths separate from production.

Prefer session/job-scoped settings first

Use --conf on spark-submit or spark.conf.set("key", value) in-session.
Avoid editing cluster-wide defaults until the fix is proven.

Change one variable at a time

For example, first raise spark.sql.shuffle.partitions (or lower), then evaluate; only then touch memory or broadcast thresholds.

Observe effects immediately

Track stage run times, shuffle read/write sizes, task failures, GC time, and skew indicators in the Spark UI.

Codify the fix

Once stable, pin the configuration in code (where appropriate) or document the exact submit flags. Keep a rollback setting ready.

Common Errors and Safe First Fixes

The table below maps frequent errors to the primary cause and a first safe fix to try on your pilot job.

Error message (abridged)	Likely cause	Safe first fix
`SparkException: Job aborted due to stage failure (FetchFailedException)`	Large shuffle, unstable network, or executor lost mid-shuffle	Reduce shuffle size (repartition by key, increase partitions modestly), enable retries, validate executor stability
`OutOfMemoryError: Java heap space (executor)`	Partitions too large or wide transformations materializing huge rows	Increase partitions, persist selectively to DISK or MEMORY_AND_DISK, avoid `.collect()` on large RDD/DataFrame
`OutOfMemoryError: Java heap space (driver)`	Driver collecting or holding big structures	Replace `collect()` with `take()`, write to storage, move logic to executors
`SparkException: Task not serializable`	Non-serializable object captured in closure	Use serializable case class, broadcast small lookups, use `mapPartitions` with local factory
`AnalysisException: Path already exists when saving`	Output path not cleaned or save mode not set	Use `.mode("overwrite")` intentionally, or remove target path before write
`ClassNotFoundException (e.g., Kafka source)`	Missing connector JAR/package	Add correct package via `--packages` matching Spark/Scala version
`Py4JJavaError (generic wrapper)`	Underlying JVM exception surfaced to Python	Expand cause in logs, read innermost cause to pick a targeted fix
Skew: long tails in a few tasks	Hot keys or imbalanced partitions	Salting/skew join hints, AQE enabled, or custom partitioner

Verification and Diagnostics

Turn a vague failure into a concrete diagnosis with these steps and checks.

1) Inspect the Spark UI and Logs

Stages tab: look for very large shuffle reads/writes, or tasks with extreme durations compared to peers (skew).
Executors tab: check executor lost counts, GC time %, and peak memory.
Driver and executor logs: find the first root-cause exception (not the wrapper). For PySpark, expand the Py4J stack to the innermost Java exception.

2) Reproduce with a Minimal Query or Function

Create a small, representative dataset and run the problematic transformation.
Leave only one suspected change at a time and compare before/after metrics.

3) Concrete Diagnostics and Examples

A) FetchFailedException during join or aggregation

Constructed example symptom:

Error: org.apache.spark.SparkException: Job aborted due to stage failure: FetchFailed ... after a wide shuffle.

Checks and fixes:

Check shuffle size: in SQL, read the SQL tab or in code print partition counts.
Modestly increase shuffle partitions for SQL:

  spark.conf.set("spark.sql.shuffle.partitions", 400)  # constructed example

If using RDD API, use rdd.repartition(400) or, for known keys, use partitionBy(numPartitions) on PairRDDs.
Enable adaptive query execution (AQE) if available:

  spark.conf.set("spark.sql.adaptive.enabled", True)

Verify: re-run the stage; compare max shuffle read per task and the number of failed fetches.

B) Executor OutOfMemoryError

Constructed example symptom:

Error: java.lang.OutOfMemoryError: Java heap space in executor logs while doing groupBy().agg() on wide rows.

Safe steps:

Reduce partition size by increasing parallelism carefully:

  spark.conf.set("spark.sql.shuffle.partitions", 400)

Persist selectively using a storage level that can spill:

  df_to_reuse = df_heavy.transform(some_fn)
  df_to_reuse.persist(storageLevel="MEMORY_AND_DISK")

Avoid .collect() or .toPandas() on large data. Use .limit().collect() for sampling or write to storage.
If still failing, increase executor memory modestly (constructed example):
--executor-memory 6g --executor-cores 4
Verify: monitor executor GC time and failed tasks; confirm no OOM in logs.

C) Driver OutOfMemoryError

Constructed example symptom:

Error on driver: java.lang.OutOfMemoryError: Java heap space after .collect() on a large DataFrame.

Fixes:

Replace .collect() with .take(1000) for sampling, or write results via .write to storage.
If you must hold moderate data on the driver, increase driver memory modestly (constructed): --driver-memory 4g.
Verify: job completes; driver logs show stable heap without GC thrash.

D) Task Not Serializable

Constructed Scala example of the problem:

case class Item(id: Long, v: Double)
class NonSerializable(val factor: Double)
val helper = new NonSerializable(2.0)
val rdd = sc.parallelize(Seq(Item(1, 3.0), Item(2, 4.0)))
val out = rdd.map(x => x.v * helper.factor).collect() // fails

Fix options:

Make state serializable or avoid capturing it in the closure:

  case class SerializableHelper(factor: Double) extends Serializable
  val helper = SerializableHelper(2.0)
  val out = rdd.map(x => x.v * helper.factor).collect()

Or instantiate per partition:

  val out = rdd.mapPartitions { it =>
    val helper = new NonSerializable(2.0)
    it.map(x => x.v * helper.factor)
  }.collect()

Verify: job runs; no Task not serializable error in executor logs.

E) AnalysisException: Path Already Exists

Constructed example:

df.write.mode("errorifexists").parquet("/data/out/2024-01-01")

Fixes:

Use .mode("overwrite") intentionally when safe:

  df.write.mode("overwrite").parquet("/data/out/2024-01-01")

Or remove target path first (HDFS example):

  hdfs dfs -rm -r /data/out/2024-01-01

Verify: path shows new files and job succeeded.

F) ClassNotFoundException for External Connectors (e.g., Kafka)

Constructed example symptom:

Error: java.lang.ClassNotFoundException: org.apache.spark.sql.kafka010.KafkaSourceProvider

Fix:

Add the correct package matching your Spark and Scala version at submit time. Pattern (constructed):

  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:<spark_version>

Verify: the session starts and readStream.format("kafka") no longer errors.

G) Skew Causing Long Tails

Diagnosis:

from pyspark.sql import functions as F
(df.groupBy("key").count()
 .orderBy(F.desc("count"))
 .limit(10)
 .show())

Fixes:

Enable AQE to handle skewed joins where supported:

  spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)

Apply salting (constructed):

  import pyspark.sql.functions as F
  salt_buckets = 10
  left_salted = left.withColumn("salt", F.rand(seed=42) * salt_buckets).withColumn("salt", F.floor(F.col("salt")))
  right_salted = right.withColumn("salt", F.lit(0))
  joined = left_salted.join(right_salted, ["key", "salt"], "left")

Verify: tail tasks shorten; fewer stragglers in the Stages tab.

Failure Modes and Recovery

Even a reasonable change can have side effects. Use this section to anticipate risks and keep a rollback plan.

Increasing spark.sql.shuffle.partitions too high
Risk: Too many tiny tasks and small output files; higher scheduler overhead.
Recovery: Revert to the previous value; compact small files with a targeted coalesce/repartition write.

Disabling broadcast joins globally
Risk: Large shuffles replacing efficient broadcasts; longer runtimes.
Recovery: Restore broadcast threshold; use join hints only where needed.

Over-allocating executor memory
Risk: Fewer executors fit on cluster; longer queues; possible container kill by resource manager.
Recovery: Return to prior memory; consider increasing partitions or spill-friendly storage levels instead.

Aggressive caching without unpersist
Risk: Cache fills executor memory; eviction thrashing and OOMs.
Recovery: Call unpersist() on DataFrames/RDDs no longer needed; restart the application if memory is fragmented.

Overwriting output paths unintentionally
Risk: Data loss.
Recovery: Stop the job, restore from backup or previous partition if available. Enforce a guard that compares record counts or checks for watermarks before overwrite.

Frequent FetchFailed due to unstable nodes
Risk: Repeated retries and long tails.
Recovery: Quarantine the bad executors or nodes; resubmit after node health stabilizes.

Operations Checklist

Use this short checklist when handling Spark errors on-call or during postmortems.

Inventory

Record Spark, Java/Scala/Python versions.
Capture master, executors, cores, memory, and key --conf flags.
Note input/output paths and connector packages.

Reproduce on a Pilot

Narrow the dataset by date/partition.
Save logs, Spark UI screenshots, and the minimal failing code/query.

Diagnose

Spark UI: find the first failing stage and the largest shuffle read/write.
Logs: extract the innermost exception and stack trace.
Check skew via top key counts or extreme task durations.

Apply a Scoped Fix

Prefer --conf or spark.conf.set over cluster defaults.
Change one variable at a time; document expected and observed effects.

Verify

Compare before/after metrics: stage time, shuffle sizes, GC %, failures.
Validate row counts or checksums of outputs.

Rollback

Keep previous conf values handy; revert quickly if side effects appear.
Clean partial outputs and re-run idempotently.

Close-out

Codify the fix in code or submission scripts.
Update runbooks and add a guardrail test where feasible.

Practical Configuration Reference

Use these knobs carefully and prefer job-scoped changes first. Values below are constructed examples; tune based on measurements.

Setting	Scope	Safe starter change	Effect
`spark.sql.shuffle.partitions`	SQL session/job	200 → 400	Smaller partition sizes reduce per-task memory pressure; may increase task count
`spark.sql.adaptive.enabled`	SQL session/job	false → true	Lets Spark optimize joins/shuffles at runtime; helps with skew
`spark.sql.autoBroadcastJoinThreshold`	SQL session/job	default → 50MB	Tune to allow/avoid broadcast joins; avoid broadcasting huge tables
`spark.serializer`	App/cluster	Java → Kryo (with registration)	Faster serialization for heavy RDD pipelines; validate compatibility
`spark.executor.memory`	Job/cluster	+1-2g	More heap per executor; reduces OOMs if partitions still too large
`spark.executor.cores`	Job/cluster	4 → 2-4	Adjust to balance CPU vs memory per task; avoid oversubscription
`spark.memory.fraction`	Job/cluster	default	Rarely change first; prefer partitioning and spill-friendly storage

End-to-End Worked Example (Constructed)

Scenario: A daily join between a 120M-row fact table and a 200K-row dimension table fails with FetchFailedException and occasional executor OOM.

Inventory

Spark 3.3.x, master on YARN, 12 executors, 4 cores each, 6g executor memory.
spark.sql.shuffle.partitions=200 (default), AQE disabled.

Reproduce on a Pilot

Filter fact to a single day (~5% of data) and run the join in a dev workspace.

Diagnose

Spark UI shows 1.2 TB shuffle read on the big join; a few tasks reading 20+ GB each; long tails.

Scoped Fixes

Enable AQE and skew handling:

     spark.conf.set("spark.sql.adaptive.enabled", True)
     spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)

Raise shuffle partitions to 400 to shrink per-task load:

     spark.conf.set("spark.sql.shuffle.partitions", 400)

Verify that the 200K-row dimension is broadcasting instead of shuffling:

     spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)

Verify

Shuffle read drops per task; no FetchFailed; runtime improves from 45 min to 18 min on pilot. No OOM in executor logs.

Rollout

Bake these session confs into the job submit; leave cluster defaults unchanged.
Monitor the next full run; keep a rollback option that returns shuffle partitions to 200 and disables AQE if regressions appear.

Close-out

Document the new settings and the measured impact. Add a runbook entry for handling future skew joins.

Conclusion

Most Spark errors boil down to a few root causes: too much work per task, imbalanced partitions, missing dependencies, or unsafe patterns such as collecting large datasets on the driver. By inventorying your environment, applying scoped and measurable changes first, verifying with Spark UI and logs, and keeping clear rollback steps, you can turn chaotic incident response into a deliberate, low-risk practice. Start small, measure, and promote the fix only when you can show it works on a pilot. That approach steadily reduces repeat incidents and makes daily operations calmer and faster.

Apache Spark Common Errors and Fixes with Practical Examples

Version and Environment Inventory

Safe Configuration Path

Common Errors and Safe First Fixes

Verification and Diagnostics

1) Inspect the Spark UI and Logs

2) Reproduce with a Minimal Query or Function

3) Concrete Diagnostics and Examples

Failure Modes and Recovery

Operations Checklist

Practical Configuration Reference

End-to-End Worked Example (Constructed)

Conclusion

Related Research

Article Quality Score