UnifiedMemoryManager takes the following to be created:
UnifiedMemoryManager requires that:
apply( conf: SparkConf, numCores: Int): UnifiedMemoryManager
apply computes the size of the on-heap storage region which is a fraction of the maximum heap memory based on spark.memory.storageFraction configuration property (default:
In the end,
apply creates a UnifiedMemoryManager (with the given and computed values).
apply is used when
SparkEnv utility is used to create a SparkEnv (for the driver and executors).
getMaxMemory( conf: SparkConf): Long
getMaxMemory calculates the maximum memory to use for execution and storage.
// local mode with --conf spark.driver.memory=2g scala> sc.getConf.getSizeAsBytes("spark.driver.memory") res0: Long = 2147483648 scala> val systemMemory = Runtime.getRuntime.maxMemory // fixed amount of memory for non-storage, non-execution purposes val reservedMemory = 300 * 1024 * 1024 // minimum system memory required val minSystemMemory = (reservedMemory * 1.5).ceil.toLong val usableMemory = systemMemory - reservedMemory val memoryFraction = sc.getConf.getDouble("spark.memory.fraction", 0.6) scala> val maxMemory = (usableMemory * memoryFraction).toLong maxMemory: Long = 956615884 import org.apache.spark.network.util.JavaUtils scala> JavaUtils.byteStringAsMb(maxMemory + "b") res1: Long = 912
getMaxMemory reads the maximum amount of memory that the Java virtual machine will attempt to use and decrements it by reserved system memory (for non-storage and non-execution purposes).
getMaxMemory makes sure that the following requirements are met:
System memory is not smaller than about 1,5 of the reserved system memory.
spark.executor.memory is not smaller than about 1,5 of the reserved system memory.
getMaxMemory returns spark.memory.fraction of the maximum amount of memory for the JVM (minus the reserved system memory).
|FIXME omnigraffle it.|
acquireExecutionMemory( numBytes: Long, taskAttemptId: Long, memoryMode: MemoryMode): Long
acquireExecutionMemory varies per
acquireStorageMemory( blockId: BlockId, numBytes: Long, memoryMode: MemoryMode): Boolean
acquireStorageMemory has two modes of operation per
MemoryMode.OFF_HEAP, for execution and storage pools, and the maximum amount of memory to use.
|FIXME Where are they used?|
onHeapStorageMemoryPool, and maxOnHeapStorageMemory are used.
maxOffHeapMemory are used.
|FIXME What is the difference between them?|
It makes sure that the requested number of bytes
numBytes (for a block to store) fits the available memory. If it is not the case, you should see the following INFO message in the logs and the method returns
INFO Will not store [blockId] as the required space ([numBytes] bytes) exceeds our memory limit ([maxMemory] bytes)
If the requested number of bytes
numBytes is greater than
memoryFree in the storage pool,
acquireStorageMemory will attempt to use the free memory from the execution pool.
|The storage pool can use the free memory from the execution pool.|
It will take as much memory as required to fit
memoryFree in the execution pool (up to the whole free memory in the pool).
acquireStorageMemory requests the storage pool for
maxOnHeapStorageMemory is the difference between
maxHeapMemory of the UnifiedMemoryManager and the memory currently in use in
onHeapExecutionMemoryPool execution memory pool.