超过`spark.driver.maxResultSize`而不向驱动程序提供任何数据

Question

我有一个执行大型连接的Spark应用程序

val joined = uniqueDates.join(df, $"start_date" <= $"date" && $"date" <= $"end_date")

然后将生成的DataFrame汇总到一个可能有13k行的一个.在连接过程中,作业失败,并显示以下错误消息:

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 78021 tasks is bigger than spark.driver.maxResultSize (2.0 GB)

这是在没有设置之前发生的spark.driver.maxResultSize,所以我设置了spark.driver.maxResultSize=2G.然后,我对连接条件稍作修改,错误重新出现.

编辑:在调整集群,我也加倍数据帧呈现在分区数.coalesce(256)的.coalesce(512),所以我不能确定这是因为,不是.

我的问题是,既然我没有向司机收集任何东西,为什么要spark.driver.maxResultSize在这里重要？驱动程序的内存是否用于我不知道的连接中的某些内容？

Answer 1

仅仅因为你没有明确地收集任何东西,这并不意味着没有收集任何东西.由于在连接期间出现问题,最可能的解释是执行计划使用广播连接.在这种情况下,Spark将首先收集数据,然后进行广播.

根据配置和管道:

• 确保spark.sql.autoBroadcastJoinThreshold小于spark.driver.maxResultSize.
• 确保不强制对未知大小的数据进行广播连接.
• 虽然没有任何迹象表明这是问题,但在使用Spark ML实用程序时要小心.其中一些(最值得注意的是索引器)可以为驱动程序带来大量数据.

要确定广播是否确实存在问题,请检查执行计划,如果需要,删除广播提示并禁用自动广播:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)