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 causeSafe first fix
SparkException: Job aborted due to stage failure (FetchFailedException)Large shuffle, unstable network, or executor lost mid-shuffleReduce 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 rowsIncrease 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 structuresReplace collect() with take(), write to storage, move logic to executors
SparkException: Task not serializableNon-serializable object captured in closureUse serializable case class, broadcast small lookups, use mapPartitions with local factory
AnalysisException: Path already exists when savingOutput path not cleaned or save mode not setUse .mode("overwrite") intentionally, or remove target path before write
ClassNotFoundException (e.g., Kafka source)Missing connector JAR/packageAdd correct package via --packages matching Spark/Scala version
Py4JJavaError (generic wrapper)Underlying JVM exception surfaced to PythonExpand cause in logs, read innermost cause to pick a targeted fix
Skew: long tails in a few tasksHot keys or imbalanced partitionsSalting/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: - 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.
SettingScopeSafe starter changeEffect
spark.sql.shuffle.partitionsSQL session/job200 → 400Smaller partition sizes reduce per-task memory pressure; may increase task count
spark.sql.adaptive.enabledSQL session/jobfalse → trueLets Spark optimize joins/shuffles at runtime; helps with skew
spark.sql.autoBroadcastJoinThresholdSQL session/jobdefault → 50MBTune to allow/avoid broadcast joins; avoid broadcasting huge tables
spark.serializerApp/clusterJava → Kryo (with registration)Faster serialization for heavy RDD pipelines; validate compatibility
spark.executor.memoryJob/cluster+1-2gMore heap per executor; reduces OOMs if partitions still too large
spark.executor.coresJob/cluster4 → 2-4Adjust to balance CPU vs memory per task; avoid oversubscription
spark.memory.fractionJob/clusterdefaultRarely 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.