Spark on Yarn: Where Have All the Memory Gone?

2015/01/08 by wdong | Uncategorizedinhadoop, Spark, Yarn
原文：http://www.wdong.org/wordpress/blog/2015/01/08/spark-on-yarn-where-have-all-my-memory-gone/

Spark on Yarn: Where Have All the Memory Gone?
Efficient processing of big data, especially with Spark,is really all about how much memory one can afford, or how efficient use onecan make of the limited amount of available memory. Efficient memoryutilization, however, is not what one can
take for granted with defaultconfiguration shipped with Spark and Yarn. Rather, it takes very carefulprovisioning and tuning to get as much as possible from the bare metal. In thispost I’ll demonstrate a case when not-so-careful configuration of Spark on Yarnleads
to poor memory utilization for caching, explain the math that leads toall the observed numbers, and give some tips on parameter tuning to address theproblem.
A little bit background first. I’m not working with oneof those big names, and do not have thousands of machines at my disposal. Mygroup has less than ten for data crunching. Months of experimental usage ofSpark in a standalone configuration
showed very promising results, and we wantto use those few machines for both production and development. That is, we wantto be able to run two instances of Spark apps in parallel. Since we alreadyhave Hadoop, Spark on Yarn seems to be a natural choice. It
is not difficult atall to setup Spark on Yarn, but I quickly found that I was not able to fire upthe second instance of Spark because I was, as seen by Yarn, out of memory.I’ll use the following simplified one-machine setup to demonstrate the problemI’ve seen.
1.Demonstration of the Problem
This demo was run on a desktop with 64g memory. I usedthe following setting:

1	yarn.nodemanager.resource.memory-mb = 49152 # 48G
2	yarn.scheduler.maximum-allocation-mb = 24576 # 24G

3	SPARK_EXECUTOR_INSTANCES=1
4	SPARK_EXECUTOR_MEMORY=18G

5	SPARK_DRIVER_MEMORY=4G

The Yarn parameters went into yarn-site.xml, and theSpark ones in spark-env.sh. I didn’t set any other memory-related parameters.
So the total memory allocated to Yarn was 48G, with 24Gmaximum for one app. Spark should use 18+4 = 22G memory, which is below the 24Gcap. So I should be able to run two Spark apps in parallel.
Following are the numbers I got from all the logs and WebUIs when I actually fired up one Spark app.

(Yarn) Memory Total: 48G
(Yarn) Memory Used: 25G
(Yarn) Container 1, TotalMemoryNeeded: 5120M
(Yarn) Container 2, TotalMemoryNeeded: 20480M
(Spark) Driver memory requirement: 4480 MB memory including 384 MB overhead (From output of Spark-Shell)

(Spark) Driver available memory to App: 2.1G
(Spark) Executor available memory to App: 9.3G

Below are the relevant screen shots.








So here are the problems that I see with the driver:

I’ve configured Spark driver to use 4G, and Spark asked Yarn for 4G plus an overhead of 384MB.

What reflected in Yarn is that the driver has used 5G.
What’s really available in the driver’s block manager is only 2.1G.

One has to understand that Spark has to reserve a portionof memory for code execution and cannot give everything to the block manager(the cache), but still,
WHEREHAVE ALL THE MEMORY GONE???
2. The MathBehind
Rule 1. Yarn always rounds up memory requirement tomultiples of yarn.scheduler.minimum-allocation-mb, which by default is 1024 or1GB. That’s why the driver’s requirement of 4G+384M showed up as 5G in Yarn.The parameter yarn.scheduler.minimum-allocation-mb
is really“minimum-allocation-unit-mb”. This can be easily verified by setting theparameter to a prime number, such as 97, and see Yarn allocate by multiples ofthe number.
Rule 2. Spark adds an overhead toSPARK_EXECUTOR_MEMORY/SPARK_DRIVER_MEMORY before asking Yarn for the amount.The rule of overhead is the same for both executor and driver, which is

1	//yarn/common/src/main/scala/org/apache/spark/deploy/yarn/YarnSparkHadoopUtil.scala
2	val MEMORY_OVERHEAD_FACTOR = 0.07

3	val MEMORY_OVERHEAD_MIN = 384
4

