当我将--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)
因此,我将maxResultSize
增加到2.5 Gb,但无论如何Spark工作都失败了(上面显示的错误)。如何解决这个问题?
看起来问题是您尝试撤回到驱动程序的数据量太大。您很可能使用collect方法从DataFrame / RDD中检索所有值。驱动程序是一个进程,通过收集DataFrame,您将所有已在群集中分发的数据提取回一个节点。这违背了分发它的目的!只有在将数据减少到可管理的数量后才能执行此操作。
您有两种选择:
更新:
还有另一个原因可以解决这个不那么明显的错误。当您明确调用collect时,Spark将尝试将数据发送回驱动程序。如果您正在使用累加器,广播连接的数据以及有关每个任务的一些小状态数据,它还将发送每个任务的累加器结果。如果您有很多分区(根据我的经验,20k +),您有时会看到此错误。这是一个known issue有一些改进,和更多in the works。
如果这是您的问题,过去的选项是:
spark.driver.maxResultSize
或将其设置为0表示无限制spark.sql.autoBroadcastJoinThreshold
以限制广播连接数据的大小