Spark Out of Memory Error

The JVM documentation says the Java heap space message means an object could not be allocated in the heap, and that it doesn't necessarily imply a leak. In Spark the usual reason is that 1 task needed more memory than its share. Spark's tuning guide describes it: the working set of a task, such as a reduce task in groupByKey, was too large, because shuffle operations build a hash table inside each task.

The simplest fix, in the same guide, is more parallelism so each task's input is smaller. The defaults make the arithmetic concrete. A 100 GiB shuffle split into the default 200 partitions puts 512 MiB in each, and in-memory objects take more room than the bytes on disk (the guide says a decompressed block is often 2 or 3 times its size). At the 64 MiB advisory size that adaptive query execution targets, the same shuffle is 1,600 partitions. The guide recommends 2 to 3 tasks per CPU core, and notes Spark runs tasks as short as 200 ms efficiently, so many small tasks cost little.

Adaptive query execution has been on by default since 3.2.0, according to the SQL tuning documentation, and it coalesces small shuffle partitions and splits skewed ones. A join partition counts as skewed when it's over 5 times the median partition and over 256 MiB. If 1 task fails while its siblings finish, compare its input size with the median in the Spark UI: a gap that size is skew, and no executor size fixes it for long. The PySpark groupBy guide covers the shuffle and skew patterns.

GC overhead limit exceeded is the same condition seen from the collector. The JVM throws it after spending about 98% of its time collecting while recovering less than 2% of the heap, for 5 consecutive collections. Treat it as a heap that is full of live data, and fix the data per task before reaching for JVM flags. The tuning guide adds that collection cost grows with the number of Java objects, so serialized caching is the first thing to try when GC is the problem.

The driver holds the query plan and the task bookkeeping, and it receives everything an action returns. Its default heap is 1g. In client mode you cannot raise it through SparkConf in your application, because the driver JVM has already started; give it on the command line as the driver memory option of spark-submit or set it in the properties file.

collect and toPandas bring every partition's result to 1 JVM, and so do similar actions. spark.driver.maxResultSize (default 1g) caps the total serialized size of results, and a job over the limit is aborted with a message that starts "Total size of serialized results of N tasks". That abort is the limit working, not a crash. The documentation warns that a high limit may cause out-of-memory errors in the driver, so raising it moves the failure rather than removing it.

The better fix is to stop returning the data: aggregate first, write the result to storage and read a sample, or call limit before collect. Broadcasts need the same care. A table under spark.sql.autoBroadcastJoinThreshold (10 MiB by default, -1 to disable) is sent to all worker nodes, so raising the threshold spends memory on every executor heap, not only the driver.

ExecutorLostFailure means the driver lost an executor and the tasks on it failed. It names no cause, so the cause is in the executor's log or the cluster manager's diagnostic. The most common memory cause is a container killed for going past its limit, and it leaves no OutOfMemoryError in the executor log because the process was killed from outside.

On YARN, Spark's source builds the message Container killed by YARN for exceeding physical memory limits, followed by the sizes and a suggestion to boost spark.executor.memoryOverhead. A virtual memory variant also names yarn.nodemanager.vmem-pmem-ratio and yarn.nodemanager.vmem-check-enabled. On Kubernetes the pod memory request and limit are the sum of the memory and overhead settings, and Spark's documentation describes exit code 137 as a SIGKILL that may be a container OOM, with exit code 52 for a JVM OOM.

The container limit is the sum of 4 settings: spark.executor.memory and spark.executor.memoryOverhead, then spark.memory.offHeap.size and spark.executor.pyspark.memory. If you do not set the PySpark memory, Python workers count against the overhead, and Spark doesn't limit their use. That's why PySpark jobs hit container kills more often: the default overhead is sized for a JVM. Spark's own default overhead factor for non-JVM jobs on Kubernetes is 0.40 for this reason, because such tasks commonly fail with "Memory Overhead Exceeded" errors. Set spark.executor.memoryOverhead directly and the factor is ignored.

Not every lost executor is a memory problem. An executor that stops answering for longer than spark.network.timeout (120s by default) is removed, and heartbeats go out every 10s. A node failure or a preemption also loses executors. Spark source separates these cases: a loss unrelated to the task is not counted toward the 4 failures (spark.task.maxFailures) that fail a job, but a memory kill by YARN is treated as the application's fault and counts. Repeated executor loss is limited too, by spark.executor.maxNumFailures, which the configuration documentation says takes effect on YARN and Kubernetes.

Interviewers ask this as a diagnosis: a job fails on 1 task out of 2,000, what do you check and in what order. The strong answer separates heap from container, reads the message, finds skew by comparing the failed task with the median, and raises partitions before memory. Practice saying that sequence aloud with the PySpark interview questions, then work the PySpark practice problems that ask for shuffle and skew handling.

The same reasoning shows up in tuning questions, covered by the Spark optimization interview questions, and in the wider data engineer interview prep plan. Those questions expect you to say why a setting helps, not just what it is called.