Hadoop YARN同時支援記憶體和CPU兩種資源的排程，本文介紹如何配置YARN對記憶體和CPU的使用。

YARN作為一個資源排程器，應該考慮到叢集裡面每一臺機子的計算資源，然後根據application申請的資源進行分配Container。Container是YARN裡面資源分配的基本單位，具有一定的記憶體以及CPU資源。

在YARN叢集中，平衡記憶體、CPU、磁碟的資源的很重要的，根據經驗，每兩個container使用一塊磁碟以及一個CPU核的時候可以使叢集的資源得到一個比較好的利用。

記憶體配置

YARN以及MAPREDUCE所有可用的記憶體資源應該要除去系統執行需要的以及其他的hadoop的一些程式，總共保留的記憶體=系統記憶體+HBASE記憶體。

可以參考下面的表格確定應該保留的記憶體：

計算每臺機子最多可以擁有多少個container，可以使用下面的公式:

containers = min (2*CORES, 1.8*DISKS, (Total available RAM) / MIN_CONTAINER_SIZE)

說明：

• CORES為機器CPU核數
• DISKS為機器上掛載的磁碟個數
• Total available RAM為機器總記憶體
• MIN_CONTAINER_SIZE是指container最小的容量大小，這需要根據具體情況去設定，可以參考下面的表格：

每個container的平均使用記憶體大小計算方式為：

RAM-per-container = max(MIN_CONTAINER_SIZE, (Total Available RAM) / containers))

通過上面的計算，YARN以及MAPREDUCE可以這樣配置：

舉個例子：對於128G記憶體、32核CPU的機器，掛載了7個磁碟，根據上面的說明，系統保留記憶體為24G，不適應HBase情況下，系統剩餘可用記憶體為104G，計算containers值如下：

containers = min (2*32, 1.8* 7 , (128-24)/2) = min (64, 12.6 , 51) = 13

計算RAM-per-container值如下：

RAM-per-container = max (2, (124-24)/13) = max (2, 8) = 8

#!/usr/bin/env python
import optparse
from pprint import pprint
import logging
import sys
import math
import ast

''' Reserved for OS + DN + NM, Map: Memory => Reservation '''
reservedStack = { 4:1, 8:2, 16:2, 24:4, 48:6, 64:8, 72:8, 96:12, 
                   128:24, 256:32, 512:64}
''' Reserved for HBase. Map: Memory => Reservation '''
  
reservedHBase = {4:1, 8:1, 16:2, 24:4, 48:8, 64:8, 72:8, 96:16, 
                   128:24, 256:32, 512:64}
GB = 1024

def getMinContainerSize(memory):
  if (memory <= 4):
    return 256
  elif (memory <= 8):
    return 512
  elif (memory <= 24):
    return 1024
  else:
    return 2048
  pass

def getReservedStackMemory(memory):
  if (reservedStack.has_key(memory)):
    return reservedStack[memory]
  if (memory <= 4):
    ret = 1
  elif (memory >= 512):
    ret = 64
  else:
    ret = 1
  return ret

def getReservedHBaseMem(memory):
  if (reservedHBase.has_key(memory)):
    return reservedHBase[memory]
  if (memory <= 4):
    ret = 1
  elif (memory >= 512):
    ret = 64
  else:
    ret = 2
  return ret
                    
def main():
  log = logging.getLogger(__name__)
  out_hdlr = logging.StreamHandler(sys.stdout)
  out_hdlr.setFormatter(logging.Formatter(' %(message)s'))
  out_hdlr.setLevel(logging.INFO)
  log.addHandler(out_hdlr)
  log.setLevel(logging.INFO)
  parser = optparse.OptionParser()
  memory = 0
  cores = 0
  disks = 0
  hbaseEnabled = True
  parser.add_option('-c', '--cores', default = 16,
                     help = 'Number of cores on each host')
  parser.add_option('-m', '--memory', default = 64, 
                    help = 'Amount of Memory on each host in GB')
  parser.add_option('-d', '--disks', default = 4, 
                    help = 'Number of disks on each host')
  parser.add_option('-k', '--hbase', default = "True",
                    help = 'True if HBase is installed, False is not')
  (options, args) = parser.parse_args()
  
  cores = int (options.cores)
  memory = int (options.memory)
  disks = int (options.disks)
  hbaseEnabled = ast.literal_eval(options.hbase)
  
  log.info("Using cores=" + str(cores) + " memory=" + str(memory) + "GB" +
            " disks=" + str(disks) + " hbase=" + str(hbaseEnabled))
  minContainerSize = getMinContainerSize(memory)
  reservedStackMemory = getReservedStackMemory(memory)
  reservedHBaseMemory = 0
  if (hbaseEnabled):
    reservedHBaseMemory = getReservedHBaseMem(memory)
  reservedMem = reservedStackMemory + reservedHBaseMemory
  usableMem = memory - reservedMem
  memory -= (reservedMem)
  if (memory < 2):
    memory = 2
    reservedMem = max(0, memory - reservedMem)
    
  memory *= GB
  
  containers = int (min(2 * cores,
                         min(math.ceil(1.8 * float(disks)),
                              memory/minContainerSize)))
  if (containers <= 2):
    containers = 3

  log.info("Profile: cores=" + str(cores) + " memory=" + str(memory) + "MB"
           + " reserved=" + str(reservedMem) + "GB" + " usableMem="
           + str(usableMem) + "GB" + " disks=" + str(disks))
    
  container_ram = abs(memory/containers)
  if (container_ram > GB):
    container_ram = int(math.floor(container_ram / 512)) * 512
  log.info("Num Container=" + str(containers))
  log.info("Container Ram=" + str(container_ram) + "MB")
  log.info("Used Ram=" + str(int (containers*container_ram/float(GB))) + "GB")
  log.info("Unused Ram=" + str(reservedMem) + "GB")
  log.info("yarn.scheduler.minimum-allocation-mb=" + str(container_ram))
  log.info("yarn.scheduler.maximum-allocation-mb=" + str(containers*container_ram))
  log.info("yarn.nodemanager.resource.memory-mb=" + str(containers*container_ram))
  map_memory = container_ram
  reduce_memory = 2*container_ram if (container_ram <= 2048) else container_ram
  am_memory = max(map_memory, reduce_memory)
  log.info("mapreduce.map.memory.mb=" + str(map_memory))
  log.info("mapreduce.map.java.opts=-Xmx" + str(int(0.8 * map_memory)) +"m")
  log.info("mapreduce.reduce.memory.mb=" + str(reduce_memory))
  log.info("mapreduce.reduce.java.opts=-Xmx" + str(int(0.8 * reduce_memory)) + "m")
  log.info("yarn.app.mapreduce.am.resource.mb=" + str(am_memory))
  log.info("yarn.app.mapreduce.am.command-opts=-Xmx" + str(int(0.8*am_memory)) + "m")
  log.info("mapreduce.task.io.sort.mb=" + str(int(0.4 * map_memory)))
  pass

