16 个任务的序列化结果总大小 (1048.5 MB) 大于 spark.driver.maxResultSize (1024.0 MB)

Question

将 --conf spark.driver.maxResultSize=2050 添加到我的 spark-submit 命令时出现以下错误。

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$$anonfun$run.apply$mcV$sp(Executor.scala:755)
        at org.apache.spark.executor.Executor$$anon$$anonfun$run.apply(Executor.scala:755)
        at org.apache.spark.executor.Executor$$anon$$anonfun$run.apply(Executor.scala:755)
        at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1954)
        at org.apache.spark.executor.Executor$$anon.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.access1(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)

添加此配置的原因是错误：

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)

因此，我将 maxResultSize 增加到 2.5 Gb，但 Spark 作业仍然失败（如上所示的错误）。如何解决这个问题？

Answer 1

问题似乎出在您尝试返回给您的驱动程序的数据量太大。您很可能正在使用 collect 方法从 DataFrame/RDD 中检索所有值。 driver 是一个单独的进程，通过收集 DataFrame，您可以将分布在集群中的所有数据拉回一个节点。 这违背了分发它的目的！只有在将数据减少到可管理的数量后才有意义。

你有两个选择：

如果你真的需要处理所有这些数据，那么你应该把它放在执行程序上。使用 HDFS 和 Parquet 以分布式方式保存数据，并使用 Spark 方法处理集群上的数据，而不是尝试收集数据全部回到一个地方。
如果您真的需要将数据返回给驱动程序，您应该检查您是否真的需要所有数据。如果您只需要汇总统计信息，那么在调用 collect 之前在执行程序上计算出来。或者，如果您只需要前 100 个结果，则只收集前 100 个。

更新：

您可以运行解决此错误的另一个原因不太明显。 Spark 将尝试将数据发送回驱动程序，而不仅仅是在您显式调用 collect 时。如果您正在使用累加器，它还会发回每个任务的累加器结果、广播连接的数据以及关于每个任务的一些小状态数据。如果您有很多分区（根据我的经验，超过 20k），您有时会看到此错误。这是一个known issue with some improvements made, and more in the works.

如果这是您的问题，则可以通过以下选项：

增加 spark.driver.maxResultSize 或设置为 0 无限制
如果广播连接是罪魁祸首，您可以减少spark.sql.autoBroadcastJoinThreshold以限制广播连接数据的大小
减少分区数

Answer 2

原因：由 RDD 的 collect() 等操作向驱动程序发送大块数据引起[=13=]

解决方法：由 SparkConf 设置：conf.set("spark.driver.maxResultSize", "4g") 要么由 spark-defaults.conf 设置：spark.driver.maxResultSize 4g 要么调用时设置 spark-submit: --conf spark.driver.maxResultSize=4g

16 个任务的序列化结果总大小 (1048.5 MB) 大于 spark.driver.maxResultSize (1024.0 MB)

Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)

python

apache-spark

pyspark

spark-dataframe