spark scalability: what am I doing wrong?

Question

I am processing data with spark and it works with a day worth of data (40G) but fails with OOM on a week worth of data:

import pyspark
import datetime
import operator
sc = pyspark.SparkContext()
sqc = pyspark.sql.SQLContext(sc)
sc.union([sqc.parquetFile(hour.strftime('.....'))
          .map(lambda row:(row.id, row.foo))
          for hour in myrange(beg,end,datetime.timedelta(0,3600))]) \
  .reduceByKey(operator.add).saveAsTextFile("myoutput")

The number of different IDs is less than 10k. Each ID is a smallish int. The job fails because too many executors fail with OOM. When the job succeeds (on small inputs), "myoutput" is about 100k.

what am I doing wrong?
I tried replacing saveAsTextFile with collect (because I actually want to do some slicing and dicing in python before saving), there was no difference in behavior, same failure. is this to be expected?
I used to have reduce(lambda x,y: x.union(y), [sqc.parquetFile(...)...]) instead of sc.union - which is better? Does it make any difference?

The cluster has 25 nodes with 825GB RAM and 224 cores among them.

Invocation is spark-submit --master yarn --num-executors 50 --executor-memory 5G.

A single RDD has ~140 columns and covers one hour of data, so a week is a union of 168(=7*24) RDDs.

I would start by looking at the performance UI and also keep in mind that union is not your typical union. If you want a distinct union then you will need to call distinct after the union. — Justin Pihony, Mar 20 '15 at 01:58
@JustinPihony: I know that `union` means concatenation and that is precisely what I want. What is "performance UI"? — sds, Mar 20 '15 at 02:01
Why are you invoking `--num-executors 50` if you only have 25 nodes? — Mikel Urkia, Mar 20 '15 at 10:43
How many RDDs are you unioning? it looks like you are producing a potentially huge number of RDDs and partitions — Sean Owen, Mar 20 '15 at 15:21
@SeanOwen: I have one rdd per hour; so one week of data is 7*24=168 RDDs. — sds, Mar 20 '15 at 15:23
Not too bad. I can't see anything here using significant memory so it doesn't really make sense. — Sean Owen, Mar 20 '15 at 15:26
Are individual rows very large? What about one rdd? I can't see anything here that takes non trivial memory unless something is very unusual about the input. — Sean Owen, Mar 20 '15 at 15:39
@SeanOwen: the individual RDDs have about 140 columns. How many RDDs in a union is "too much"? — sds, Mar 20 '15 at 15:45
@sds I might be incorrect, but I'd say you should use `--num-executors 25 --executor-cores N` (N being core quantity of each node) instead. — Mikel Urkia, Mar 20 '15 at 21:51
Did you try repartitioning the dataframe: sqc.parquetFile(hour.strftime('.....')).repartition(50).map ... ? Might help — pzecevic, Mar 21 '15 at 08:18

score 2 · Answer 1 · edited Sep 16 '16 at 17:25

2

It turned out that the problem was not with spark, but with yarn. The solution is to run spark with

spark-submit --conf spark.yarn.executor.memoryOverhead=1000

(or modify yarn config).

edited Sep 16 '16 at 17:25

gsamaras

71,951
46
188
305

answered Mar 27 '15 at 14:05

sds

58,617
29
161
278

score 2 · Accepted Answer · edited Sep 22 '17 at 06:03

Spark very often suffers from Out-Of-Memory errors when scaling. In these cases, fine tuning should be done by the programmer. Or recheck your code, to make sure that you don't do anything that is way too much, such as collecting all the bigdata in the driver, which is very likely to exceed the memoryOverhead limit, no matter how big you set it.

To understand what is happening you should realize when yarn decides to kill a container for exceeding memory limits. That will happen when the container goes beyond the memoryOverhead limit.

In the Scheduler you can check the Event Timeline to see what happened with the containers. If Yarn has killed a container, it will be appear red and when you hover/click over it, you will see a message like:

Container killed by YARN for exceeding memory limits. 16.9 GB of 16 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

So in that case, what you want to focus on is these configuration properties (values are examples on my cluster):

# More executor memory overhead
spark.yarn.executor.memoryOverhead          4096

# More driver memory overhead
spark.yarn.driver.memoryOverhead            8192

# Max on my nodes
#spark.executor.cores                        8
#spark.executor.memory                       12G

# For the executors
spark.executor.cores                        6
spark.executor.memory                       8G

# For the driver
spark.driver.cores                          6
spark.driver.memory                         8G

The first thing to do is to increase the memoryOverhead.

In the driver or in the executors?

When you are overviewing your cluster from the UI, you can click on the Attempt ID and check the Diagnostics Info which should mention the ID of the container that was killed. If it is the same as with your AM Container, then it's the driver, else the executor(s).

That didn't resolve the issue, now what?

You have to fine tune the number of cores and the heap memory you are providing. You see pyspark will do most of the work in off-heap memory, so you want not to give too much space for the heap, since that would be wasted. You don't want to give too less, because the Garbage Collector will have issues then. Recall that these are JVMs.

As described here, a worker can host multiple executors, thus the number of cores used affects how much memory every executor has, so decreasing the #cores might help.

I have it written in memoryOverhead issue in Spark and Spark – Container exited with a non-zero exit code 143 in more detail, mostly that I won't forget! Another option, that I haven't tried would be spark.default.parallelism or/and spark.storage.memoryFraction, which based on my experience, didn't help.

You can pass configurations flags as sds mentioned, or like this:

spark-submit --properties-file my_properties

where "my_properties" is something like the attributes I list above.

For non numerical values, you could do this:

spark-submit --conf spark.executor.memory='4G'

spark scalability: what am I doing wrong?

2 Answers2

Linked