超过 `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) 中的 DataFrame 假定的分区数量翻了一番，达到了 .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)

Answer 2

理论上，异常并不总是与客户数据相关。

关于任务执行结果的技术信息以序列化的形式发送到驱动节点，这些信息可能会占用比阈值更多的内存。

证明： 错误消息位于 org.apache.spark.scheduler.TaskSetManager#canFetchMoreResults

val msg = s"Total size of serialized results of ${calculatedTasks} tasks " +

在 org.apache.spark.scheduler.TaskResultGetter#enqueueSuccessfulTask​​ 中调用的方法

        val (result, size) = serializer.get().deserialize[TaskResult[_]](serializedData) match {
        case directResult: DirectTaskResult[_] =>
          if (!taskSetManager.canFetchMoreResults(serializedData.limit())) {
            return
          }

如果任务数量很大，可能会发生上述异常。