[SPARK-25410] Spark executor on YARN does not include memoryOverhead when starting an ExecutorRunnable

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Not A Bug
• Affects Version/s: 2.3.1
• Component/s: Spark Core, YARN
• Assignee: Unassigned
• Reporter: Anbang Hu
• Created: 12/Sep/18 00:48
• Updated: 17/May/20 18:14
• Resolved: 12/Sep/18 04:50

Description

When deploying on YARN, only executorMemory is used to launch executors in YarnAllocator.scala#L529:

              try {
                new ExecutorRunnable(
                  Some(container),
                  conf,
                  sparkConf,
                  driverUrl,
                  executorId,
                  executorHostname,
                  executorMemory,
                  executorCores,
                  appAttemptId.getApplicationId.toString,
                  securityMgr,
                  localResources
                ).run()
                updateInternalState()
              } catch {

However, resource capability requested for each executor is executorMemory + memoryOverhead in YarnAllocator.scala#L142:

  // Resource capability requested for each executors
  private[yarn] val resource = Resource.newInstance(executorMemory + memoryOverhead, executorCores)

This means that the amount of memoryOverhead will not be used in running the job, hence wasted.

Checking both k8s and Mesos, it looks like they both include overhead memory.
For k8s, in ExecutorPodFactory.scala#L179:

    val executorContainer = new ContainerBuilder()
      .withName("executor")
      .withImage(executorContainerImage)
      .withImagePullPolicy(imagePullPolicy)
      .withNewResources()
        .addToRequests("memory", executorMemoryQuantity)
        .addToLimits("memory", executorMemoryLimitQuantity)
        .addToRequests("cpu", executorCpuQuantity)
        .endResources()
      .addAllToEnv(executorEnv.asJava)
      .withPorts(requiredPorts.asJava)
      .addToArgs("executor")
      .build()

For Mesos, inMesosSchedulerUtils.scala#L374:

  /**
   * Return the amount of memory to allocate to each executor, taking into account
   * container overheads.
   *
   * @param sc SparkContext to use to get `spark.mesos.executor.memoryOverhead` value
   * @return memory requirement as (0.1 * memoryOverhead) or MEMORY_OVERHEAD_MINIMUM
   *         (whichever is larger)
   */
  def executorMemory(sc: SparkContext): Int = {
    sc.conf.getInt("spark.mesos.executor.memoryOverhead",
      math.max(MEMORY_OVERHEAD_FRACTION * sc.executorMemory, MEMORY_OVERHEAD_MINIMUM).toInt) +
      sc.executorMemory
  }