'spark.driver.maxResultSize'范围

Question

我正在运行一个Spark作业来聚合数据。我有一个名为Profile的自定义数据结构，它基本上包含一个mutable.HashMap[Zone, Double]。我想合并所有共享给定键(UUID)的配置文件，代码如下:

def merge = (up1: Profile, up2: Profile) => { up1.addWeights(up2); up1}
val aggregated = dailyProfiles
  .aggregateByKey(new Profile(), 3200)(merge, merge).cache()

奇怪的是，Spark出现以下错误:

org.apache.spark。SparkException: Job aborted due to stage failure: 116318个任务序列化结果的总大小(1024.0 MB)大于spark.driver.maxResultSize (1024.0 MB)

显而易见的解决方案是增加"spark.driver"。maxResultSize"，但有两件事让我困惑。

1. 太巧了，我得到的1024.0大于1024.0
2. 所有的文档和帮助，我发现谷歌这个特定的错误和配置参数表明，它会影响函数的值返回到驱动程序。(说take()或collect())，但我没有采取任何驱动程序，只是从HDFS读取，聚合，保存回HDFS。

有人知道为什么我得到这个错误吗?

Answer 1

是，它失败是因为我们在异常消息中看到的值是

序列化后的输出大于1024.0 MB小于1024.1 MB

检查添加的Apache Spark代码片段，这是非常有趣的，很少得到这个错误。:)

这里totalResultSize > maxResultSize都是Long类型，并且in以字节为单位保存值。但是msg保留了Utils.bytesToString()的四舍五入值。

//TaskSetManager.scala
  def canFetchMoreResults(size: Long): Boolean = sched.synchronized {
    totalResultSize += size
    calculatedTasks += 1
    if (maxResultSize > 0 && totalResultSize > maxResultSize) {
      val msg = s"Total size of serialized results of ${calculatedTasks} tasks " +
        s"(${Utils.bytesToString(totalResultSize)}) is bigger than spark.driver.maxResultSize " +
        s"(${Utils.bytesToString(maxResultSize)})"
      logError(msg)
      abort(msg)
      false
    } else {
      true
    }
  }

Apache Spark 1.3 - source

---

//Utils.scala
  def bytesToString(size: Long): String = {
    val TB = 1L << 40
    val GB = 1L << 30
    val MB = 1L << 20
    val KB = 1L << 10
    val (value, unit) = {
      if (size >= 2*TB) {
        (size.asInstanceOf[Double] / TB, "TB")
      } else if (size >= 2*GB) {
        (size.asInstanceOf[Double] / GB, "GB")
      } else if (size >= 2*MB) {
        (size.asInstanceOf[Double] / MB, "MB")
      } else if (size >= 2*KB) {
        (size.asInstanceOf[Double] / KB, "KB")
      } else {
        (size.asInstanceOf[Double], "B")
      }
    }
    "%.1f %s".formatLocal(Locale.US, value, unit)
  }

Apache Spark 1.3 - source