Apache Spark jobs fail for a handful of recurring reasons: resource pressure, shuffle problems, serialization mistakes, missing dependencies, data source issues, or environment drift. This guide gives you a practical path to isolate the cause, change settings safely, and recover without amplifying the blast radius. You will inventory your environment, run a narrow reproduction, read the right logs, try small, reversible changes, and verify the fix before scaling out. Key goals: - Shorten time-to-diagnosis with a simple inventory and a minimal reproduction. - Make reversible, job-scoped changes first; expand only after verification. - Map observable symptoms to common Spark failure modes and apply safe fixes. ## Version and environment inventory Before changing anything, gather these facts. They close off entire classes of errors and speed every later step. - Confirm Spark, Scala, Java, and Hadoop versions. On an edge or driver node: spark-submit --version pyspark --version # if using PySpark scala -version # if using Scala APIs java -version hadoop version # if connecting to HDFS/YARN Record exact versions and build numbers. Mismatches (for example, Spark 3.3 compiled for Scala 2.12 but a jar compiled for 2.13) often surface as ClassNotFoundException or NoSuchMethodError. - Note the execution mode and topology. - Local, standalone cluster, or YARN cluster mode. - Where the driver runs (client vs cluster mode). - Data sources in play (HDFS paths, object storage buckets, Kafka topics) and their network reachability. - Locate configs and logging. - Spark defaults and overrides: - /etc/spark/conf/spark-defaults.conf - /etc/spark/conf/spark-env.sh - /etc/spark/conf/log4j2.properties (or legacy log4j.properties) - Job-level configs are preferable for quick, reversible tests: spark-submit --conf key=value . - Capture SparkConf at runtime. Within your application, print effective settings early: Scala: spark.sparkContext.getConf.getAll.foreach{ case (k, v) => println(s"CONF: $k=$v") } Python: for k, v in spark.sparkContext.getConf().getAll(): print(f"CONF: {k}={v}") - Verify data source reachability. HDFS example: hdfs dfs -ls /data/input hdfs dfs -test -e /data/input && echo exists || echo missing File system permissions example: hdfs dfs -ls /data | grep -E "^d|^-" # quick visibility into perms/owners ## Safe configuration path Make the smallest, safest change first. Prefer job-scoped settings and tiny inputs. Only promote to broader scope after you confirm the effect. Principles: - Use a narrow, measurable pilot run that reproduces the failure quickly. For example, filter to a single partition or a sample of 100k rows. - Apply overrides via spark-submit --conf or Session builder, not by editing cluster files at first. - Increase logging verbosity at the driver and executor long enough to capture the failure, then revert. Examples of safe, job-level adjustments: - Raise or lower log verbosity for this run only. Scala: spark.sparkContext.setLogLevel("INFO") // or WARN/ERROR for noise control Python: spark.sparkContext.setLogLevel("INFO") - Override executor sizing cautiously (example values). spark-submit \ --executor-memory 4g \ --conf spark.executor.cores=2 \ --conf spark.executor.instances=4 \ --class com.example.Job \ app.jar - Limit shuffle pressure during diagnosis (example values). spark-submit \ --conf spark.sql.shuffle.partitions=100 \ --conf spark.default.parallelism=100 \ --class com.example.Job \ app.jar - Add dependencies explicitly for this run. spark-submit \ --jars deps/udf-lib.jar,libs/metrics.jar \ --py-files deps/util.zip \ --class com.example.Job \ app.jar Promotions after verification: - If a setting consistently fixes a class of failures, codify it in your job packaging or spark-defaults.conf. - Avoid cluster-wide changes until multiple jobs have validated the new baseline. ## Verification and diagnostics Use the smallest dataset that still fails to speed iteration. Observe from three angles: logs, Spark UI, and environment validation. Quick access to logs and UI:
ModeDriver logsExecutor logsSpark UI
Standalone$SPARK_HOME/logs on master/worker nodes$SPARK_HOME/logs on worker nodeshttp://driver-host:4040 while app runs
YARNyarn logs -applicationId <app_id>yarn logs -applicationId <app_id>Spark History Server if enabled
- Identify the application. - From spark-submit output, capture the applicationId (YARN) or application name. - If lost, scan recent apps by time in the Spark History Server and match your job name and start time. - Pull logs with relevant filters. yarn logs -applicationId | grep -i -E "exception|error|killed|memory|fetchfailed|timeout" -n - Capture the failure signature. Common signatures: - java.lang.OutOfMemoryError, GC overhead limit exceeded - org.apache.spark.shuffle.FetchFailedException - Task not serializable - ClassNotFoundException or NoSuchMethodError - Container killed by YARN for exceeding memory limit - Py4JJavaError with nested Java stack trace (read the root cause near "Caused by:") - Inspect the Spark UI (Jobs -> Stages -> Tasks). - Skew indicator: one or a few tasks run much longer or process far more input. - Shuffle read size spikes are visible in the Stages detail. - Failed/killed tasks count, plus error messages per task. - Validate inputs and outputs exist and are writable. hdfs dfs -ls /data/output_tmp && echo ok || echo missing hdfs dfs -test -w /data/output_tmp && echo writable || echo not_writable - Reproduce minimally. If the job reads from a large table, select a small, representative subset: Scala: val sampleDf = spark.read.parquet("/data/input").limit(100000) sampleDf.repartition(16).groupBy("key").count().collect() Python: sample_df = spark.read.parquet("/data/input").limit(100000) sample_df.repartition(16).groupBy("key").count().collect() Expect to see the same failure class (for example, FetchFailedException) faster. If the failure disappears, the root cause may be scale-related (skew, memory pressure, or timeouts). ## Failure modes and recovery This section maps high-frequency Spark errors to likely causes, safe fixes, verification steps, and rollback guidance. ### 1) OutOfMemoryError or GC overhead limit exceeded Symptoms: - Driver or executors die with OutOfMemoryError. - YARN message: Container killed by YARN for exceeding memory limit. Likely causes: - Partitions too large; wide transformations materialize large shuffles. - DataFrame actions collect too much to the driver. - Caching large unfiltered tables without enough memory. Safe fixes (apply incrementally): - Reduce partition size; increase partition count for shuffles. - Avoid collect() to driver on large datasets; use take(n) or show(n, truncate=false) sparingly. - Increase executor memory and overhead modestly; check container limits. Example adjustments (example values): spark-submit \ --conf spark.sql.shuffle.partitions=400 \ --conf spark.executor.memoryOverhead=1024 \ --executor-memory 6g \ app.jar Verification: - Spark UI shows lower shuffle spill and stable task memory. - No new OOM entries in executor logs. Rollback: - If cluster utilization spikes or other jobs starve, revert memory increases and prioritize partitioning fixes. ### 2) FetchFailedException (shuffle fetch failed) Symptoms: - Stages fail repeatedly with FetchFailedException. - Tasks retry many times; eventually the stage aborts. Likely causes: - Lost executor removed blocks mid-stage. - Shuffle service not serving blocks reliably. - Network instability or timeouts fetching large blocks. - Severe skew concentrating shuffle data on few executors. Safe fixes: - Increase shuffle retry and timeout settings (example values): spark-submit \ --conf spark.shuffle.io.maxRetries=10 \ --conf spark.shuffle.io.retryWait=5s \ --conf spark.network.timeout=600s \ app.jar - Enable and verify external shuffle service when using dynamic allocation on YARN or standalone: # job-level --conf spark.shuffle.service.enabled=true \ --conf spark.dynamicAllocation.enabled=true - Reduce skew by increasing shuffle partitions and pre-aggregating. Verification: - Stages progress without repeated fetch failures; fewer lost tasks. - UI shows more even task durations after skew mitigation. Rollback: - If longer timeouts just hide network issues, scale back and investigate the failing node; temporarily exclude flaky hosts at the resource manager level if supported. ### 3) Task not serializable Symptoms: - Failure shortly after job start. - Stack trace mentions java.io.NotSerializableException. Likely causes: - Closures capture non-serializable objects (for example, database clients, loggers, SparkSession) inside map/flatMap. Safe fixes: - Move non-serializable objects out of closures; pass primitive parameters. - Use broadcast variables for large read-only data structures. Scala example fix: val bcConfig = spark.sparkContext.broadcast(heavyConfig) rdd.map(x => compute(x, bcConfig.value)) Verification: - Stage runs; no NotSerializableException in logs. Rollback: - None needed beyond reverting code changes if behavior regresses; tests should cover closure boundaries. ### 4) ClassNotFoundException / NoSuchMethodError Symptoms: - Failures at runtime when accessing classes or UDFs. Likely causes: - Missing jars or wrong Scala binary version. - Dependency shaded under different package or conflict on classpath. Safe fixes: - Supply all needed jars at submit time: spark-submit --jars libs/udfs.jar,libs/dep.jar --class com.example.Job app.jar - Ensure Scala binary version matches Spark build (for example, 2.12). - For PySpark, pass zip files for modules: spark-submit --py-files deps/util.zip job.py Verification: - The failing class loads; UDF registration succeeds. Rollback: - If a new jar introduced conflicts, remove it and isolate with shading in a later build. ### 5) Data skew causing long tails and timeouts Symptoms: - 1% of tasks run 10-100x longer. - Shuffle read size spikes on few tasks. Likely causes: - Highly skewed keys in groupBy/join. Safe fixes: - Increase shuffle partitions (example: 200 -> 800) to reduce per-task payload. - Apply pre-aggregation (map-side combine) before wide operations. - Salt keys for extreme skew; e.g., duplicate keys with a small random suffix to spread load, then aggregate results. Example adjustment (example values): spark-submit --conf spark.sql.shuffle.partitions=800 app.jar Verification: - Task durations are more uniform; no single task dominates stage time. Rollback: - If higher partition counts increase overhead, back down to a balanced value and combine with pre-aggregation. ### 6) Py4JJavaError and Python environment mismatches Symptoms: - Py4JJavaError wrapping a Java stack trace. - Errors about Python version mismatch or missing modules on executors. Likely causes: - Different Python versions on driver and executors. - Missing PySpark dependencies on workers. Safe fixes: - Set Python interpreter consistently: export PYSPARK_PYTHON=python3 export PYSPARK_DRIVER_PYTHON=python3 - Ship Python modules with --py-files and import them in the job. Verification: - Module imports succeed on executors; errors disappear. Rollback: - Revert interpreter changes if they break other jobs; scope env changes to the submit shell or wrapper script. ### 7) HDFS path or permission errors Symptoms: - FileNotFoundException or AccessControlException. Likely causes: - Wrong path, missing partition directory, or insufficient permissions. Safe fixes: - Validate with HDFS commands: hdfs dfs -ls /data/input/date=2024-01-01 - Request or set correct permissions; avoid running as a superuser for routine jobs. Verification: - Paths resolve; writes succeed to a temporary output path. Rollback: - None, aside from reverting any permission changes not aligned with policy. ### 8) Hanging jobs and stalled progress Symptoms: - No stage progress; executors appear idle. Likely causes: - Insufficient resources in the queue. - Deadlocks on external systems (for example, slow metastore or remote file system). - Very long locality wait delaying task scheduling. Safe fixes: - Reduce spark.locality.wait to move work without waiting for ideal locality (example value): --conf spark.locality.wait=1s - Right-size executor instances; too-large executors cause poor slot utilization. Verification: - Stages begin quickly; no long gaps before tasks launch. Rollback: - If network I/O spikes, restore default locality waits and investigate the storage system. ## Quick error-to-action map Use this compact map during an incident to accelerate first actions.
Error signatureLikely causeWhat to checkSafe first fix
OutOfMemoryErrorLarge partitions, collect to driverExecutor logs; stage metrics; spillIncrease partitions; modest memory bump
FetchFailedExceptionLost blocks, timeouts, skewLost executors; shuffle retriesRaise retry/timeout; enable shuffle service
Task not serializableClosure captured non-serializableCode in map/flatMapBroadcast or refactor closures
ClassNotFoundExceptionMissing jar or version mismatchspark-submit args; Scala versionAdd --jars/--py-files; align versions
Hanging jobLocality wait, resource starvationUI shows idle executorsLower locality wait; right-size executors
## Recovery patterns and rollback Partial or failed writes can leave messy state. Favor patterns that let you confirm before going visible. - Atomic output with temp-and-rename. - Write to a temp path, then rename on success: Scala: val tmp = "/data/output_tmp/run_20240101" df.write.mode("overwrite").parquet(tmp) // after successful write import org.apache.hadoop.fs.{FileSystem, Path} val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration) fs.delete(new Path("/data/output"), true) fs.rename(new Path(tmp), new Path("/data/output")) Python: tmp = "/data/output_tmp/run_20240101" df.write.mode("overwrite").parquet(tmp) from py4j.java_gateway import java_import jconf = spark.sparkContext._jsc.hadoopConfiguration() fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(jconf) fs.delete(spark._jvm.org.apache.hadoop.fs.Path("/data/output"), True) fs.rename(spark._jvm.org.apache.hadoop.fs.Path(tmp), spark._jvm.org.apache.hadoop.fs.Path("/data/output")) Rollback: - If validation fails, delete the temp path; the visible path remains intact. - Checkpoints and incremental reruns. - Use DataFrame checkpoints to cut lineage when repeated failures stem from deep plans: spark.sparkContext.setCheckpointDir("/checkpoints/jobA") val stabilized = df.checkpoint(eager = true) - Rerun from the last good checkpoint after a transient failure. Rollback: - If a change increases runtime or cost, revert the config and reuse the prior checkpoint. - Configuration rollback. - Keep job-level configs in a submit wrapper or job definition file with a previous-known-good block you can restore instantly. Example submit file snippet (pseudo-shell): # known-good CONF=( "spark.sql.shuffle.partitions=200" "spark.network.timeout=120s" ) # experimental # CONF+=("spark.dynamicAllocation.enabled=true") ## A minimal, measurable pilot Validate every change with a pilot run you can inspect quickly. This reduces rework and ambiguity when multiple knobs move at once. - Reproduce the failure on a small dataset (for example, limit(100k) with representative keys). - Change a single setting. - Capture two measurements: runtime and error rate (0 vs 1 failure) for that stage. - If the pilot passes, scale the input 5-10x and observe again before full volume. ## Operations checklist Run this when a Spark job fails or misbehaves. Preflight (2-5 minutes): - Identify applicationId and job name. - Record Spark, Scala, Java, Hadoop versions. - Confirm execution mode (client vs cluster) and data sources involved. Logs and UI (5-10 minutes): - Fetch driver and executor logs; grep for error signatures. - Open Spark UI; note skew, shuffle sizes, and failed tasks. - Validate input/output paths exist and are writable. Classify failure (2-5 minutes): - Memory pressure: OutOfMemoryError, container killed for memory. - Shuffle issue: FetchFailedException, lost executor. - Serialization: NotSerializableException. - Dependency: ClassNotFoundException. - Environment: Py4J errors, Python mismatch. - Source/permission: FileNotFound, AccessControl. Apply safe first fix (5-15 minutes): - Memory: raise partitions; modest memory/overhead increase. - Shuffle: increase retries/timeouts; verify shuffle service; mitigate skew. - Serialization: refactor closures; broadcast configuration/data. - Dependency: add --jars/--py-files; align Scala version. - Environment: set consistent Python; ship modules. - Source: fix path/perms; validate with hdfs dfs. Pilot and verify (5-20 minutes): - Re-run with a narrow dataset. - Confirm from logs and UI that the signature is gone. - If improved, scale to full dataset. Stabilize and document (5-10 minutes): - Promote job-level overrides into code or job config after success. - Add a unit or integration test that would catch the regression next time. ## Conclusion Most Spark incidents fall into recognizable patterns you can diagnose quickly with a crisp environment inventory, the right logs, and a minimal reproduction. Make small, reversible changes first, verify the outcome on a narrow pilot, then promote fixes carefully. The tables and runbook here give you a repeatable path to isolate memory pressure, shuffle instability, serialization mistakes, dependency gaps, and environment drift, and to recover safely without collateral damage.