5	//yarn/common/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
6	protected val memoryOverhead: Int = sparkConf.getInt("spark.yarn.executor.memoryOverhead",

7	math.max((MEMORY_OVERHEAD_FACTOR * executorMemory).toInt, MEMORY_OVERHEAD_MIN))
8	......

9	val totalExecutorMemory = executorMemory + memoryOverhead
10	numPendingAllocate.addAndGet(missing)

11	logInfo(s"Will allocate $missing executor containers, each with $totalExecutorMemory MB " +
12	s"memory including $memoryOverhead MB overhead")

This overhead is necessary because when a JVM program isallowed certain amount of memory (by -Xmx), the overall memory usage of JVMcould be more than that, and Yarn literally kills programs which uses morememory than allowed (with complicated
rules). One can only adjust the twomagic numbers by modifying the source.
The above two rules determine how the configuredSPARK_XXX_MEMORY finally show up in Yarn.
Rule 3. How much memory does driver/executor see.
One limits the maximal heap memory of JVM by the option“-Xmx”. Part of the specified memory get used by the Scala runtime and othersystem component, and what a Scala program sees is less then the specifiedamount. This can be illustrated with
the following example.

1	$ scala -J-Xmx4g
2	Welcome to Scala version 2.10.3 (OpenJDK 64-Bit Server VM, Java 1.7.0_51).

3	Type in expressions to have them evaluated.
4	Type :help for more information.

5
6	scala> Runtime.getRuntime.maxMemory

7	res0: Long = 3817865216
8	scala>

The runtime eats about 455M. (The above process has an RSS of140.3M in Linux, so a big portion of the 455M is more like being reserved
thanactually being used.)
The Spark driver is allocated the configured 4G by JVMoptions. This can be verified by running the following from inside the Sparkshell.

1	scala>
2

3	scala> import java.lang.management.ManagementFactory
4	import java.lang.management.ManagementFactory

5
6	scala> ManagementFactory.getRuntimeMXBean.getInputArguments

7	res0: java.util.List[String] = [-XX:MaxPermSize=128m, -Djava.library.path=/home/hadoop/hadoop-2.4.1/lib/native, -Xms4G, -Xmx4G]
8

9	scala> Runtime.getRuntime.maxMemory
10	res1: Long = 4116709376

11
12	scala>

Rule 4. How Spark determines maximal usable memory

1	//core/src/main/scala/org/apache/spark/storage/BlockManager.scala
2	/** Return the total amount of storage memory available. */

3	private def getMaxMemory(conf: SparkConf): Long = {
4	val memoryFraction = conf.getDouble("spark.storage.memoryFraction", 0.6)

5	val safetyFraction = conf.getDouble("spark.storage.safetyFraction", 0.9)
6	(Runtime.getRuntime.maxMemory * memoryFraction * safetyFraction).toLong

7

}

We have 4116709376 * 0.6 * 0.9 = 2.07G. That is where the2.1G value comes from. The maximal available memory of executor is derived inthe same way.
Overall, the following two formulas guide memoryallocation:

What’s seen by Yarn: (SPECIFIED_MEMORY + OVERHEAD) round up to multiples of minimum-allocation-mb , with OVERHEAD = max(SPECIFIED_MEMORY * 0.07, 384M)

What’s usable for cache: (SPECIFIED_MEMORY – MEMORY_USED_BY_RUNTIME) * spark.storage.memoryFraction * spark.storage.safetyFraction

3. TuningSuggestions
We see that the root of memory under utilization is theover-provisioning done in almost each step. Even if a process really reachesthe configured memory cap, it’s unlikely it will keep using that amount ofmemory all the time. Because Yarn actually
kills a process when it exceeds thememory cap, we have to keep SPARK_XXX_MEMORY big enough. It is also verydifficult to determine the actual amount of memory gets used by Spark codeexecution, so dealing with spark.storage.memoryFraction is tricky. But if oneis
sure that it’s unlikely for the overall memory consumption of parallel Sparkapps to exceed the physical memory, the easiest way to improve memoryutilization is to counter the over-provisioning with overcommitment. That is,to set the
Yarn parameteryarn.nodemanager.resource.memory-mb to MORE THAN THE AVAILABLE PHYSICAL MEMORY (luckilyYarn does not check that). It also helps a little bit to setyarn.scheduler.minimum-allocation-mb to a small value like 100M, so an app doesnot get much
more that what it asks for.
wdong