4

My input dataset is about 150G. I am setting

--conf spark.cores.max=100 
--conf spark.executor.instances=20 
--conf spark.executor.memory=8G 
--conf spark.executor.cores=5 
--conf spark.driver.memory=4G

but since data is not evenly distributed across executors, I kept getting 

Container killed by YARN for exceeding memory limits. 9.0 GB of 9 GB physical memory used

here are my questions:

1. Did I not set up enough memory in the first place? I think 20 * 8G > 150G, but it's hard to make perfect distribution, so some executors will suffer
2. I think about repartition the input dataFrame, so how can I determine how many partition to set? the higher the better, or?
3. The error says "9 GB physical memory used", but i only set 8G to executor memory, where does the extra 1G come from?

Thank you!

user2628641
  • 2,035
  • 4
  • 29
  • 45
  • You are setting spark executor instances to 20. This means you will be starting 20 executors of 8G each, if I recall correctly. Can you try reducing them and then try? Reducing it to 1 will be a start. – ar7 Oct 27 '16 at 14:09
  • Hi sir, so if I reduce executor to 1, i will only have 1 executor with 8G memory, then the 150G input can't fit in – user2628641 Oct 27 '16 at 14:20
  • It could be because the yarn container was not configured to enough memory. Possible solution for the error you are facing? http://stackoverflow.com/a/33118489/6199146 and http://m.blog.csdn.net/article/details?id=50387104 – ar7 Oct 27 '16 at 14:36

2 Answers2

5

When using yarn, there is another setting that figures into how big to make the yarn container request for your executors:

spark.yarn.executor.memoryOverhead

It defaults to 0.1 * your executor memory setting. It defines how much extra overhead memory to ask for in addition to what you specify as your executor memory. Try increasing this number first.

Also, a yarn container won't give you memory of an arbitrary size. It will only return containers allocated with a memory size that is a multiple of it's minimum allocation size, which is controlled by this setting:

yarn.scheduler.minimum-allocation-mb

Setting that to a smaller number will reduce the risk of you "overshooting" the amount you asked for.

I also typically set the below key to a value larger than my desired container size to ensure that the spark request is controlling how big my executors are, instead of yarn stomping on them. This is the maximum container size yarn will give out.

nodemanager.resource.memory-mb
Ryan Widmaier
  • 7,948
  • 2
  • 30
  • 32
4

The 9GB is composed of the 8GB executor memory which you add as a parameter, spark.yarn.executor.memoryOverhead which is set to .1, so the total memory of the container is spark.yarn.executor.memoryOverhead + (spark.yarn.executor.memoryOverhead * spark.yarn.executor.memoryOverhead) which is 8GB + (.1 * 8GB) ≈ 9GB.

You could run the entire process using a single executor, but this would take ages. To understand this you need to know the notion of partitions and tasks. The number of partition is defined by your input and the actions. For example, if you read a 150gb csv from hdfs and your hdfs blocksize is 128mb, you will end up with 150 * 1024 / 128 = 1200 partitions, which maps directly to 1200 tasks in the Spark UI.

Every single tasks will be picked up by an executor. You don't need to hold all the 150gb in memory ever. For example, when you have a single executor, you obviously won't benefit from the parallel capabilities of Spark, but it will just start at the first task, process the data, and save it back to the dfs, and start working on the next task.

What you should check:

  • How big are the input partitions? Is the input file splittable at all? If a single executor has to load a massive amount of memory, it will run out of memory for sure.
  • What kind of actions are you performing? For example, if you do a join with very low cardinality, you end up with a massive partitions because all the rows with a specific value, end up in the same partitions.
  • Very expensive or inefficient actions performed? Any cartesian product etc.

Hope this helps. Happy sparking!

Community
  • 1
  • 1
Fokko Driesprong
  • 2,075
  • 19
  • 31
  • 2
    I think you wanted to say executor-memory + (spark.yarn.executor.memoryOverhead * executor-memory) which is 8GB + (.1 * 8GB) ≈ 9GB – altayseyhan Dec 14 '18 at 09:13