如何判断执行程序中的可用内存量

Question

Spark执行程序使用一些内存用于缓存(spark.storage.memoryFraction)和一些用于shuffle(spark.shuffle.memoryFraction)的内存.其余部分可供应用程序代码使用,例如在RDD.map操作中运行.

我想知道这个可用内存的数量.(我希望有大的分区仍然适合内存.我想将数据大小除以每个分区的可用内存以获得分区数.)

这是我如何计算:

val numExecutors = sc.getExecutorStorageStatus.size - 1 // Exclude driver.
val totalCores = numExecutors * numCoresPerExecutor
val cacheMemory = sc.getExecutorMemoryStatus.values.map(_._1).sum
val conf = sc.getConf
val cacheFraction = conf.getDouble("spark.storage.memoryFraction", 0.6)
val shuffleFraction = conf.getDouble("spark.shuffle.memoryFraction", 0.2)
val workFraction = 1.0 - cacheFraction - shuffleFraction
val workMemory = workFraction * cacheMemory / cacheFraction
val workMemoryPerCore = workMemory / totalCores

我相信你会同意这很可怕.最糟糕的是,如果Spark中的默认值发生变化,我的结果将是不正确的.但默认值在Spark中是硬编码的.我没办法搞定他们.

有更好的方法workMemoryPerCore吗？

Answer 1

据我了解，应用程序可用的最大内存是这样确定的（使用默认值）：

  private def getMaxMemory(conf: SparkConf): Long = {
    val memoryFraction = conf.getDouble("spark.storage.memoryFraction", 0.6)
    val safetyFraction = conf.getDouble("spark.storage.safetyFraction", 0.9)
    (Runtime.getRuntime.maxMemory * memoryFraction * safetyFraction).toLong
  }

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/storage/BlockManager.scala#L1227

这就是我之前计算应用程序内存的方式 - 尽管不是特别严格：

例如，如果spark.executor.memory = 4g

4 x 0.6 x 0.9 = 2.16g