After reading through the documentation I do not understand how does Spark running on YARN account for Python memory consumption.
Does it count towards spark.executor.memory
, spark.executor.memoryOverhead
or where?
In particular I have a PySpark application with spark.executor.memory=25G
, spark.executor.cores=4
and I encounter frequent Container killed by YARN for exceeding memory limits. errors when running a map
on an RDD. It operates on a fairly large amount of complex Python objects so it is expected to take up some non-trivial amount of memory but not 25GB. How should I configure the different memory variables for use with heavy Python code?
In cluster mode, the Spark driver runs inside an application master process which is managed by YARN on the cluster, and the client can go away after initiating the application. In client mode, the driver runs in the client process, and the application master is only used for requesting resources from YARN.
Memory usage in Spark largely falls under one of two categories: execution and storage. Execution memory refers to that used for computation in shuffles, joins, sorts and aggregations, while storage memory refers to that used for caching and propagating internal data across the cluster.
Memory overhead is the amount of off-heap memory allocated to each executor. By default, memory overhead is set to either 10% of executor memory or 384, whichever is higher.
Spark on YARN Spark uses two key components – a distributed file storage system, and a scheduler to manage workloads. Typically, Spark would be run with HDFS for storage, and with either YARN (Yet Another Resource Manager) or Mesos, two of the most common resource managers.
I'd try to increase memory to spark.python.worker.memory
default (512m) because of heavy Python code and this property value does not count in spark.executor.memory
.
Amount of memory to use per python worker process during aggregation, in the same format as JVM memory strings (e.g. 512m, 2g). If the memory used during aggregation goes above this amount, it will spill the data into disks. link
MEMORY_OVERHEAD_FRACTION = 0.10 MEMORY_OVERHEAD_MINIMUM = 384 val executorMemoryOverhead = max(MEMORY_OVERHEAD_FRACTION * ${spark.executor.memory}, MEMORY_OVERHEAD_MINIMUM))
The property is spark.{yarn|mesos}.executor.memoryOverhead
for YARN and Mesos.
YARN kills the processes which are taking more memory than they requested which is sum of executorMemoryOverhead
and executorMemory
.
In given image python processes in worker uses
spark.python.worker.memory
, thenspark.yarn.executor.memoryOverhead
+spark.executor.memory
is specific JVM.Image credits
Additional resource Apache mailing thread
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With