Spark using python: How to resolve Stage x contains a task of very large size (xxx KB). The maximum recommended task size is 100 KB

Question

I've just created python list of range(1,100000).

Using SparkContext done the following steps:

a = sc.parallelize([i for i in range(1, 100000)])
b = sc.parallelize([i for i in range(1, 100000)])

c = a.zip(b)

>>> [(1, 1), (2, 2), -----]

sum  = sc.accumulator(0)

c.foreach(lambda (x, y): life.add((y-x)))

Which gives warning as follows:

ARN TaskSetManager: Stage 3 contains a task of very large size (4644 KB). The maximum recommended task size is 100 KB.

How to resolve this warning? Is there any way to handle size? And also, will it affect the time complexity on big data?

Answer 1

The general idea is that PySpark creates as many java processes than there are executors, and then ships data to each process. If there are too few processes, a memory bottleneck happens on the java heap space.

In your case, the specific error is that the RDD that you created with sc.parallelize([...]) did not specify the number of partition (argument numSlices, see the docs). And the RDD defaults to a number of partition that is too small (possibly it is constituted by a single partition).

To solve this problem, simply specify the number of partitions wanted:

a = sc.parallelize([...], numSlices=1000)   # and likewise for b

As you specify higher and higher number of slices, you will see a decrease in the size stated in the warning message. Increase the number of slices until you get no more warning message. For example, getting

Stage 0 contains a task of very large size (696 KB). The maximum recommended task size is 100 KB

means that you need to specify more slices.

---

Another tip that may be useful when dealing with memory issues (but this is unrelated to the warning message): by default, the memory available to each executor is 1 GB or so. You can specify larger amounts through the commandline, for example with --executor-memory 64G.

Answer 2

Spark natively ships a copy of each variable over during the shipping of the task. For large sizes of such variables you may want to use Broadcast Variables

If you are still facing size problems, Then perhaps this data should be an RDD in itself

Answer 3

Expanding @leo9r comment: consider using not a python range, but sc.range https://spark.apache.org/docs/1.6.0/api/python/pyspark.html#pyspark.SparkContext.range.

Thus you avoid transfer of huge list from your driver to executors.

Of course, such RDDs are usually used for testing purposes only, so you do not want them to be broadcasted.