70

I get the following error when I add --conf spark.driver.maxResultSize=2050 to my spark-submit command.

17/12/27 18:33:19 ERROR TransportResponseHandler: Still have 1 requests outstanding when connection from /XXX.XX.XXX.XX:36245 is closed
17/12/27 18:33:19 WARN Executor: Issue communicating with driver in heartbeater
org.apache.spark.SparkException: Exception thrown in awaitResult:
        at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:205)
        at org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:75)
        at org.apache.spark.rpc.RpcEndpointRef.askSync(RpcEndpointRef.scala:92)
        at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$reportHeartBeat(Executor.scala:726)
        at org.apache.spark.executor.Executor$$anon$2$$anonfun$run$1.apply$mcV$sp(Executor.scala:755)
        at org.apache.spark.executor.Executor$$anon$2$$anonfun$run$1.apply(Executor.scala:755)
        at org.apache.spark.executor.Executor$$anon$2$$anonfun$run$1.apply(Executor.scala:755)
        at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1954)
        at org.apache.spark.executor.Executor$$anon$2.run(Executor.scala:755)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: Connection from /XXX.XX.XXX.XX:36245 closed
        at org.apache.spark.network.client.TransportResponseHandler.channelInactive(TransportResponseHandler.java:146)

The reason of adding this configuration was the error:

py4j.protocol.Py4JJavaError: An error occurred while calling o171.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)

Therefore, I increased maxResultSize to 2.5 Gb, but the Spark job fails anyway (the error shown above). How to solve this issue?

4
  • --conf spark.driver.maxResultSize=2.5g can you try passing memory size like this? Commented Dec 27, 2017 at 17:59
  • also check stacktrace more thoroughly is there any outofmemory happening anywhere that might have led the worker to be killed? Commented Dec 27, 2017 at 18:03
  • @SumeetSharma: I tested it as well. There was the same error.
    – Markus
    Commented Dec 28, 2017 at 8:52
  • --conf spark.driver.maxResultSize=2050 is 2050 bytes (~2MB)
    – David Liao
    Commented Jun 7, 2022 at 19:56

2 Answers 2

86

It seems like the problem is the amount of data you are trying to pull back to to your driver is too large. Most likely you are using the collect method to retrieve all values from a DataFrame/RDD. The driver is a single process and by collecting a DataFrame you are pulling all of that data you had distributed across the cluster back to one node. This defeats the purpose of distributing it! It only makes sense to do this after you have reduced the data down to a manageable amount.

You have two options:

  1. If you really need to work with all that data, then you should keep it out on the executors. Use HDFS and Parquet to save the data in a distributed manner and use Spark methods to work with the data on the cluster instead of trying to collect it all back to one place.

  2. If you really need to get the data back to the driver, you should examine whether you really need ALL of the data or not. If you only need summary statistics then compute that out on the executors before calling collect. Or if you only need the top 100 results, then only collect the top 100.

Update:

There is another reason you can run into this error that is less obvious. Spark will try to send data back the driver beyond just when you explicitly call collect. It will also send back accumulator results for each task if you are using accumulators, data for broadcast joins, and some small status data about each task. If you have LOTS of partitions (20k+ in my experience) you can sometimes see this error. This is a known issue with some improvements made, and more in the works.

The options for getting past if if this is your issue are:

  1. Increase spark.driver.maxResultSize or set it to 0 for unlimited
  2. If broadcast joins are the culprit, you can reduce spark.sql.autoBroadcastJoinThreshold to limit the size of broadcast join data
  3. Reduce the number of partitions
7
  • 2
    Can you please elaborate with regards to point (1) ? Say I want to describe() a huge Dataframe which is being read from a parquet file. How can I achieve this, by keeping the parquet file out of the executors? Commented May 1, 2019 at 13:25
  • Use the DF transformations to create the statistics you need, THEN call collect/show to get the result back to the driver. That way you are only downloading the stats, not the full data. If you want to look at example rows, use show to get just the first few. Commented May 1, 2019 at 13:46
  • 1
    Yeah, I was keeping my answer more general, but describe().show() will work fine. Commented May 1, 2019 at 13:59
  • setting spark.driver.maxResultSize = 0 solved my problem in pyspark. I was using pyspark standalone on a single machine, and I believed it was okay to set unlimited size.
    – TG Gowda
    Commented Sep 21, 2020 at 6:36
  • 1
    Getting the same error but when I am writing the data out as parquet to s3. SO I guess its not just for operations like collect alone.
    – Rohit Anil
    Commented Nov 15, 2022 at 12:28
22

Cause: caused by actions like RDD's collect() that send big chunk of data to the driver

Solution: set by SparkConf: conf.set("spark.driver.maxResultSize", "4g") OR set by spark-defaults.conf: spark.driver.maxResultSize 4g OR set when calling spark-submit: --conf spark.driver.maxResultSize=4g

2
  • Is there a way we can do this using Luigi PySparkTask properties?
    – TheTank
    Commented Apr 15, 2020 at 16:31
  • 1
    setting spark.driver.maxResultSize = 0 solved my problem in pyspark. I was using pyspark standalone on a single machine, and I believed it was okay to set unlimited size.
    – TG Gowda
    Commented Sep 21, 2020 at 6:37

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.