Spark off heap memory expanding with caching

Question

I am struggling to make my Spark program avoid exceeding YARN memory limits (on executors).
The "known" error which I get is:

Container killed by YARN for exceeding memory limits.  3.8 GB of 3.6 GB physical memory used.

Instead of just raising the executor memory, executor memory overhead or tune my resources or partitions, I'de like to know why my off heap memory is expanding at all.

I'm using pyspark v2.4.4, as far as I understand YARN memory overhead (for executors) is any off heap memory allocated by my Spark program (outside JVM).

The only additional "non-related" off heap memory I'm aware of is the Python memory, which is not not part of the memory overhead by Spark documentation:

The maximum memory size of container to running executor is determined by the sum of spark.executor.memoryOverhead, spark.executor.memory, spark.memory.offHeap.size and spark.executor.pyspark.memory.

I'm using heavy caching in my program; disabling the persist call somehow remedies the memory overhead issue (less executors die and the program can finish eventually), but since this data should be managed only inside the JVM/disk against the execution memory with UnifiedMemoryManager, it shouldn't use any off heap memory.

Since off heap mode is disabled by default (spark.memory.offHeap.use), which scenarios may cause expanding memory overhead, and why disabling caching in my program helps to reduce the overhead size?

Edit
Using larger executors (twice memory and cores) remedies this as well.
It's makes sense that larger memory on JVM means larger overhead (10% when larger than 386m), but still we are using 2 cores (i.e 2 tasks) on this JVM and I don't understand on which situations memory overhead expands.

Answer 1

Spark may use off-heap memory during shuffle and cache block transfers; even if spark.memory.offHeap.use=false.

This problem is also referenced in Spark Summit 2016 (minute 4:05).

Spark 3.0.0 and Above

This behavior can be disabled since Spark 3.0.0 with spark.network.io.preferDirectBufs=false.

Spark configuration elaborates a short explanation for this:

If enabled then off-heap buffer allocations are preferred by the shared allocators. Off-heap buffers are used to reduce garbage collection during shuffle and cache block transfer. For environments where off-heap memory is tightly limited, users may wish to turn this off to force all allocations to be on-heap.

Spark < 3.0.0

For versions lower than 3.0.0, using larger executors with modified higher memory overhead significantly remedies this problem while keeping the same allocated memory per executor and same overall resources consumption by your Spark job.
For example:

Before:

spark.executor.cores=1
spark.executor.memory=2g

Total container memory: 2g + 384m (minimum overhead) = 2.375g
Executor memory per core: 2.375g
JVM memory per core: 2g

After:

spark.executor.cores=4
spark.executor.memory=6g
spark.executor.memoryOverhead=3g

Total container memory: 6g + 3g = 9g
Executor memory per core: 2.25g
JVM memory per core: 1.5g

In case your containers are being killed by YARN (like in this Q) the purposed change should help, but note that that's a trade off with other things:
The overall JVM memory per core is lower, so you are more opened to memory bottlenecks in User Memory (mostly objects you create in the executors) and Spark Memory (execution memory and storage memory).
Exceeded Spark Memory is generally spilled to disk (with additional non-relevant complexities) thus sacrifice performance and lack of User Memory may lead to OutOfMemory errors in your executors, so be careful.