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$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)

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

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)

因此，我增加到maxResultSize2.5 Gb，但是Spark作业仍然失败（上面显示的错误）。如何解决这个问题？

Answer 1

似乎问题在于您试图拉回驱动程序的数据量太大。您很可能正在使用 collect 方法从 DataFrame / RDD中 检索所有值。该
驱动程序 是一个过程，通过收集 数据框 你拉你已经在集群回分发到一个节点的数据的。 这违反了分发它的目的！
仅在将数据减少到可管理的数量之后才执行此操作。

您有两种选择：

1. 如果确实需要处理所有这些数据，则应将其保留在执行程序中。使用 HDFS 和 Parquet 以分布式方式保存数据，并使用Spark方法在群集上使用数据，而不是尝试将所有数据收集回一个位置。

2. 如果确实需要将数据送回驱动程序，则应检查是否确实需要所有数据。如果只需要摘要统计信息，则在调用collect之前在执行程序上进行计算。或者，如果您仅需要前100个结果，则仅 收集 前100个结果。

更新：

还有另一个原因会导致您不太明显地遇到此错误。除了您显式调用collect之外，Spark还将尝试将数据发送回驱动程序。如果您正在使用累加器，用于广播联接的数据以及有关每个任务的一些小状态数据，它还将为每个任务发送回累加器结果。如果您有很多分区（根据我的经验，超过20k），有时您会看到此错误。这是一个已知问题，需要进行一些改进，还有很多工作要做。

如果这是您遇到的问题，可以选择的解决方法是：

1. 增加spark.driver.maxResultSize或将其设置为0可无限
2. 如果是广播联接的元凶，则可以减小spark.sql.autoBroadcastJoinThreshold以限制广播联接数据的大小
3. 减少分区数