Troubleshooting Out-of-Memory or Slow Execution – spark
Title: Memory Overhead and OOM in Spark - Tuning Executor Settings
Category: Troubleshooting → Spark
Applies To: Apache Spark 2.x, 3.x (running on YARN, Mesos, or Standalone)
Issue Summary
Spark applications either fail with OutOfMemoryError (OOM) exceptions, causing tasks or executors to crash, or they run significantly slower than expected, leading to prolonged job completion times and inefficient resource utilization. These issues often arise from misconfigurations, inefficient code, or data characteristics not handled optimally by Spark.
Possible Cause(s):
Resource Constraints:
Insufficient Executor/Driver Memory: The allocated memory for JVM heap is not enough to hold data or intermediate results.
Insufficient Memory Overhead: Non-heap memory usage (JVM overhead, off-heap data) exceeds the configured overhead, leading to container kills by the resource manager (e.g., YARN).
Insufficient Cores: Not enough CPU power for computation, leading to tasks queueing or slow processing.
Data Skew: One or more partitions contain a disproportionately large amount of data, causing tasks processing these partitions to become bottlenecks (run slow, or OOM).
Inefficient Shuffles: Excessive data movement between executors, large shuffle sizes, or network/disk I/O bottlenecks during shuffle operations.
Inefficient Code/Algorithm:
Using RDD APIs where DataFrame/Dataset APIs with Catalyst Optimizer would be more efficient.
Expensive operations inside User-Defined Functions (UDFs).
Suboptimal join strategies (e.g., hash join on very large tables).
Unnecessary data materialization or caching.
Small Files Problem: Reading many small files leads to high I/O overhead and excessive task scheduling.
Incorrect Parallelism:
Too Few Partitions: Not enough parallelism to utilize all cluster resources, leading to long-running tasks.
Too Many Partitions: Excessive overhead due to task scheduling, context switching, and small data chunks.
Garbage Collection (GC) Overhead: The JVM spends too much time performing garbage collection, pausing application threads and reducing effective throughput.
Serialization Issues: Inefficient data serialization/deserialization leading to larger data transfers and higher CPU usage.
Data Locality Issues: Tasks running on executors that are far from the data blocks they need, leading to increased network I/O.
Step-by-Step Resolution / Debugging Flow
Initial Triage (Spark UI & Logs):
Access Spark UI: During or after job execution, navigate to http://<driver_hostname>:4040 (or through YARN UI link).
Jobs Tab: Check the status of the job, overall duration.
Stages Tab: This is critical. Identify failed stages. Look at Duration, Input Size/Records, Shuffle Read/Write, and GC Time for each stage.
Long Duration on a Few Tasks: Often indicates data skew.
High GC Time / Long GC Pauses: Points to JVM memory issues.
Large Shuffle Read/Write: Indicates heavy shuffling.
Executors Tab: Monitor allocated memory, storage memory, active tasks, failed tasks, and GC activity per executor. Look for memory spikes or executors running out of memory.
Review Logs:
Driver Logs: (spark-submit output, yarn logs -applicationId <app_id>) for overall application errors, driver OOMs, and high-level execution issues.
Executor Logs: (via YARN UI Containers tab, or yarn logs -applicationId <app_id> -containerId <container_id>) for task-specific errors, OutOfMemoryError stack traces, and details of container kills.
Key phrases to search for: OutOfMemoryError, Container killed by YARN, Exception, Task failed, GC overhead limit exceeded, long GC pause.
Diagnose and Resolve Out-of-Memory (OOM) Errors:
java.lang.OutOfMemoryError: Java heap space (in Executor/Driver logs):
Action: Increase spark.executor.memory (for executors) or spark.driver.memory (for driver).
Container killed by YARN for exceeding memory limits (in YARN NodeManager logs, but NO Java OOM in Spark logs):
Action: Increase spark.executor.memoryOverhead (for executors) or spark.driver.memoryOverhead (for driver). This indicates non-heap memory usage is too high.
If OOM/Kill is on only a few tasks/executors: Likely data skew. Proceed to step 6.
JVM GC Overhead: See step 8.
Optimize Resource Allocation (Executor Sizing):
spark.executor.memory: Adjust based on OOM diagnosis and workload. For I/O-bound jobs, prioritize more executors with less memory. For compute/cache-bound jobs, fewer, larger executors.
spark.executor.cores: A balance of 3-5 cores per executor is often optimal. Too many can lead to I/O contention; too few might under-utilize CPU.
spark.executor.instances / --num-executors: Adjust to utilize available cluster cores. Set if spark.dynamicAllocation.enabled is false.
YARN Limits: Ensure spark.executor.memory + spark.executor.memoryOverhead does not exceed yarn.scheduler.maximum-allocation-mb.
Tune Parallelism (spark.default.parallelism, spark.sql.shuffle.partitions):
Too Few Partitions (long tasks): Increase partitions to utilize all cores. Aim for 2-4x the total number of cores in the cluster (Total Cores = spark.executor.instances * spark.executor.cores).
Too Many Partitions (high overhead): Reduce partitions. This helps when tasks are very small, leading to excessive scheduling overhead.
Address Data Skew:
Identify: Look for highly variable task durations or large shuffle read/write sizes in the Spark UI.
Strategies:
Spark SQL Adaptive Query Execution (AQE): Enable spark.sql.adaptive.enabled=true (Spark 3.x+). AQE can dynamically handle skew in joins and aggregations by re-optimizing query plans at runtime.
Salting: For joins on skewed keys, introduce a random suffix/prefix to the skewed key in both DataFrames before joining, then remove it.
Filter out Skewed Keys: If a few keys are extremely skewed and not critical, consider processing them separately or filtering them out.
repartition(): Use df.repartition(numPartitions, col("skewed_key")) to redistribute data before a shuffle.
Minimize and Optimize Shuffles:
Enable External Shuffle Service: Set spark.shuffle.service.enabled=true in spark-defaults.conf. This decouples shuffle file lifespan from executor lifespan, improving stability and enabling dynamic allocation.
Broadcast Joins: For joining a large DataFrame with a small one, use spark.sql.autoBroadcastJoinThreshold (set to 50MB-200MB) or org.apache.spark.sql.functions.broadcast() hint to avoid shuffling the large table.
coalesce() vs. repartition(): Use coalesce() to reduce partitions without a full shuffle when possible (e.g., before writing output). Use repartition() when a full shuffle is necessary for re-distribution.
Tune Shuffle Buffers: spark.reducer.maxSizeInFlight (default 48MB) can be increased for higher network throughput.
Improve Code Efficiency & Data Formats:
Use DataFrame/Dataset APIs: Prefer these over RDDs as they leverage the Catalyst Optimizer for query plan optimization, including predicate pushdown, column pruning, and join reordering.
Avoid Expensive UDFs: If UDFs perform complex logic or access external systems, they can be bottlenecks. Consider refactoring logic using built-in Spark functions.
Filter/Project Early: Apply filters (.filter()) and select only necessary columns (.select()) as early as possible to reduce data size.
Efficient File Formats: Store data in Parquet or ORC format in HDFS. These columnar formats are optimized for analytical queries, offering better compression and predicate pushdown.
Compression: Use efficient compression codecs like Snappy or Zstd for both intermediate and persistent data.
JVM Garbage Collection (GC) Tuning:
Enable GC Logging: Add spark.executor.extraJavaOptions="-XX:+PrintGCDetails -XX:+PrintGCDateStamps" to spark-defaults.conf. Analyze logs for long pause times.
Use G1GC: For executors with large heaps (e.g., >6GB), set spark.executor.extraJavaOptions="-XX:+UseG1GC". This collector aims for more predictable and shorter pause times.
Serialization Optimization:
Use Kryo: Set spark.serializer=org.apache.spark.serializer.KryoSerializer. Kryo is significantly faster and more compact than Java serialization.
Register Custom Classes: For custom classes, register them with Kryo for further performance gains.
spark.kryoserializer.buffer.max: Increase if you encounter "buffer overflow" errors with Kryo for large objects.
Improve Data Locality:
Ensure that enough Spark executors are deployed on the same nodes as your HDFS DataNodes.
Monitor "Locality Level" in the Spark UI tasks tab. Aim for PROCESS_LOCAL or NODE_LOCAL. If tasks are consistently RACK_LOCAL or ANY, it indicates poor data locality, leading to higher network I/O.
Leverage Dynamic Allocation:
If your workload fluctuates or you have many concurrent small jobs, enable spark.dynamicAllocation.enabled=true (requires external shuffle service).
Carefully set spark.dynamicAllocation.minExecutors, spark.dynamicAllocation.maxExecutors, and spark.dynamicAllocation.initialExecutors to balance resource elasticity with startup time.