scala - 超过 `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。然后，我对连接条件稍作更改，错误再次出现。

编辑: 在调整集群大小时，我还将 DataFrame 在 .coalesce(256) 中假设的分区数量翻了一番，使其变为 .coalesce(512 )，所以我不能确定不是因为这个。

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

Answer 1

仅仅因为您没有明确收集任何东西并不意味着没有收集任何东西。由于问题发生在连接期间，最可能的解释是执行计划使用广播连接。在这种情况下，Spark 会先收集数据，然后再广播它。

取决于配置和管道:

• 确保 spark.sql.autoBroadcastJoinThreshold 小于 spark.driver.maxResultSize。
• 确保您没有 force broadcast join在未知大小的数据上。
• 虽然没有任何迹象表明这是这里的问题，但在使用 Spark ML 实用程序时要小心。其中一些(最显着的索引器)可以为驱动程序带来大量数据。

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

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