## Intro Misconfigured Spark jobs waste compute, stall pipelines, and hide performance headroom. This guide focuses on the mistakes teams make most often, how to diagnose them quickly in the Spark UI, and how to change settings safely. You will see practical spark-submit examples, a repeatable pilot plan you can run before touching production, and callouts for common interactions with Kafka, HDFS/cloud storage, Airflow, and NiFi. What you will take away: - A safe workflow to change Spark configs without surprises - The most common configuration mistakes and how to fix them - Concrete examples for batch and Structured Streaming - A short, repeatable troubleshooting path you can apply across jobs ## A safe change workflow Use this loop for any configuration change: - Baseline - Capture job version, full config (sc.getConf.getAll), input size, runtime, and failure symptoms. - Save event logs (spark.eventLog.enabled=true) or History Server links from a typical run. - Hypothesis - Pick one change that targets a single measurable bottleneck (e.g., skew, shuffle size, GC time). - Local pilot - Test on a small, representative slice with a fixed seed or time window so results are repeatable. - Staging replay - Re-run the last successful production window with the new config. Compare runtime, stage skew, and correctness (row counts and key aggregates). - Guarded rollout - Enable for a subset of jobs or partitions. Define abort criteria (runtime +30%, error rate above baseline, OOMs). - Monitoring - Watch scheduler delay, task failures, shuffle spill, GC, IO throughput, and executor churn in the Spark UI. - Roll back fast - Keep changes in a single config block you can revert in one move. Use driver/executor logs and the History Server to refine your next hypothesis. ## Fast Spark UI diagnostics Open the UI while a job runs (or the History Server post-run) and check: - Jobs/Stages tabs: long scheduler delay vs task time, skewed tasks, shuffle read/write, spill (memory/disk), and stage retries. - SQL tab (for DataFrame/SQL): join strategies, number of shuffle partitions, adaptive coalescing and skew hints. - Executors tab: GC time, peak storage memory, task failure counts, and blacklisted executors. - Environment: effective configuration at runtime. Rule of thumb: if scheduler delay or shuffle spill dominates, fix partitioning; if GC time is high, fix memory/cores/overhead; if a few tasks run much longer, address skew. ## Common mistakes and practical fixes ### 1) Executors and cores mismatch - Symptoms: low CPU utilization or long GC pauses. - Fix: size cores and memory together. Aim for 1 task per core and 2–4 GiB heap per core for mixed workloads. - Example: spark-submit \ --conf spark.executor.instances=12 \ --conf spark.executor.cores=4 \ --conf spark.executor.memory=8g \ --conf spark.driver.memory=4g - Validate: Executors tab shows high task time vs scheduler delay; GC time is a small fraction of task time. ### 2) Memory overhead too low (especially with PySpark or wide shuffles) - Symptoms: container OOM during shuffle, Python worker OOM, or frequent executor loss. - Fix: increase non-heap overhead and Python worker memory where used. - Example: spark-submit \ --conf spark.executor.memory=6g \ --conf spark.executor.memoryOverhead=1536 \ --conf spark.executor.pyspark.memory=1g - Validate: no container OOMs in logs; shuffle spill stays within container memory limits; stable executor count. ### 3) Shuffle partitions set poorly - Symptoms: too many tiny tasks (overhead) or too few massive tasks (OOM, skew). - Fix: start with a target 128–256 MB per shuffle partition. With Adaptive Query Execution (AQE), set a reasonable upper bound and let Spark coalesce. - Example: spark-submit \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.shuffle.partitions=800 \ --conf spark.sql.adaptive.coalescePartitions.enabled=true \ --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=256m - Validate: SQL tab shows partitions coalesced; task durations cluster tightly with minimal skew. ### 4) AQE disabled or not tuned - Symptoms: large skews, unnecessary shuffles, or inefficient join strategies. - Fix: enable AQE and skew handling; tune advisory partition size. - Example: spark-submit \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.adaptive.skewJoin.enabled=true \ --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=256m - Validate: explain plan shows BroadcastHashJoin or SortMergeJoin as expected; SQL tab indicates skewed partitions split by AQE. ### 5) Dynamic allocation misconfigured - Symptoms: idle executors linger or the job starves under load. - Fix: bound the range and set a reasonable initial value. On Kubernetes, prefer shuffle tracking; on YARN, use the external shuffle service. - Example (Kubernetes or when no external shuffle service): spark-submit \ --conf spark.dynamicAllocation.enabled=true \ --conf spark.dynamicAllocation.shuffleTracking.enabled=true \ --conf spark.dynamicAllocation.initialExecutors=8 \ --conf spark.dynamicAllocation.minExecutors=4 \ --conf spark.dynamicAllocation.maxExecutors=64 - Validate: executor count tracks input size; backlog in the scheduler stays low; limited idle time post-shuffle. ### 6) Oversized broadcast joins - Symptoms: executor OOM during join or driver memory pressure when collecting broadcasted data. - Fix: cap broadcast size or disable when not beneficial; consider join hints or repartitioning. - Example: spark-submit \ --conf spark.sql.autoBroadcastJoinThreshold=64m \ --conf spark.sql.broadcastTimeout=1200 - Validate: physical plan shows expected join strategies; memory stable on driver/executors. ### 7) Serialization overhead in RDD pipelines - Symptoms: high CPU in serialization and large shuffle footprints. - Fix: prefer Kryo for custom classes and register them; consider DataFrame/SQL APIs for Tungsten encoders. - Example: spark-submit \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ --conf spark.kryo.registrationRequired=true \ --conf spark.kryo.registrator=com.acme.support.MyRegistrator - Validate: lower shuffle write/read sizes and shorter task times compared to Java serializer. ### 8) Speculative execution on write-heavy or non-idempotent sinks - Symptoms: duplicated side effects or sink conflicts (e.g., REST, JDBC upserts, external systems). - Fix: disable speculation or restrict it to long-tail tasks on read-only jobs. - Example: spark-submit --conf spark.speculation=false - Validate: no duplicate writes; runtime unchanged on stable clusters. ### 9) Small files explosion on HDFS or cloud storage - Symptoms: many tiny output files degrade directory listings and downstream scans. - Fix: partition and compact intentionally; prefer repartition over coalesce for wide datasets. - Examples: # Increase read partition size for file scanning spark.conf.set("spark.sql.files.maxPartitionBytes", "256m") # Repartition by a stable key and time, then write final_df = transformed_df.repartitionByRange(200, "event_date", "user_id") final_df.write.mode("overwrite").partitionBy("event_date").parquet(path) # Run a follow-up compaction job if needed spark.read.parquet(path).repartition(200).write.mode("overwrite").parquet(path) - Validate: fewer, larger files (e.g., 128–512 MB); faster downstream queries and listings. ### 10) Kafka Structured Streaming pressure - Symptoms: rising end-to-end latency or consumer lag that never recovers. - Fix: limit per-trigger input and right-size shuffle partitions; checkpoint reliably. - Example: stream = (spark.readStream .format("kafka") .option("kafka.bootstrap.servers", servers) .option("subscribe", topics) .option("maxOffsetsPerTrigger", 500000) .load()) spark.conf.set("spark.sql.shuffle.partitions", 200) query = (process(stream) .writeStream .option("checkpointLocation", ckpt_path) .trigger(processingTime="30 seconds") .start()) - Validate: micro-batch duration stabilizes; consumer lag remains bounded across spikes. ### 11) Network timeouts too aggressive for large shuffles - Symptoms: executor lost, fetch failures, or intermittent stage retries on big shuffles. - Fix: lengthen network timeout and heartbeat for heavy or noisy networks. - Example: spark-submit \ --conf spark.network.timeout=800s \ --conf spark.executor.heartbeatInterval=60s - Validate: reduced shuffle fetch failures; stable executor set across long stages. ### 12) Orchestrator oversubscription (Airflow/NiFi) - Symptoms: cluster overload when multiple DAG tasks or NiFi processors start at once. - Fix: apply concurrency limits in the orchestrator; size executors to fit quota. - Examples: - Airflow: use max_active_runs_per_dag, task-level pools and concurrency. - NiFi: set back pressure thresholds and concurrent tasks per processor. - Validate: acceptable queue times; no widespread task preemption or stragglers. Validation checklist - Confirm effective config at runtime using the Environment tab or sc.getConf.getAll. - Snapshot core metrics before/after: total runtime, stage/task time, spill, GC, IO throughput. - Guard correctness with record counts and key aggregates (e.g., per-day totals). - Keep one change per test to isolate effects. ## Local pilot plan Goal: prove one improvement on one job in a controlled run you can inspect locally or on a dev cluster before deployment. Scope - Choose a single batch job or one streaming window. - Limit to one datasource (e.g., one Kafka topic or one storage directory). - Improve one metric (e.g., runtime, max executor memory, skew spread). Guardrails - Fix input size and time window. - Set a short time bound (2–4 runs). - Define pass/fail thresholds and a rollback trigger. Example pilot steps - Extract a representative 5–10 GB sample (or N minutes of Kafka data) to a dev bucket. - Apply one change, such as spark.sql.shuffle.partitions=400 or enabling AQE. - Run spark-submit locally or on a dev cluster with identical code but the new conf. - Collect total runtime, stage skew metrics, spill sizes, and output correctness checks. - Decision: if runtime improves by at least 20% and correctness holds, promote to staging; otherwise roll back and try the next hypothesis. Deterministic sampling examples # File-based batch job sample = df.sample(withReplacement=False, fraction=0.1, seed=42) # Kafka time-bounded window (Structured Streaming) stream = (spark.readStream.format("kafka") .option("kafka.bootstrap.servers", servers) .option("subscribe", topic) .option("startingOffsets", "earliest") .load()) Operational tips - Store job configs next to code with clear versioning. - Print sc.getConf.toDebugString at job start for traceability. - Tag logs with a config-version label for quick diffing. ## Troubleshooting workflow Follow this path to isolate configuration issues quickly: - Identify the failure signature - OOM, long scheduler delay, skewed stages, high GC, frequent fetch failures, or slow filesystem. - Reproduce on a small sample - Keep the dataset constant; disable unrelated adaptive features only while debugging to isolate causes. - Verify effective configuration - Dump sc.getConf.getAll and compare to intended settings; capture the Environment tab. - Inspect Spark UI and logs - Jobs/SQL tabs: longest stages, skew hints, shuffle read/write sizes. - Executors: GC time, peak memory, task failures, blacklisting. - Driver/executor logs: container OOMs, serialization errors, network timeouts. - Apply the smallest viable change - Adjust one setting, rerun, and measure the same metrics. - Decide and document - If metrics improve without correctness regressions, keep the change. Otherwise, roll back and test the next hypothesis. ## Conclusion The fastest way to improve Spark reliability is to avoid a handful of common configuration mistakes and to change settings with discipline: baseline, single-change pilot, staging replay, guarded rollout, and instant rollback. Build habits that stick: keep a versioned config catalog per job, log the effective config for every run, track a short metric set (runtime, skew, GC, IO), and maintain a one-command rollback for each change. When in doubt, test locally first, measure, and prefer small, reversible steps.