Spark中执行程序和任务的内存分配

Question

我的集群配置如下：-7个节点，每个节点有32个核心和252 GB内存。

纱线配置如下：-

yarn.scheduler.maximum-allocation-mb - 10GB
yarn.scheduler.minimum-allocation-mb - 2GB
yarn.nodemanager.vmem-pmem-ratio - 2.1
yarn.nodemanager.resource.memory-mb - 22GB
yarn.scheduler.maximum-allocation-vcores - 25
yarn.scheduler.minimum-allocation-vcores - 1
yarn.nodemanager.resource.cpu-vcores - 25

map reduce配置如下：-

mapreduce.map.java.opts - -Xmx1638m
mapreduce.map.memory.mb - 2GB
mapreduce.reduce.java.opts - -Xmx3276m
mapreduce.reduce.memory.mb - 4Gb

spark.yarn.driver.memoryOverhead 384
spark.yarn.executor.memoryOverhead 384

null

在这种情况下，对于纱线调度程序，执行器内存+384最大不能超过10GB。在本例中，9856M+384 MB=10GB，因此它工作正常。现在，一旦spark shell启动，执行程序的总数是124个，而不是请求的175个。每个执行器的spark shell启动日志或spark UI中的存储内存为6.7GB（即10GB的67%）。

spark shell进程的top命令输出如下：-

PID     USER      PR    NI  VIRT  RES   SHR S  %CPU %MEM  TIME+  
8478    hdp66-ss  20    0   13.5g 1.1g  25m S  1.9  0.4   2:11.28

因此虚拟内存为13.5G，物理内存为1.1G

PID     USER      PR    NI  VIRT  RES   SHR S  %CPU %MEM  TIME+  
5256    hdp66-ss  20    0   13.2g 1.1g  25m S  2.6  0.4   1:25.25

因此虚拟内存为13.2G，物理内存为1.1G

在这种情况下，对于纱线调度程序，执行器内存+384最大不能超过10GB。在本例中，4096M+384 MB=4GB，因此工作正常。现在，一旦spark shell启动，执行程序的总数为200个。在spark shell启动日志或spark UI中，每个执行器的存储内存为2.7GB（即4GB的67%）。

spark shell进程的top命令输出如下：-

PID     USER      PR    NI  VIRT  RES   SHR S  %CPU %MEM  TIME+  
21518   hdp66-ss  20    0   19.2g 1.4g  25m S  3.9  0.6   2:24.46

Answer 1

Spark几乎总是分配用户为执行程序请求的65%到70%的内存。Spark的这一行为缘于一张Spark JIRA票“spark-12579”。

这个链接指向Apache Spark存储库中的scala文件，该文件用于计算执行器内存等。

    if (conf.contains("spark.executor.memory")) {
  val executorMemory = conf.getSizeAsBytes("spark.executor.memory")
  if (executorMemory < minSystemMemory) {
    throw new IllegalArgumentException(s"Executor memory $executorMemory must be at least " +
      s"$minSystemMemory. Please increase executor memory using the " +
      s"--executor-memory option or spark.executor.memory in Spark configuration.")
  }
}
val usableMemory = systemMemory - reservedMemory
val memoryFraction = conf.getDouble("spark.memory.fraction", 0.6)
(usableMemory * memoryFraction).toLong

}