if __name__ == '__main__':
  try:
    main()
  except(KeyboardInterrupt, EOFError):
    print("\nAborting ... Keyboard Interrupt.")
    sys.exit(1)

執行下面命令：

python yarn-utils.py -c 32 -m 128 -d 7 -k False 

返回結果如下：

Using cores=32 memory=128GB disks=7 hbase=False
 Profile: cores=32 memory=106496MB reserved=24GB usableMem=104GB disks=7
 Num Container=13
 Container Ram=8192MB
 Used Ram=104GB
 Unused Ram=24GB
 yarn.scheduler.minimum-allocation-mb=8192
 yarn.scheduler.maximum-allocation-mb=106496
 yarn.nodemanager.resource.memory-mb=106496
 mapreduce.map.memory.mb=8192
 mapreduce.map.java.opts=-Xmx6553m
 mapreduce.reduce.memory.mb=8192
 mapreduce.reduce.java.opts=-Xmx6553m
 yarn.app.mapreduce.am.resource.mb=8192
 yarn.app.mapreduce.am.command-opts=-Xmx6553m
 mapreduce.task.io.sort.mb=3276

這樣的話，每個container記憶體為8G，似乎有點多，我更願意根據叢集使用情況任務將其調整為2G記憶體，則叢集中下面的引數配置值如下：

對應的xml配置為：

<property>
      <name>yarn.nodemanager.resource.memory-mb</name>
      <value>106496</value>
  </property>
  <property>
      <name>yarn.scheduler.minimum-allocation-mb</name>
      <value>2048</value>
  </property>
  <property>
      <name>yarn.scheduler.maximum-allocation-mb</name>
      <value>106496</value>
  </property>
  <property>
      <name>yarn.app.mapreduce.am.resource.mb</name>
      <value>4096</value>
  </property>
  <property>
      <name>yarn.app.mapreduce.am.command-opts</name>
      <value>-Xmx3276m</value>
  </property>

另外，還有一下幾個引數：

• yarn.nodemanager.vmem-pmem-ratio：任務每使用1MB實體記憶體，最多可使用虛擬記憶體量，預設是2.1。
• yarn.nodemanager.pmem-check-enabled：是否啟動一個執行緒檢查每個任務正使用的實體記憶體量，如果任務超出分配值，則直接將其殺掉，預設是true。
• yarn.nodemanager.vmem-pmem-ratio：是否啟動一個執行緒檢查每個任務正使用的虛擬記憶體量，如果任務超出分配值，則直接將其殺掉，預設是true。

第一個引數的意思是當一個map任務總共分配的實體記憶體為2G的時候，該任務的container最多內分配的堆記憶體為1.6G，可以分配的虛擬記憶體上限為2*2.1=4.2G。另外，照這樣算下去，每個節點上YARN可以啟動的Map數為104/2=52個。

CPU配置

YARN中目前的CPU被劃分成虛擬CPU（CPU virtual Core），這裡的虛擬CPU是YARN自己引入的概念，初衷是，考慮到不同節點的CPU效能可能不同，每個CPU具有的計算能力也是不一樣的，比如某個物理CPU的計算能力可能是另外一個物理CPU的2倍，這時候，你可以通過為第一個物理CPU多配置幾個虛擬CPU彌補這種差異。使用者提交作業時，可以指定每個任務需要的虛擬CPU個數。

在YARN中，CPU相關配置引數如下：

• yarn.nodemanager.resource.cpu-vcores：表示該節點上YARN可使用的虛擬CPU個數，預設是8，注意，目前推薦將該值設值為與物理CPU核數數目相同。如果你的節點CPU核數不夠8個，則需要調減小這個值，而YARN不會智慧的探測節點的物理CPU總數。
• yarn.scheduler.minimum-allocation-vcores：單個任務可申請的最小虛擬CPU個數，預設是1，如果一個任務申請的CPU個數少於該數，則該對應的值改為這個數。
• yarn.scheduler.maximum-allocation-vcores：單個任務可申請的最多虛擬CPU個數，預設是32。

對於一個CPU核數較多的叢集來說，上面的預設配置顯然是不合適的，在我的測試叢集中，4個節點每個機器CPU核數為31，留一個給作業系統，可以配置為：

<property>
      <name>yarn.nodemanager.resource.cpu-vcores</name>
      <value>31</value>
  </property>
  <property>
      <name>yarn.scheduler.maximum-allocation-vcores</name>
      <value>124</value>
  </property>