devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Hadoop-Yarn-启动篇_error: refusing to run as root: roott account is n

作者：小小林熬夜学编程 | 2024-04-15 18:09:06

赞

踩

error: refusing to run as root: roott account is not found. aborting.

  一、源码下载

下面是hadoop官方源码下载地址，我下载的是hadoop-3.2.4，那就一起来看下吧

Index of /dist/hadoop/core

二、脚本部分

1、start-yarn.sh

如果我们想单独启动Yarn会用到$HADOOP_HOME/sbin/start-yarn.sh，下面我们就看看start-yarn.sh的内容 

  1. #!/usr/bin/env bash
  2.  
  3.  
  4. ## @description  usage info
  5. ## @audience     private
  6. ## @stability    evolving
  7. ## @replaceable  no
  8. function hadoop_usage
  9. {
  10.   hadoop_generate_usage "${MYNAME}" false
  11. }
  12. #获取当前文件的文件名
  13. MYNAME="${BASH_SOURCE-$0}"
  14.  
  15. bin=$(cd -P -- "$(dirname -- "${MYNAME}")" >/dev/null && pwd -P)
  16.  
  17. #找到 libexec
  18. if [[ -n "${HADOOP_HOME}" ]]; then
  19.   HADOOP_DEFAULT_LIBEXEC_DIR="${HADOOP_HOME}/libexec"
  20. else
  21.   HADOOP_DEFAULT_LIBEXEC_DIR="${bin}/../libexec"
  22. fi
  23.  
  24. HADOOP_LIBEXEC_DIR="${HADOOP_LIBEXEC_DIR:-$HADOOP_DEFAULT_LIBEXEC_DIR}"
  25. # shellcheck disable=SC2034
  26. # shellcheck扫描出的每一个错误都会有一个编号，以SC+4位数字组成
  27. # SC2034表示：变量赋值后未被使用
  28. HADOOP_NEW_CONFIG=true
  29. if [[ -f "${HADOOP_LIBEXEC_DIR}/yarn-config.sh" ]]; then
  30.   . "${HADOOP_LIBEXEC_DIR}/yarn-config.sh"
  31. else
  32.   echo "ERROR: Cannot execute ${HADOOP_LIBEXEC_DIR}/yarn-config.sh." 2>&1
  33.   exit 1
  34. fi
  35.  
  36. HADOOP_JUMBO_RETCOUNTER=0
  37.  
  38. # start resourceManager
  39. # 判断集群中的yarn是否开启了HA，这里我们先看不开启HA的部分
  40. HARM=$("${HADOOP_HDFS_HOME}/bin/hdfs" getconf -confKey yarn.resourcemanager.ha.enabled 2>&-)
  41. if [[ ${HARM} = "false" ]]; then
  42.   echo "Starting resourcemanager"
  43.   #到这里会不会想 hadoop_uservar_su 是个什么东西？
  44.   #hadoop_uservar_su其实是hadoop源码中common 模块中的hadoop-functions.sh中定义的一个函数
  45.   #以下还有很多类似的函数，基本都在这里面定义，下面让我们看看hadoop-functions.sh
  46.   hadoop_uservar_su yarn resourcemanager "${HADOOP_YARN_HOME}/bin/yarn" \
  47.       --config "${HADOOP_CONF_DIR}" \
  48.       --daemon start \
  49.       resourcemanager
  50.   (( HADOOP_JUMBO_RETCOUNTER=HADOOP_JUMBO_RETCOUNTER + $? ))
  51. else
  52.   logicals=$("${HADOOP_HDFS_HOME}/bin/hdfs" getconf -confKey yarn.resourcemanager.ha.rm-ids 2>&-)
  53.   logicals=${logicals//,/ }
  54.   for id in ${logicals}
  55.   do
  56.       rmhost=$("${HADOOP_HDFS_HOME}/bin/hdfs" getconf -confKey "yarn.resourcemanager.hostname.${id}" 2>&-)
  57.       RMHOSTS="${RMHOSTS} ${rmhost}"
  58.   done
  59.   echo "Starting resourcemanagers on [${RMHOSTS}]"
  60.   hadoop_uservar_su yarn resourcemanager "${HADOOP_YARN_HOME}/bin/yarn" \
  61.       --config "${HADOOP_CONF_DIR}" \
  62.       --daemon start \
  63.       --workers \
  64.       --hostnames "${RMHOSTS}" \
  65.       resourcemanager
  66.   (( HADOOP_JUMBO_RETCOUNTER=HADOOP_JUMBO_RETCOUNTER + $? ))
  67. fi
  68.  
  69. # start nodemanager
  70. echo "Starting nodemanagers"
  71. hadoop_uservar_su yarn nodemanager "${HADOOP_YARN_HOME}/bin/yarn" \
  72.     --config "${HADOOP_CONF_DIR}" \
  73.     --workers \
  74.     --daemon start \
  75.     nodemanager
  76. (( HADOOP_JUMBO_RETCOUNTER=HADOOP_JUMBO_RETCOUNTER + $? ))
  77.  
  78.  
  79. # start proxyserver
  80. PROXYSERVER=$("${HADOOP_HDFS_HOME}/bin/hdfs" getconf -confKey  yarn.web-proxy.address 2>&- | cut -f1 -d:)
  81. if [[ -n ${PROXYSERVER} ]]; then
  82.  hadoop_uservar_su yarn proxyserver "${HADOOP_YARN_HOME}/bin/yarn" \
  83.       --config "${HADOOP_CONF_DIR}" \
  84.       --workers \
  85.       --hostnames "${PROXYSERVER}" \
  86.       --daemon start \
  87.       proxyserver
  88.  (( HADOOP_JUMBO_RETCOUNTER=HADOOP_JUMBO_RETCOUNTER + $? ))
  89. fi
  90.  
  91. exit ${HADOOP_JUMBO_RETCOUNTER}

整体看start-yarn.sh可以看出，总共需要启动三个角色即：resourceManager、nodemanager、proxyserver。我们先看resourceManager

启动resourceManager会判断Yarn是否开启了HA，这里我们只分析没有开启HA的情况

2、hadoop-functions.sh

hadoop-functions.sh总共有2700多行，这里只列举Yarn启动期间涉及的函数，有兴趣的同学可以自行阅读。

  1. #!/usr/bin/env bash
  2.  
  3. ## @description  当以root身份运行时，通过su执行命令，并额外支持可能以root身份合法启动的命令
  4. ## @description  （例如，datanode）（这将由start-*/stop-*脚本使用。）
  5. ## @audience     private
  6. ## @stability    evolving
  7. ## @replaceable  no
  8. ## @param        user
  9. ## @param        commandstring
  10. ## @return       exitstatus
  11. function hadoop_uservar_su
  12. {
  13.  
  14.   declare program=$1
  15.   declare command=$2
  16.   shift 2
  17.  
  18.   declare uprogram
  19.   declare ucommand
  20.   declare uvar
  21.   declare svar
  22.  
  23.   if hadoop_privilege_check; then
  24.     uvar=$(hadoop_build_custom_subcmd_var "${program}" "${command}" USER)
  25.  
  26.     svar=$(hadoop_build_custom_subcmd_var "${program}" "${command}" SECURE_USER)
  27.  
  28.     if [[ -n "${!uvar}" ]]; then
  29.       hadoop_su "${!uvar}" "$@"
  30.     elif [[ -n "${!svar}" ]]; then
  31.       ## 如果我们在这里，那么没有定义USER的SECURE_USER我们已经有特权了，
  32.       ## 所以只需运行该命令，并希望一切顺利
  33.       "$@"
  34.     else
  35.       hadoop_error "ERROR: Attempting to operate on ${program} ${command} as root"
  36.       hadoop_error "ERROR: but there is no ${uvar} defined. Aborting operation."
  37.       return 1
  38.     fi
  39.   else
  40.     "$@"
  41.   fi
  42. }
  43.  
  44. ## @description  生成自定义子命令var
  45. ## @audience     public
  46. ## @stability    stable
  47. ## @replaceable  yes
  48. ## @param        command
  49. ## @param        subcommand
  50. ## @param        customid
  51. ## @return       string
  52. function hadoop_build_custom_subcmd_var
  53. {
  54.   declare program=$1
  55.   declare command=$2
  56.   declare custom=$3
  57.   declare uprogram
  58.   declare ucommand
  59.  
  60.   if [[ -z "${BASH_VERSINFO[0]}" ]] \
  61.      || [[ "${BASH_VERSINFO[0]}" -lt 4 ]]; then
  62.     uprogram=$(echo "${program}" | tr '[:lower:]' '[:upper:]')
  63.     ucommand=$(echo "${command}" | tr '[:lower:]' '[:upper:]')
  64.   else
  65.     uprogram=${program^^}
  66.     ucommand=${command^^}
  67.   fi
  68.  
  69.   echo "${uprogram}_${ucommand}_${custom}"
  70. }
  71.  
  72.  
  73. ## @description  如果找到给定的用户，则在以root身份运行时通过su执行命令；
  74. ## @description  如果没有找到，则退出并失败。
  75. ## @description  否则只需运行它。（这将由start-*/stop-*脚本使用。）
  76. ## @audience     private
  77. ## @stability    evolving
  78. ## @replaceable  yes
  79. ## @param        user
  80. ## @param        commandstring
  81. ## @return       exitstatus
  82. function hadoop_su
  83. {
  84.   declare user=$1
  85.   shift
  86.  
  87.   if hadoop_privilege_check; then
  88.     if hadoop_verify_user_resolves user; then
  89.        su -l "${user}" -- "$@"
  90.     else
  91.       hadoop_error "ERROR: Refusing to run as root: ${user} account is not found. Aborting."
  92.       return 1
  93.     fi
  94.   else
  95.     "$@"
  96.   fi
  97. }
  98.  
  99. ## @description  给定一个文件名或目录，返回它的绝对版本
  100. ## @description  这是readlink的替代品，它是不可移植的
  101. ## @audience     public
  102. ## @stability    stable
  103. ## @param        fsobj
  104. ## @replaceable  no
  105. ## @return       0 success
  106. ## @return       1 failure
  107. ## @return       stdout abspath
  108. function hadoop_abs
  109. {
  110.   declare obj=$1
  111.   declare dir
  112.   declare fn
  113.   declare dirret
  114.  
  115.   if [[ ! -e ${obj} ]]; then
  116.     return 1
  117.   elif [[ -d ${obj} ]]; then
  118.     dir=${obj}
  119.   else
  120.     dir=$(dirname -- "${obj}")
  121.     fn=$(basename -- "${obj}")
  122.     fn="/${fn}"
  123.   fi
  124.  
  125.   dir=$(cd -P -- "${dir}" >/dev/null 2>/dev/null && pwd -P)
  126.   dirret=$?
  127.   if [[ ${dirret} = 0 ]]; then
  128.     echo "${dir}${fn}"
  129.     return 0
  130.   fi
  131.   return 1
  132. }
  133.  
  134.  
  135. ## @description  验证是否允许${USER}执行给定的子命令。
  136. ## @audience     public
  137. ## @stability    stable
  138. ## @replaceable  yes
  139. ## @param        subcommand
  140. ## @return       1 on no re-exec needed
  141. ## @return       0 on need to re-exec
  142. function hadoop_need_reexec
  143. {
  144.   declare program=$1
  145.   declare command=$2
  146.   declare uvar
  147.  
  148.  
  149.   #默认false
  150.   if [[ "${HADOOP_REEXECED_CMD}" = true ]]; then
  151.     return 1
  152.   fi
  153.  
  154.   # if we have privilege, and the _USER is defined, and _USER is
  155.   # set to someone who isn't us, then yes, we should re-exec.
  156.   # otherwise no, don't re-exec and let the system deal with it.
  157.  
  158.   if hadoop_privilege_check; then
  159.     uvar=$(hadoop_build_custom_subcmd_var "${program}" "${command}" USER)
  160.     if [[ -n ${!uvar} ]]; then
  161.       if [[ ${!uvar} !=  "${USER}" ]]; then
  162.         return 0
  163.       fi
  164.     fi
  165.   fi
  166.   return 1
  167. }
  168.  
  169.  
  170. ## @description  验证是否允许${USER}执行给定的子命令
  171. ## @audience     public
  172. ## @stability    stable
  173. ## @replaceable  yes
  174. ## @param        command
  175. ## @param        subcommand
  176. ## @return       return 0 on success
  177. ## @return       exit 1 on failure
  178. function hadoop_verify_user_perm
  179. {
  180.   declare program=$1
  181.   declare command=$2
  182.   declare uvar
  183.  
  184.   uvar=$(hadoop_build_custom_subcmd_var "${program}" "${command}" USER)
  185.  
  186.   if [[ -n ${!uvar} ]]; then
  187.     if [[ ${!uvar} !=  "${USER}" ]]; then
  188.       hadoop_error "ERROR: ${command} can only be executed by ${!uvar}."
  189.       exit 1
  190.     fi
  191.   fi
  192.   return 0
  193. }
  194.  
  195.  
  196. ## @description  如果HADOOP_SUBCMD_SUPPORTDAEMONIZATION为false，
  197. ## @description  则将HADOOP_CLIENT_OPTS变量添加到HADOOP_OPTS
  198. ## @audience     public
  199. ## @stability    stable
  200. ## @replaceable  yes
  201. function hadoop_add_client_opts
  202. {
  203.   if [[ "${HADOOP_SUBCMD_SUPPORTDAEMONIZATION}" = false
  204.      || -z "${HADOOP_SUBCMD_SUPPORTDAEMONIZATION}" ]]; then
  205.     hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
  206.     HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
  207.   fi
  208. }
  209.  
  210. ## @description  要处理的实用程序例程--工人模式
  211. ## @audience     private
  212. ## @stability    evolving
  213. ## @replaceable  yes
  214. ## @param        commandarray
  215. function hadoop_common_worker_mode_execute
  216. {
  217.   #
  218.   # 输入应该是用户以数组形式给出的命令行
  219.   #
  220.   local argv=("$@")
  221.  
  222.   # 如果--workers仍在命令行上，请将其删除以防止循环。
  223.   # 还可以删除--hostname和--hosts以及arg值
  224.   local argsSize=${#argv[@]};
  225.   for (( i = 0; i < argsSize; i++ ))
  226.   do
  227.     if [[ "${argv[$i]}" =~ ^--workers$ ]]; then
  228.       unset argv[$i]
  229.     elif [[ "${argv[$i]}" =~ ^--hostnames$ ]] ||
  230.       [[ "${argv[$i]}" =~ ^--hosts$ ]]; then
  231.       unset argv[$i];
  232.       let i++;
  233.       unset argv[$i];
  234.     fi
  235.   done
  236.   if [[ ${QATESTMODE} = true ]]; then
  237.     echo "${argv[@]}"
  238.     return
  239.   fi
  240.   hadoop_connect_to_hosts -- "${argv[@]}"
  241. }
  242.  
  243.  
  244. ## @description  连接到${HADOOP_WORKERS}或${HADOOP_WORKER_NAMES}并执行命令。
  245. ## @audience     private
  246. ## @stability    evolving
  247. ## @replaceable  yes
  248. ## @param        command
  249. ## @param        [...]
  250. function hadoop_connect_to_hosts
  251. {
  252.   # shellcheck disable=SC2124
  253.   local params="$@"
  254.   local worker_file
  255.   local tmpslvnames
  256.  
  257.   #
  258.   # ssh (or whatever) to a host
  259.   #
  260.   # 用户可以指定主机名或主机名所在的文件（不能同时指定两者）
  261.   if [[ -n "${HADOOP_WORKERS}" && -n "${HADOOP_WORKER_NAMES}" ]] ; then
  262.     hadoop_error "ERROR: Both HADOOP_WORKERS and HADOOP_WORKER_NAMES were defined. Aborting."
  263.     exit 1
  264.   elif [[ -z "${HADOOP_WORKER_NAMES}" ]]; then
  265.     if [[ -n "${HADOOP_WORKERS}" ]]; then
  266.       worker_file=${HADOOP_WORKERS}
  267.     elif [[ -f "${HADOOP_CONF_DIR}/workers" ]]; then
  268.       worker_file=${HADOOP_CONF_DIR}/workers
  269.     elif [[ -f "${HADOOP_CONF_DIR}/slaves" ]]; then
  270.       hadoop_error "WARNING: 'slaves' file has been deprecated. Please use 'workers' file instead."
  271.       worker_file=${HADOOP_CONF_DIR}/slaves
  272.     fi
  273.   fi
  274.  
  275.   # 如果pdsh可用，让我们使用它。否则默认为ssh循环。（啊）
  276.   if [[ -e '/usr/bin/pdsh' ]]; then
  277.     if [[ -z "${HADOOP_WORKER_NAMES}" ]] ; then
  278.       # 如果给了我们一个文件，就让pdsh来处理它
  279.       # shellcheck disable=SC2086
  280.       PDSH_SSH_ARGS_APPEND="${HADOOP_SSH_OPTS}" pdsh \
  281.       -f "${HADOOP_SSH_PARALLEL}" -w ^"${worker_file}" $"${@// /\\ }" 2>&1
  282.     else
  283.       # pdsh-arg主机列表中不允许有空格
  284.       # shellcheck disable=SC2086
  285.       tmpslvnames=$(echo ${HADOOP_WORKER_NAMES} | tr -s ' ' ,)
  286.       PDSH_SSH_ARGS_APPEND="${HADOOP_SSH_OPTS}" pdsh \
  287.         -f "${HADOOP_SSH_PARALLEL}" \
  288.         -w "${tmpslvnames}" $"${@// /\\ }" 2>&1
  289.     fi
  290.   else
  291.     if [[ -z "${HADOOP_WORKER_NAMES}" ]]; then
  292.       HADOOP_WORKER_NAMES=$(sed 's/#.*$//;/^$/d' "${worker_file}")
  293.     fi
  294.     hadoop_connect_to_hosts_without_pdsh "${params}"
  295.   fi
  296. }
  297.  
  298.  
  299. ## @description  连接到${HADOOP_WORKER_NAMES}并在不支持pdsh的环境下执行命令
  300. ## @audience     private
  301. ## @stability    evolving
  302. ## @replaceable  yes
  303. ## @param        command
  304. ## @param        [...]
  305. function hadoop_connect_to_hosts_without_pdsh
  306. {
  307.   # shellcheck disable=SC2124
  308.   local params="$@"
  309.   local workers=(${HADOOP_WORKER_NAMES})
  310.   for (( i = 0; i < ${#workers[@]}; i++ ))
  311.   do
  312.     if (( i != 0 && i % HADOOP_SSH_PARALLEL == 0 )); then
  313.       wait
  314.     fi
  315.     # shellcheck disable=SC2086
  316.     hadoop_actual_ssh "${workers[$i]}" ${params} &
  317.   done
  318.   wait
  319. }
  320.  
  321. ## @description  通过ssh，登录“hostname”并运行“command”`
  322. ## @audience     private
  323. ## @stability    evolving
  324. ## @replaceable  yes
  325. ## @param        hostname
  326. ## @param        command
  327. ## @param        [...]
  328. function hadoop_actual_ssh
  329. {
  330.   # 我们将此函数传递给xargs，应获取hostname，然后是命令行的其余部分
  331.   local worker=$1
  332.   shift
  334.   # shellcheck disable=SC2086
  335.   ssh ${HADOOP_SSH_OPTS} ${worker} $"${@// /\\ }" 2>&1 | sed "s/^/$worker: /"
  336. }
  339. ## @description  通用shell脚本opton解析器。将HADOOP_PARSE_COUNTER设置为调用方应移位的数字
  340. ## @audience     private
  341. ## @stability    evolving
  342. ## @replaceable  yes
  343. ## @param        [parameters, typically "$@"]
  344. function hadoop_parse_args
  345. {
  346.   HADOOP_DAEMON_MODE="default"
  347.   HADOOP_PARSE_COUNTER=0
  349.   # 并非所有命令都支持此处支持的所有选项，这些选项是:
  350.   hadoop_add_option "--config dir" "Hadoop config directory"
  351.   hadoop_add_option "--debug" "turn on shell script debug mode"
  352.   hadoop_add_option "--help" "usage information"
  354.   while true; do
  355.     hadoop_debug "hadoop_parse_args: processing $1"
  356.     case $1 in
  357.       --buildpaths)
  358.         HADOOP_ENABLE_BUILD_PATHS=true
  359.         shift
  360.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+1))
  361.       ;;
  362.       --config)
  363.         shift
  364.         confdir=$1
  365.         shift
  366.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+2))
  367.         if [[ -d "${confdir}" ]]; then
  368.           HADOOP_CONF_DIR="${confdir}"
  369.         elif [[ -z "${confdir}" ]]; then
  370.           hadoop_error "ERROR: No parameter provided for --config "
  371.           hadoop_exit_with_usage 1
  372.         else
  373.           hadoop_error "ERROR: Cannot find configuration directory \"${confdir}\""
  374.           hadoop_exit_with_usage 1
  375.         fi
  376.       ;;
  377.       --daemon)
  378.         shift
  379.         HADOOP_DAEMON_MODE=$1
  380.         shift
  381.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+2))
  382.         if [[ -z "${HADOOP_DAEMON_MODE}" || \
  383.           ! "${HADOOP_DAEMON_MODE}" =~ ^st(art|op|atus)$ ]]; then
  384.           hadoop_error "ERROR: --daemon must be followed by either \"start\", \"stop\", or \"status\"."
  385.           hadoop_exit_with_usage 1
  386.         fi
  387.       ;;
  388.       --debug)
  389.         shift
  390.         HADOOP_SHELL_SCRIPT_DEBUG=true
  391.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+1))
  392.       ;;
  393.       --help|-help|-h|help|--h|--\?|-\?|\?)
  394.         hadoop_exit_with_usage 0
  395.       ;;
  396.       --hostnames)
  397.         shift
  398.         HADOOP_WORKER_NAMES="$1"
  399.         shift
  400.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+2))
  401.       ;;
  402.       --hosts)
  403.         shift
  404.         hadoop_populate_workers_file "$1"
  405.         shift
  406.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+2))
  407.       ;;
  408.       --loglevel)
  409.         shift
  410.         # shellcheck disable=SC2034
  411.         HADOOP_LOGLEVEL="$1"
  412.         shift
  413.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+2))
  414.       ;;
  415.       --reexec)
  416.         shift
  417.         if [[ "${HADOOP_REEXECED_CMD}" = true ]]; then
  418.           hadoop_error "ERROR: re-exec fork bomb prevention: --reexec already called"
  419.           exit 1
  420.         fi
  421.         HADOOP_REEXECED_CMD=true
  422.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+1))
  423.       ;;
  424.       --workers)
  425.         shift
  426.         # shellcheck disable=SC2034
  427.         HADOOP_WORKER_MODE=true
  428.         ((HADOOP_PARSE_COUNTER=HADOOP_PARSE_COUNTER+1))
  429.       ;;
  430.       *)
  431.         break
  432.       ;;
  433.     esac
  434.   done
  436.   hadoop_debug "hadoop_parse: asking caller to skip ${HADOOP_PARSE_COUNTER}"
  437. }
  440. ## @description  将自定义（程序）_（命令）_OPTS添加到HADOOP_OPTS。
  441. ## @description  还处理3.x之前版本中不推荐使用的案例。
  442. ## @audience     public
  443. ## @stability    evolving
  444. ## @replaceable  yes
  445. ## @param        program
  446. ## @param        subcommand
  447. ## @return       will exit on failure conditions
  448. function hadoop_subcommand_opts
  449. {
  450.   declare program=$1
  451.   declare command=$2
  452.   declare uvar
  453.   declare depvar
  454.   declare uprogram
  455.   declare ucommand
  457.   if [[ -z "${program}" || -z "${command}" ]]; then
  458.     return 1
  459.   fi
  461.   # bash 4 and up have built-in ways to upper and lower
  462.   # case the contents of vars.  This is faster than
  463.   # calling tr.
  465.   ## We don't call hadoop_build_custom_subcmd_var here
  466.   ## since we need to construct this for the deprecation
  467.   ## cases. For Hadoop 4.x, this needs to get cleaned up.
  469.   if [[ -z "${BASH_VERSINFO[0]}" ]] \
  470.      || [[ "${BASH_VERSINFO[0]}" -lt 4 ]]; then
  471.     uprogram=$(echo "${program}" | tr '[:lower:]' '[:upper:]')
  472.     ucommand=$(echo "${command}" | tr '[:lower:]' '[:upper:]')
  473.   else
  474.     uprogram=${program^^}
  475.     ucommand=${command^^}
  476.   fi
  478.   uvar="${uprogram}_${ucommand}_OPTS"
  480.   # Let's handle all of the deprecation cases early
  481.   # HADOOP_NAMENODE_OPTS -> HDFS_NAMENODE_OPTS
  483.   depvar="HADOOP_${ucommand}_OPTS"
  485.   if [[ "${depvar}" != "${uvar}" ]]; then
  486.     if [[ -n "${!depvar}" ]]; then
  487.       hadoop_deprecate_envvar "${depvar}" "${uvar}"
  488.     fi
  489.   fi
  491.   if [[ -n ${!uvar} ]]; then
  492.     hadoop_debug "Appending ${uvar} onto HADOOP_OPTS"
  493.     HADOOP_OPTS="${HADOOP_OPTS} ${!uvar}"
  494.     return 0
  495.   fi
  496. }
  500. ## @description 处理主程序项中的子命令
  501. ## @audience private
  502. ## @stability evolving
  503. ## @replaceable yes
  504. function hadoop_generic_java_subcmd_handler
  505. {
  506.   declare priv_outfile
  507.   declare priv_errfile
  508.   declare priv_pidfile
  509.   declare daemon_outfile
  510.   declare daemon_pidfile
  511.   declare secureuser
  513.   # 确定守护程序是否将在安全模式下运行的默认/预期方式由hadoop_detect_priv_subcmd定义。
  514.   # 如果返回true，则设置安全用户var并告诉世界我们处于安全模式
  516.   if hadoop_detect_priv_subcmd "${HADOOP_SHELL_EXECNAME}" "${HADOOP_SUBCMD}"; then
  517.     HADOOP_SUBCMD_SECURESERVICE=true
  518.     secureuser=$(hadoop_build_custom_subcmd_var "${HADOOP_SHELL_EXECNAME}" "${HADOOP_SUBCMD}" SECURE_USER)
  520.     if ! hadoop_verify_user_resolves "${secureuser}"; then
  521.       hadoop_error "ERROR: User defined in ${secureuser} (${!secureuser}) does not exist. Aborting."
  522.       exit 1
  523.     fi
  525.     HADOOP_SECURE_USER="${!secureuser}"
  526.   fi
  528.   # 检查我们是否在安全模式下运行。
  529.   # 从上面的内容来看，第三方可以做一些不同的事情——安全服务需要一些额外的设置——
  530.   # 如果是，那么我们需要定义所有的priv和daemon内容
  531.   # 如果不是，那么我们只需要定义daemon的内容。
  532.   # 请注意，两者之间的守护进程变量有目的地不同
  534.   if [[ "${HADOOP_SUBCMD_SECURESERVICE}" = true ]]; then
  536.     hadoop_subcommand_secure_opts "${HADOOP_SHELL_EXECNAME}" "${HADOOP_SUBCMD}"
  538.     hadoop_verify_secure_prereq
  539.     hadoop_setup_secure_service
  540.     priv_outfile="${HADOOP_LOG_DIR}/privileged-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}-${HOSTNAME}.out"
  541.     priv_errfile="${HADOOP_LOG_DIR}/privileged-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}-${HOSTNAME}.err"
  542.     priv_pidfile="${HADOOP_PID_DIR}/privileged-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}.pid"
  543.     daemon_outfile="${HADOOP_LOG_DIR}/hadoop-${HADOOP_SECURE_USER}-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}-${HOSTNAME}.out"
  544.     daemon_pidfile="${HADOOP_PID_DIR}/hadoop-${HADOOP_SECURE_USER}-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}.pid"
  545.   else
  546.     daemon_outfile="${HADOOP_LOG_DIR}/hadoop-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}-${HOSTNAME}.out"
  547.     daemon_pidfile="${HADOOP_PID_DIR}/hadoop-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}.pid"
  548.   fi
  550.   # 我们真的处于守护进程模式吗？
  551.   # 如果是，请使用守护程序记录器和相应的日志文件。
  552.   if [[ "${HADOOP_DAEMON_MODE}" != "default" ]]; then
  553.     HADOOP_ROOT_LOGGER="${HADOOP_DAEMON_ROOT_LOGGER}"
  554.     if [[ "${HADOOP_SUBCMD_SECURESERVICE}" = true ]]; then
  555.       HADOOP_LOGFILE="hadoop-${HADOOP_SECURE_USER}-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}-${HOSTNAME}.log"
  556.     else
  557.       HADOOP_LOGFILE="hadoop-${HADOOP_IDENT_STRING}-${HADOOP_SUBCMD}-${HOSTNAME}.log"
  558.     fi
  559.   fi
  561.   # 完成环境定义：系统属性、env-var、类路径等。
  562.   hadoop_finalize
  564.   # 完成启动守护进程的艰巨工作，或者只是执行我们的交互式java类
  565.   if [[ "${HADOOP_SUBCMD_SUPPORTDAEMONIZATION}" = true ]]; then
  566.     if [[ "${HADOOP_SUBCMD_SECURESERVICE}" = true ]]; then
  567.       hadoop_secure_daemon_handler \
  568.         "${HADOOP_DAEMON_MODE}" \
  569.         "${HADOOP_SUBCMD}" \
  570.         "${HADOOP_SECURE_CLASSNAME}" \
  571.         "${daemon_pidfile}" \
  572.         "${daemon_outfile}" \
  573.         "${priv_pidfile}" \
  574.         "${priv_outfile}" \
  575.         "${priv_errfile}" \
  576.         "${HADOOP_SUBCMD_ARGS[@]}"
  577.     else
  578.       hadoop_daemon_handler \
  579.         "${HADOOP_DAEMON_MODE}" \
  580.         "${HADOOP_SUBCMD}" \
  581.         "${HADOOP_CLASSNAME}" \
  582.         "${daemon_pidfile}" \
  583.         "${daemon_outfile}" \
  584.         "${HADOOP_SUBCMD_ARGS[@]}"
  585.     fi
  586.     exit $?
  587.   else
  588.     hadoop_java_exec "${HADOOP_SUBCMD}" "${HADOOP_CLASSNAME}" "${HADOOP_SUBCMD_ARGS[@]}"
  589.   fi
  590. }

总结下来就是：切换到yarn用户执行

"${HADOOP_YARN_HOME}/bin/yarn" \

--config "${HADOOP_CONF_DIR}" \

--daemon start \

resourcemanager

下面我们看下yarn命令中的实现

3、yarn命令

  1. #!/usr/bin/env bash
  2.  
  3.  
  4. #正在执行的脚本的名称
  5. HADOOP_SHELL_EXECNAME="yarn"
  6. #MYNAME会得到当前脚本名称，即：yarn
  7. MYNAME="${BASH_SOURCE-$0}"
  8.  
  9. ## @description  构建yarn命令的用法文本
  10. ## @audience     public
  11. ## @stability    stable
  12. ## @replaceable  no
  13. function hadoop_usage
  14. {
  15.   hadoop_add_option "--buildpaths" "attempt to add class files from build tree"
  16.   hadoop_add_option "--daemon (start|status|stop)" "operate on a daemon"
  17.   hadoop_add_option "--hostnames list[,of,host,names]" "hosts to use in worker mode"
  18.   hadoop_add_option "--loglevel level" "set the log4j level for this command"
  19.   hadoop_add_option "--hosts filename" "list of hosts to use in worker mode"
  20.   hadoop_add_option "--workers" "turn on worker mode"
  21.  
  22.   hadoop_add_subcommand "app|application" client "prints application(s) report/kill application/manage long running application"
  23.   hadoop_add_subcommand "applicationattempt" client "prints applicationattempt(s) report"
  24.   hadoop_add_subcommand "classpath" client "prints the class path needed to get the hadoop jar and the required libraries"
  25.   hadoop_add_subcommand "cluster" client "prints cluster information"
  26.   hadoop_add_subcommand "container" client "prints container(s) report"
  27.   hadoop_add_subcommand "daemonlog" admin "get/set the log level for each daemon"
  28.   hadoop_add_subcommand "envvars" client "display computed Hadoop environment variables"
  29.   hadoop_add_subcommand "jar <jar>" client "run a jar file"
  30.   hadoop_add_subcommand "logs" client "dump container logs"
  31.   hadoop_add_subcommand "node" admin "prints node report(s)"
  32.   hadoop_add_subcommand "nodemanager" daemon "run a nodemanager on each worker"
  33.   hadoop_add_subcommand "proxyserver" daemon "run the web app proxy server"
  34.   hadoop_add_subcommand "queue" client "prints queue information"
  35.   hadoop_add_subcommand "registrydns" daemon "run the registry DNS server"
  36.   hadoop_add_subcommand "resourcemanager" daemon "run the ResourceManager"
  37.   hadoop_add_subcommand "rmadmin" admin "admin tools"
  38.   hadoop_add_subcommand "router" daemon "run the Router daemon"
  39.   hadoop_add_subcommand "schedulerconf" client "Updates scheduler configuration"
  40.   hadoop_add_subcommand "scmadmin" admin "SharedCacheManager admin tools"
  41.   hadoop_add_subcommand "sharedcachemanager" daemon "run the SharedCacheManager daemon"
  42.   hadoop_add_subcommand "timelinereader" client "run the timeline reader server"
  43.   hadoop_add_subcommand "timelineserver" daemon "run the timeline server"
  44.   hadoop_add_subcommand "top" client "view cluster information"
  45.   hadoop_add_subcommand "nodeattributes" client "node attributes cli client"
  46.   hadoop_add_subcommand "version" client "print the version"
  47.   hadoop_generate_usage "${HADOOP_SHELL_EXECNAME}" true
  48. }
  49.  
  50. ## @description  yarn命令的默认命令处理程序
  51. ## @audience     public
  52. ## @stability    stable
  53. ## @replaceable  no
  54. ## @param        CLI arguments
  55. function yarncmd_case
  56. {
  57.   subcmd=$1
  58.   shift
  59.  
  60.   case ${subcmd} in
  61.     app|application|applicationattempt|container)
  62.       HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.ApplicationCLI
  63.       set -- "${subcmd}" "$@"
  64.       HADOOP_SUBCMD_ARGS=("$@")
  65.       local sld="${HADOOP_YARN_HOME}/${YARN_DIR},\
  66. ${HADOOP_YARN_HOME}/${YARN_LIB_JARS_DIR},\
  67. ${HADOOP_HDFS_HOME}/${HDFS_DIR},\
  68. ${HADOOP_HDFS_HOME}/${HDFS_LIB_JARS_DIR},\
  69. ${HADOOP_COMMON_HOME}/${HADOOP_COMMON_DIR},\
  70. ${HADOOP_COMMON_HOME}/${HADOOP_COMMON_LIB_JARS_DIR}"
  71.       hadoop_translate_cygwin_path sld
  72.       hadoop_add_param HADOOP_OPTS service.libdir "-Dservice.libdir=${sld}"
  73.     ;;
  74.  
  75.     #......省略......
  76.  
  77.     nodemanager)
  78.       HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
  79.       hadoop_add_classpath "$HADOOP_YARN_HOME/$YARN_DIR/timelineservice/*"
  80.       hadoop_add_classpath "$HADOOP_YARN_HOME/$YARN_DIR/timelineservice/lib/*"
  81.       #NodeManager启动类
  82.       HADOOP_CLASSNAME='org.apache.hadoop.yarn.server.nodemanager.NodeManager'
  83.       # Backwards compatibility
  84.       if [[ -n "${YARN_NODEMANAGER_HEAPSIZE}" ]]; then
  85.         HADOOP_HEAPSIZE_MAX="${YARN_NODEMANAGER_HEAPSIZE}"
  86.       fi
  87.     ;;
  88.     proxyserver)
  89.       HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
  90.       #WebAppProxyServer启动类
  91.       HADOOP_CLASSNAME='org.apache.hadoop.yarn.server.webproxy.WebAppProxyServer'
  92.       # Backwards compatibility
  93.       if [[ -n "${YARN_PROXYSERVER_HEAPSIZE}" ]]; then
  94.         HADOOP_HEAPSIZE_MAX="${YARN_PROXYSERVER_HEAPSIZE}"
  95.       fi
  96.     ;;
  97.     resourcemanager)
  98.       HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
  99.       hadoop_add_classpath "$HADOOP_YARN_HOME/$YARN_DIR/timelineservice/*"
  100.       hadoop_add_classpath "$HADOOP_YARN_HOME/$YARN_DIR/timelineservice/lib/*"
  101.       #ResourceManager启动类
  102.       HADOOP_CLASSNAME='org.apache.hadoop.yarn.server.resourcemanager.ResourceManager'
  103.       # Backwards compatibility
  104.       if [[ -n "${YARN_RESOURCEMANAGER_HEAPSIZE}" ]]; then
  105.         HADOOP_HEAPSIZE_MAX="${YARN_RESOURCEMANAGER_HEAPSIZE}"
  106.       fi
  107.       local sld="${HADOOP_YARN_HOME}/${YARN_DIR},\
  108. ${HADOOP_YARN_HOME}/${YARN_LIB_JARS_DIR},\
  109. ${HADOOP_HDFS_HOME}/${HDFS_DIR},\
  110. ${HADOOP_HDFS_HOME}/${HDFS_LIB_JARS_DIR},\
  111. ${HADOOP_COMMON_HOME}/${HADOOP_COMMON_DIR},\
  112. ${HADOOP_COMMON_HOME}/${HADOOP_COMMON_LIB_JARS_DIR}"
  113.       #详细请看第2步中 hadoop_translate_cygwin_path  的处理逻辑
  114.       hadoop_translate_cygwin_path sld
  115.       hadoop_add_param HADOOP_OPTS service.libdir "-Dservice.libdir=${sld}"
  116.     ;;
  117.     
  118.     #......省略......
  119.  
  120.     *)
  121.       HADOOP_CLASSNAME="${subcmd}"
  122.       if ! hadoop_validate_classname "${HADOOP_CLASSNAME}"; then
  123.         hadoop_exit_with_usage 1
  124.       fi
  125.     ;;
  126.   esac
  127. }
  128.  
  129. # 找到libexec...
  130. if [[ -n "${HADOOP_HOME}" ]]; then
  131.   HADOOP_DEFAULT_LIBEXEC_DIR="${HADOOP_HOME}/libexec"
  132. else
  133.   bin=$(cd -P -- "$(dirname -- "${MYNAME}")" >/dev/null && pwd -P)
  134.   HADOOP_DEFAULT_LIBEXEC_DIR="${bin}/../libexec"
  135. fi
  136.  
  137. HADOOP_LIBEXEC_DIR="${HADOOP_LIBEXEC_DIR:-$HADOOP_DEFAULT_LIBEXEC_DIR}"
  138. HADOOP_NEW_CONFIG=true
  139. if [[ -f "${HADOOP_LIBEXEC_DIR}/yarn-config.sh" ]]; then
  140.   # shellcheck source=./hadoop-yarn-project/hadoop-yarn/bin/yarn-config.sh
  141.   . "${HADOOP_LIBEXEC_DIR}/yarn-config.sh"
  142. else
  143.   echo "ERROR: Cannot execute ${HADOOP_LIBEXEC_DIR}/yarn-config.sh." 2>&1
  144.   exit 1
  145. fi
  146.  
  147. # hadoop_abs 也是 第2步 hadoop-functions.sh 中的函数 ，感兴趣的可以看第2步
  148. # hadoop_abs 的作用是：给定一个文件名或目录，返回它的绝对版本
  149. MYNAME=$(hadoop_abs "${MYNAME}")
  150.  
  151. # 如果未指定参数，则显示用法
  152. # $#表示执行脚本传入参数的个数
  153. if [[ $# = 0 ]]; then
  154.   hadoop_exit_with_usage 1
  155. fi
  156.  
  157. # 获取参数
  158. # shift 命令左移 shift执行后可以使 后面的参数向前移动，比如 shift 可以使 $2 成为 $1 , shift 3 可以使 $4 成为 $1
  159. # 此时 HADOOP_SUBCMD=resourcemanager
  160. HADOOP_SUBCMD=$1
  161. shift
  162.  
  163. #检测用户是否可以执行给定的命令
  164. #如果此节点是resourcemanager角色，那么就启动并退出脚本
  165. #hadoop_uservar_su yarn resourcemanager yarn --reexec 
  166. if hadoop_need_reexec yarn "${HADOOP_SUBCMD}"; then
  167.   hadoop_uservar_su yarn "${HADOOP_SUBCMD}" \
  168.     "${MYNAME}" \
  169.     "--reexec" \
  170.     "${HADOOP_USER_PARAMS[@]}"
  171.   exit $?
  172. fi
  173.  
  174. #验证是否允许 yarn 执行给定的子命令。
  175. hadoop_verify_user_perm "${HADOOP_SHELL_EXECNAME}" "${HADOOP_SUBCMD}"
  176.  
  177. #得到子命令的所有参数，就像数组一样，可以得到某一个参数或全部参数
  178. HADOOP_SUBCMD_ARGS=("$@")
  179.  
  180. #判断 yarn_subcommand_"${HADOOP_SUBCMD}" 函数是否存在
  181. #如果是直接执行
  182. #如果不是运行上面的 yarncmd_case 匹配到子命令进行执行
  183. if declare -f yarn_subcommand_"${HADOOP_SUBCMD}" >/dev/null 2>&1; then
  184.   hadoop_debug "Calling dynamically: yarn_subcommand_${HADOOP_SUBCMD} ${HADOOP_SUBCMD_ARGS[*]}"
  185.   "yarn_subcommand_${HADOOP_SUBCMD}" "${HADOOP_SUBCMD_ARGS[@]}"
  186. else
  187.   yarncmd_case "${HADOOP_SUBCMD}" "${HADOOP_SUBCMD_ARGS[@]}"
  188. fi
  189.  
  190. #目前还不清楚YARN_CLIENT_OPTS是否真的是一个有用的东西，
  191. #可以与HADOOP_CLIENT _OPTS分离。有人可能会使用它，
  192. #所以在我们（潜在地）将其添加到命令行之前，
  193. #我们不要弃用它，只需重写HADOOP_CLIENT_OPTS
  194. if [[ -n "${YARN_CLIENT_OPTS}" ]]; then
  195.   HADOOP_CLIENT_OPTS=${YARN_CLIENT_OPTS}
  196. fi
  197.  
  198. #如果HADOOP_SUBCMD_SUPPORTDAEMONIZATION为false，则将HADOOP_CLIENT_OPTS变量添加到HADOOP_OPTS
  199. hadoop_add_client_opts
  200.  
  201. #命令中包含了 --workers 参数，就代表着需要链接到其他节点机器上执行命令
  202. #hadoop_common_worker_mode_execute 函数最终会通过 ssh 到远程节点执行命令 ，详见第2步
  203. if [[ ${HADOOP_WORKER_MODE} = true ]]; then
  204.   hadoop_common_worker_mode_execute "${HADOOP_YARN_HOME}/bin/yarn" "${HADOOP_USER_PARAMS[@]}"
  205.   exit $?
  206. fi
  207.  
  208. #将自定义（程序）_（命令）_OPTS添加到HADOOP_OPTS
  209. hadoop_subcommand_opts "${HADOOP_SHELL_EXECNAME}" "${HADOOP_SUBCMD}"
  210.  
  211. # 此时所有内容都在全局中，因此调用泛型处理程序
  212. # 这里会完成启动守护进程的艰巨工作，或者只是执行我们的交互式java类
  213. hadoop_generic_java_subcmd_handler

下面会执行对应的java类，启动守护进程，我们看resourcemanager、nodemanager、proxyserver对应的java类，即：

org.apache.hadoop.yarn.server.resourcemanager.ResourceManager

org.apache.hadoop.yarn.server.nodemanager.NodeManager

org.apache.hadoop.yarn.server.webproxy.WebAppProxyServer

三、java部分

1、ResourceManager

1.1、main

  1. public static void main(String argv[]) {
  2.     Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
  3.     StringUtils.startupShutdownMessage(ResourceManager.class, argv, LOG);
  4.     try {
  5.       Configuration conf = new YarnConfiguration();
  6.       GenericOptionsParser hParser = new GenericOptionsParser(conf, argv);
  7.       argv = hParser.getRemainingArgs();
  8.       // 如果格式化状态存储，则删除RMStateStore；否则正常启动
  9.       if (argv.length >= 1) {
  10.         if (argv[0].equals("-format-state-store")) {
  11.           deleteRMStateStore(conf);
  12.         } else if (argv[0].equals("-format-conf-store")) {
  13.           deleteRMConfStore(conf);
  14.         } else if (argv[0].equals("-remove-application-from-state-store")
  15.             && argv.length == 2) {
  16.           removeApplication(conf, argv[1]);
  17.         } else {
  18.           printUsage(System.err);
  19.         }
  20.       } else {
  21.         ResourceManager resourceManager = new ResourceManager();
  22.         //添加一个具有优先级的shutdownBook，优先级越高，运行得越早。
  23.         //具有相同优先级的ShutdownHook以不确定的顺序运行。
  24.         //
  25.         //ShutdownHookManager允许以确定的顺序运行shutdownHook，优先考虑更高的优先级。
  26.         //JVM以不确定的顺序或并行方式运行ShutdownHook。这个类注册一个JVM shutdownBook，
  27.         //并根据优先级按顺序运行注册到它（这个类）的所有shutdownook。
  28.         //除非一个钩子注册了通过addShutdownHook(Runnable，int，long，TimeUnit)
  29.         //显式设置的关闭，否则分配给它的关闭时间由配置选项设置
  30.         //
  31.         //CompositeServiceShutdownHook 是 CompositeService的JVM Shutdown挂钩，
  32.         //它将在JVM关闭的情况下优雅地停止给定的CompositeService。
  33.         ShutdownHookManager.get().addShutdownHook(
  34.           new CompositeServiceShutdownHook(resourceManager),
  35.           SHUTDOWN_HOOK_PRIORITY);
  36.         //初始化，详细看该类中的serviceInit()，也就是第1.2步
  37.         resourceManager.init(conf);
  38.         //启动resourceManager ，详细看该类的serviceStart()，也就是第1.3步
  39.         resourceManager.start();
  40.       }
  41.     } catch (Throwable t) {
  42.       LOG.error(FATAL, "Error starting ResourceManager", t);
  43.       System.exit(-1);
  44.     }
  45.   }

1.2、serviceInit

  1. protected void serviceInit(Configuration conf) throws Exception {
  2.     this.conf = conf;
  3.     UserGroupInformation.setConfiguration(conf);
  4.     //RMContextImpl类包含两个服务上下文。
  5.     //  1、serviceContext：这些服务被称为Always-On服务。无论RM的HA状态如何，都需要始终运行的服务
  6.     //  2、activeServiceCotext:活动服务上下文。只需要在活动RM上运行的服务。
  7.     //注意：如果有任何新服务要添加到上下文中，请按照上面的描述将其添加到正确的上下文中。
  8.     this.rmContext = new RMContextImpl();
  9.     rmContext.setResourceManager(this);
  10.  
  11.     //实现ConfigurationProvider的基类。真正的ConfigurationProvider实现需要从中派生，并实现加载方法来实际加载配置。
  12.     this.configurationProvider =
  13.         ConfigurationProviderFactory.getConfigurationProvider(conf);
  14.     this.configurationProvider.init(this.conf);
  15.     rmContext.setConfigurationProvider(configurationProvider);
  16.  
  17.     //加载 core-site.xml
  18.     loadConfigurationXml(YarnConfiguration.CORE_SITE_CONFIGURATION_FILE);
  19.  
  20.     //使用加载的core-site.xml进行refreshSuperUserGroupsConfiguration，
  21.     //或者使用RM特定的配置先覆盖常见配置（如果存在）
  22.  
  23.     //查找名称以YarnConfiguration开头的所有配置。RM_PROXY_USER_PREFIX，
  24.     //并通过将前缀替换为ProxyUsers为每个前缀添加一条记录。CONF_HADOOP_PROXYUSER
  25.     RMServerUtils.processRMProxyUsersConf(conf);
  26.     //使用属性的默认代理用户前缀刷新配置。
  27.     ProxyUsers.refreshSuperUserGroupsConfiguration(this.conf);
  28.  
  29.     // 加载 yarn-site.xml
  30.     loadConfigurationXml(YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
  31.  
  32.     //配置的健全性检查
  33.     validateConfigs(this.conf);
  34.     
  35.     //登录前应设置HA配置
  36.     //如果HA开启，验证Resource Manager HA的配置
  37.     this.rmContext.setHAEnabled(HAUtil.isHAEnabled(this.conf));
  38.     if (this.rmContext.isHAEnabled()) {
  39.       HAUtil.verifyAndSetConfiguration(this.conf);
  40.     }
  41.  
  42.     // 设置UGI并进行登录
  43.     // 如果启用了安全性，请使用登录用户
  44.     // 如果未启用安全性，则使用当前用户
  45.     this.rmLoginUGI = UserGroupInformation.getCurrentUser();
  46.     try {
  47.       //详细见下方
  48.       doSecureLogin();
  49.     } catch(IOException ie) {
  50.       throw new YarnRuntimeException("Failed to login", ie);
  51.     }
  52.  
  53.     //使用setupDispatcher()注册所有AlwaysOn服务的处理程序
  54.     //注册alwaysOn服务的处理程序
  55.     rmDispatcher = setupDispatcher();
  56.     addIfService(rmDispatcher);
  57.     rmContext.setDispatcher(rmDispatcher);
  58.  
  59.     // 以下服务的顺序不应更改，因为服务将以相同的顺序启动
  60.     // 由于选民服务需要初始化和启动管理服务，我们首先添加管理服务，然后添加选民服务
  61.  
  62.     adminService = createAdminService();
  63.     addService(adminService);
  64.     rmContext.setRMAdminService(adminService);
  65.  
  66.     //必须在管理员服务后添加选举人
  67.     if (this.rmContext.isHAEnabled()) {
  68.       //如果RM配置为使用嵌入的领导人选举人，则初始化领导人选举人。
  69.  
  70.       //判断两个配置是否开启
  71.       //    1、yarn.resourcemanager.ha.automatic-failover.enabled 默认true （启用自动故障切换。默认情况下，仅当启用HA时才启用）
  72.       //    2、yarn.resourcemanager.ha.automatic-failover.embedded 默认true （启用嵌入式自动故障切换。默认情况下，只有在启用HA时才会启用它。嵌入式elector依赖于RM状态存储来处理围栏，主要用于与ZKRMStateStore结合使用。）
  73.       if (HAUtil.isAutomaticFailoverEnabled(conf)
  74.           && HAUtil.isAutomaticFailoverEmbedded(conf)) {
  75.         EmbeddedElector elector = createEmbeddedElector();
  76.         addIfService(elector);
  77.         rmContext.setLeaderElectorService(elector);
  78.       }
  79.     }
  80.  
  81.     rmContext.setYarnConfiguration(conf);
  82.  
  83.     //创建 RMActiveServices 的实例并对其进行初始化。
  84.     //RMActiveServices处理RM中的所有活动服务。
  85.  
  86.     //创建StandByTransitionRunnable （这是一个线程）
  87.     //它是将RM转换为备用状态的类。同一个StandByTransitionRunnable对象可以在多个线程中使用，
  88.     //但只能运行一次。这是因为RM在转换到备用状态后可以返回到活动状态，
  89.     //而在旧上下文中运行的相同RM无法再次转换到待机状态。
  90.     //每次RM转换到活动状态时，都会创建一个新的可运行程序。
  91.  
  92.     //创建RMSecretManagerService，用于密钥管理
  93.  
  94.     //创建ContainerAllocationExpirer 用于容器的申请和到期管理
  95.  
  96.     //创建AMLivelinessMonitor AM的活动监控器
  97.  
  98.     //创建RMAppLifetimeMonitor 此服务将根据给定的生存期值监视应用程序。如果应用程序运行超过给定时间，则应用程序将被终止。
  99.  
  100.     //创建RMNodeLabelsManager 节点标签管理
  101.  
  102.     //创建NodeAttributesManager 此类捕获属性与RM的所有交互。
  103.     
  104.     //创建AllocationTagsManager 应用程序/容器标签和节点/机架之间的内存映射。受约束的亲和性/反亲和性和基数放置所需。
  105.  
  106.     //创建PlacementConstraintManagerService 用于存储和检索放置约束的接口
  107.  
  108.     //在此处添加资源配置文件，因为它已被AbstractYarnScheduler使用
  109.  
  110.     //创建ResourceProfilesManager 资源配置文件管理器的接口。提供一个接口以获取可用配置文件和一些辅助函数的列表。
  111.  
  112.     //创建MultiNodeSortingManager 节点排序管理器，用于运行所有排序线程和策略。扩展SchedulerNode
  113.  
  114.     //创建RMDelegatedNodeLabelsUpdater 定期更新ResourceManager的节点标签映射。它从RMNodeLabelsMappingProvider收集节点标签，并通过RMNodeLabels Manager更新节点标签映射。当配置“yars.node labels.configuration type”设置为“委派集中式”时，将启用此服务。
  115.  
  116.     //如果设置了yarn.resourcemanager.recovery.enabled 默认false （使RM在启动后恢复状态。如果为true，则必须指定yarn.resourcemanager.store.class。）则 创建 RMStateStore
  117.     //RMStateStore 实现ResourceManager状态存储的基类。负责异步通知和与YARN对象的接口。真正的存储实现需要从中派生，并实现阻塞存储和加载方法来实际存储和加载状态。
  118.  
  119.     //确定UserGroupInformation是使用Kerberos来确定用户身份，还是依赖于简单身份验证
  120.     //如果是Kerberos，则要创建DelegationTokenRenewer（续订应用程序委派令牌的服务）
  121.     
  122.     //为NodesListManager注册事件处理程序
  123.     //NodesListManager：负责读取主机/排除文件以限制对RM的访问等
  124.  
  125.     //初始化调度程序
  126.     //创建ResourceScheduler YarnScheduler的子类（使用此接口与调度程序进行通信，以便分配资源、清理资源。）
  127.     //可以通过 yarn.resourcemanager.scheduler.class 设置 默认是 容量调度，即：org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler
  128.     
  129.     //注册RmAppEvents的事件处理程序
  130.     //创建ApplicationEventDispatcher 
  131.  
  132.     //注册RmAppAttemptEvents的事件处理程序
  133.     //创建ApplicationAttemptEventDispatcher
  134.  
  135.     //为RmNodes注册事件处理程序
  136.     //创建NodeEventDispatcher
  137.  
  138.     //创建NMLivelinessMonitor 即 NodeManager生命周期监听
  139.     //nm.liveness-monitor.expiry-interval-ms 默认 600000 ms 600s 
  140.     //没 nm.liveness-monitor.expiry-interval-ms / 3 时间 监听一次 即默认 200s
  141.     
  142.     //创建ResourceTrackerService 资源跟踪
  143.  
  144.     //初始化指标系统、JVM和日志记录相关指标
  145.  
  146.     //创建JvmPauseMonitor 
  147.     //该类设置一个简单的线程，该线程在睡眠短时间间隔的循环中运行。
  148.     //如果睡眠时间明显长于其目标时间，则表示JVM或主机已暂停处理，
  149.     //这可能会导致其他问题。如果检测到这样的暂停，线程将记录一条消息。
  150.  
  151.     //初始化预订系统
  152.     //通过yarn.resourcemanager.reservation-system.enable设置启用标识 默认false 
  153.     //如果启用了，创建ReservationSystem
  154.     //可以通过yarn.resourcemanager.reservation-system.class设置启动类
  155.     //默认是CapacityReservationSystem.class
  156.  
  157.     //创建ApplicationMasterService 
  158.     
  159.     //创建ApplicationACLsManager
  160.  
  161.     //创建QueueACLsManager
  162.  
  163.     //创建RMAppManager （此类管理资源管理器的应用程序列表。）
  164.  
  165.     //注册RMAppManagerEvents的事件处理程序
  166.  
  167.     //创建ClientRMService （资源管理器的客户端接口。该模块处理从客户端到资源管理器的所有rpc接口。）
  168.     
  169.     //创建ApplicationMasterLauncher
  170.  
  171.     //创建RMNMInfo （JMX bean列出所有节点管理器的状态）
  172.  
  173.     //是否在ResourceManager上启用服务rest api
  174.     //可以通过 yarn.webapp.api-service.enable 设置 默认false
  175.     //如果启用，则创建SystemServiceManager 默认实例化类为org.apache.hadoop.yarn.service.client.SystemServiceManagerImpl
  176.     //SystemServiceManager实现
  177.     //扫描配置系统服务路径
  178.     //服务路径结构如下：
  179.     //SYSTEM_SERVICE_DIR_PATH
  180.     //|---- sync
  181.     //|     |--- user1
  182.     //|     |    |---- service1.yarnfile
  183.     //|     |    |---- service2.yarnfile
  184.     //|     |--- user2
  185.     //|     |    |---- service1.yarnfile
  186.     //|     |    ....
  187.     //|     |
  188.     //|---- async
  189.     //|     |--- user3
  190.     //|     |    |---- service1.yarnfile
  191.     //|     |    |---- service2.yarnfile
  192.     //|     |--- user4
  193.     //|     |    |---- service1.yarnfile
  194.     //|     |    ....
  195.     //|     |
  196.     //sync: 这些服务在服务同步启动时启动。这是一个阻塞服务启动
  197.     //async: 这些服务在单独的线程中启动，在服务启动后没有任何延迟。非阻塞服务启动
  198.  
  199.     createAndInitActiveServices(false);
  200.  
  201.  
  202.     //获取用于绑定的URL(RM的实际绑定地址)
  203.     //其中可以指定绑定主机名以覆盖webAppURLWithoutScheme中的主机名。将使用webAppURLWithoutScheme中指定的端口。
  204.     //可以通过 yarn.resourcemanager.bind-host 设置
  205.     webAppAddress = WebAppUtils.getWebAppBindURL(this.conf,
  206.                       YarnConfiguration.RM_BIND_HOST,
  207.                       WebAppUtils.getRMWebAppURLWithoutScheme(this.conf));
  208.  
  209.     //创建RMApplicationHistoryWriter
  210.     //ResourceManager使用这个类来编写RMApp、RMAppAttempt和RMContainer的信息。
  211.     //    这些API是非阻塞的，只需安排一个写入历史事件。
  212.     //    一个自包含的调度器向量将在单独的线程中处理事件，并提取将要持久化的所需字段。
  213.     //    然后，提取的信息将通过ApplicationHistoryStore的实现持久化
  214.     RMApplicationHistoryWriter rmApplicationHistoryWriter =
  215.         createRMApplicationHistoryWriter();
  216.     addService(rmApplicationHistoryWriter);
  217.     rmContext.setRMApplicationHistoryWriter(rmApplicationHistoryWriter);
  218.  
  219.     //首先初始化RM时间线收集器，以便系统度量发布者可以绑定到它
  220.     if (YarnConfiguration.timelineServiceV2Enabled(this.conf)) {
  221.       RMTimelineCollectorManager timelineCollectorManager =
  222.           createRMTimelineCollectorManager();
  223.       addService(timelineCollectorManager);
  224.       rmContext.setRMTimelineCollectorManager(timelineCollectorManager);
  225.     }
  226.  
  227.     //创建CombinedSystemMetricsPublisher
  228.     //度量系统发布者
  229.     SystemMetricsPublisher systemMetricsPublisher =
  230.         createSystemMetricsPublisher();
  231.     addIfService(systemMetricsPublisher);
  232.     rmContext.setSystemMetricsPublisher(systemMetricsPublisher);
  233.  
  234.     //注册ResourceManagerMXBean
  235.     //MBeans.register("ResourceManager", "ResourceManager", this);
  236.     //使用标准命名约定来注册MBean
  237.     registerMXBean();
  238.  
  239.     super.serviceInit(this.conf);
  240.   }
  241.  
  242.  
  243.   protected void doSecureLogin() throws IOException {
  244.     //从配置中检索RM绑定地址
  245. 	InetSocketAddress socAddr = getBindAddress(conf);
  246.     //以配置中指定的主体身份登录。将用户的Kerberos主体名称中的$host替换为hostname。
  247.     //如果是非安全模式-返回。如果没有可用的钥匙扣，则除一个例外
  248.  
  249.  
  250.     SecurityUtil.login(this.conf, YarnConfiguration.RM_KEYTAB,
  251.         YarnConfiguration.RM_PRINCIPAL, socAddr.getHostName());
  252.  
  253.     // 如果启用了安全性 将 UGI of loginUser 设置给 rmLoginUGI 
  254.     if (UserGroupInformation.isSecurityEnabled()) {
  255.       this.rmLoginUGI = UserGroupInformation.getLoginUser();
  256.     }
  257.   }

1.3、serviceStart

  1. protected void serviceStart() throws Exception {
  2.       RMStateStore rmStore = rmContext.getStateStore();
  3.       //无论恢复启用，状态存储都需要启动，因为应用程序需要事件才能移动到其他状态。
  4.       //RMStateStore启动
  5.       //实现ResourceManager状态存储的基类。
  6.       //负责异步通知和与YARN对象的接口。真正的存储实现需要从中派生，并实现阻塞存储和加载方法来实际存储和加载状态。
  7.       rmStore.start();
  8.  
  9.       //第1.2步 serviceInit() 时对其进行过初始化 ，默认false 
  10.       if(recoveryEnabled) {
  11.         try {
  12.           LOG.info("Recovery started");
  13.           rmStore.checkVersion();
  14.           if (rmContext.isWorkPreservingRecoveryEnabled()) {
  15.             rmContext.setEpoch(rmStore.getAndIncrementEpoch());
  16.           }
  17.           RMState state = rmStore.loadState();
  18.           recover(state);
  19.           LOG.info("Recovery ended");
  20.         } catch (Exception e) {
  21.           // the Exception from loadState() needs to be handled for
  22.           // HA and we need to give up master status if we got fenced
  23.           LOG.error("Failed to load/recover state", e);
  24.           throw e;
  25.         }
  26.       } else {
  27.         //是否开启联合身份验证
  28.         //可以通过 yarn.federation.enabled 设置 默认false 
  29.         if (HAUtil.isFederationEnabled(conf)) {
  30.           long epoch = conf.getLong(YarnConfiguration.RM_EPOCH,
  31.               YarnConfiguration.DEFAULT_RM_EPOCH);
  32.           rmContext.setEpoch(epoch);
  33.           LOG.info("Epoch set for Federation: " + epoch);
  34.         }
  35.       }
  36.  
  37.       super.serviceStart();
  38.     }
  39.  
  40.  
  41.   protected void serviceStart() throws Exception {
  42.     //获取在serviceInit中添加的所有服务列表
  43.     List<Service> services = getServices();
  44.     if (LOG.isDebugEnabled()) {
  45.       LOG.debug(getName() + ": starting services, size=" + services.size());
  46.     }
  47.     for (Service service : services) {
  48.       //启动服务。如果失败，将停止服务并引发异常
  49.       service.start();
  50.     }
  51.     super.serviceStart();
  52.   }
  53.

2、NodeManager

注意：以下启动场景都是在单个节点中运行的

2.1、main

  1.   public static void main(String[] args) throws IOException {
  2.     Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
  3.     StringUtils.startupShutdownMessage(NodeManager.class, args, LOG);
  4.     @SuppressWarnings("resource")
  5.     NodeManager nodeManager = new NodeManager();
  6.     Configuration conf = new YarnConfiguration();
  7.     //创建一个GenericOptionsParser以仅解析通用Hadoop参数。
  8.     //getRemainingArgs()可以获得除泛型参数之外的字符串参数数组
  9.     new GenericOptionsParser(conf, args);
  10.     //初始化并启动NodeManager,我们在第2.2步详细看下
  11.     nodeManager.initAndStartNodeManager(conf, false);
  12.   }

2.2、initAndStartNodeManager

  1. private void initAndStartNodeManager(Configuration conf, boolean hasToReboot) {
  2.     try {
  3.       //如果我们是基于Unix的系统，但没有bash，则无法启动。
  4.       //Bash是在基于Unix的系统下启动容器所必需的。
  5.       //也就是说容器的启动是通过Bash来启动的
  6.       if (!Shell.WINDOWS) {
  7.         if (!Shell.checkIsBashSupported()) {
  8.           String message =
  9.               "Failing NodeManager start since we're on a "
  10.                   + "Unix-based system but bash doesn't seem to be available.";
  11.           LOG.error(message);
  12.           throw new YarnRuntimeException(message);
  13.         }
  14.       }
  15.  
  16.       //如果我们正在重新启动，请移除旧的挂钩
  17.       if (hasToReboot && null != nodeManagerShutdownHook) {
  18.         ShutdownHookManager.get().removeShutdownHook(nodeManagerShutdownHook);
  19.       }
  20.       
  21.       //CompositeService的JVM Shutdown挂钩，它将在JVM关闭的情况下优雅地停止给定的CompositeService。
  22.       nodeManagerShutdownHook = new CompositeServiceShutdownHook(this);
  23.       ShutdownHookManager.get().addShutdownHook(nodeManagerShutdownHook,
  24.                                                 SHUTDOWN_HOOK_PRIORITY);
  25.       //只有当从main()函数实例化NodeManager时，才应调用系统出口
  26.       this.shouldExitOnShutdownEvent = true;
  27.       //初始化NodeManager，详细看第2.3步（本类的serviceInit()）
  28.       this.init(conf);
  29.       //启动NodeManager，详细看第2.4步（本类的serviceStart()）
  30.       this.start();
  31.     } catch (Throwable t) {
  32.       LOG.error("Error starting NodeManager", t);
  33.       System.exit(-1);
  34.     }
  35.   }

2.3、serviceInit

  1. protected void serviceInit(Configuration conf) throws Exception {
  2.     UserGroupInformation.setConfiguration(conf);
  3.     //是否启用RM保留工作的恢复
  4.     //可以通过yarn.resourcemanager.work-preserving-recovery.enabled设置，默认true
  5.     rmWorkPreservingRestartEnabled = conf.getBoolean(YarnConfiguration
  6.             .RM_WORK_PRESERVING_RECOVERY_ENABLED,
  7.         YarnConfiguration.DEFAULT_RM_WORK_PRESERVING_RECOVERY_ENABLED);
  8.  
  9.     try {
  10.       //初始化并启动RecoveryStore
  11.       initAndStartRecoveryStore(conf);
  12.     } catch (IOException e) {
  13.       String recoveryDirName = conf.get(YarnConfiguration.NM_RECOVERY_DIR);
  14.       throw new
  15.           YarnRuntimeException("Unable to initialize recovery directory at "
  16.               + recoveryDirName, e);
  17.     }
  18.  
  19.     //创建NMContainerTokenSecretManager
  20.     //NodeManager容器密钥管理
  21.     //NM只保留两个主钥匙。RM知道的当前密钥和上一个滚动间隔中的密钥。
  22.     NMContainerTokenSecretManager containerTokenSecretManager =
  23.         new NMContainerTokenSecretManager(conf, nmStore);
  24.  
  25.     //创建NMTokenSecretManagerInNM
  26.     //里面有一个应用到尝试任务列表的映射和尝试应用和其密钥的映射
  27.     NMTokenSecretManagerInNM nmTokenSecretManager =
  28.         new NMTokenSecretManagerInNM(nmStore);
  29.  
  30.     recoverTokens(nmTokenSecretManager, containerTokenSecretManager);
  31.     
  32.     this.aclsManager = new ApplicationACLsManager(conf);
  33.  
  34.     //创建LocalDirsHandlerService
  35.     //提供检查节点本地目录运行状况功能的类。
  36.     //这通过定期检查nodemanager本地目录和nodemanager日志目录的运行状况来专门管理它们。
  37.     this.dirsHandler = new LocalDirsHandlerService(metrics);
  38.  
  39.     //是否启用分布式计划
  40.     //可以通过yarn.nodemanager.distributed-scheduling.enabled 设置 默认 false
  41.     boolean isDistSchedulingEnabled =
  42.         conf.getBoolean(YarnConfiguration.DIST_SCHEDULING_ENABLED,
  43.             YarnConfiguration.DEFAULT_DIST_SCHEDULING_ENABLED);
  44.  
  45.     //创建NM上下文
  46.     this.context = createNMContext(containerTokenSecretManager,
  47.         nmTokenSecretManager, nmStore, isDistSchedulingEnabled, conf);
  48.  
  49.     //创建ResourcePluginManager
  50.     //管理在此NodeManager上配置的ResourcePlugin
  51.     //ResourcePlugin:
  52.     //ResourcePlugin是节点管理器的一个接口，可以更容易地支持新资源类型的发现/管理/隔离
  53.     //主要有两部分：
  54.     //    createResourceHandler:
  55.     //            当资源类型需要任何特殊隔离时，插件需要返回ResourceHandler。
  56.     //            这将在NodeManager启动期间添加到ResourceHandlerChain中。
  57.     //            当不需要特殊隔离时，返回null。
  58.     //    getNodeResourceHandlerInstance:
  59.     //            当资源类型需要任何发现机制时，插件需要返回NodeResourceUpdaterPlugin
  60.     //            例如，如果我们想在NM注册期间设置资源值或在NM-RM心跳期间发送更新，
  61.     //            我们可以实现NodeResourceUpdaterPlugin
  62.     //            并更新NodeHeartbeatRequest或RegisterNodeManagerRequest的字段
  63.     //            这将在每次节点状态更新或节点注册时调用，请避免每次都创建新实例
  64.     ResourcePluginManager pluginManager = createResourcePluginManager();
  65.  
  66.     //
  67.     pluginManager.initialize(context);
  68.     ((NMContext)context).setResourcePluginManager(pluginManager);
  69.  
  70.     //创建ContainerExecutor
  71.     //这个类是用于在底层操作系统上启动容器的机制的抽象。所有执行器实现都必须扩展它
  72.     //可以通过yarn.nodemanager.container-executor.class设置该执行器
  73.     //默认是DefaultContainerExecutor.class
  74.     //DefaultContainerExecuter类提供通用的容器执行服务。通过ProcessBuilder以独立于平台的方式处理流程执行
  75.     //ProcessBuilder用于创建操作系统进程
  76.     //原来容器的创建最终是调用java的ProcessBuilder在操作系统中创建一个进程来实现的
  77.     ContainerExecutor exec = createContainerExecutor(conf);
  78.     try {
  79.       //运行执行器初始化步骤。验证必要的配置和权限是否到位。
  80.       exec.init(context);
  81.     } catch (IOException e) {
  82.       throw new YarnRuntimeException("Failed to initialize container executor", e);
  83.     }    
  84.     DeletionService del = createDeletionService(exec);
  85.     addService(del);
  86.  
  87.     //NodeManager级调度器
  88.     //AsyncDispatcher:
  89.     //在单独的线程中调度事件。目前只有一个线程能做到这一点。每个事件类型类可能有多个通道，并且可以使用线程池来调度事件。
  90.     this.dispatcher = createNMDispatcher();
  91.  
  92.     //创建NodeHealthCheckerService
  93.     //提供检查节点运行状况并向要求运行状况检查器报告的服务报告的功能的类
  94.     nodeHealthChecker =
  95.         new NodeHealthCheckerService(
  96.             getNodeHealthScriptRunner(conf), dirsHandler);
  97.     addService(nodeHealthChecker);
  98.  
  99.  
  100.     ((NMContext)context).setContainerExecutor(exec);
  101.     ((NMContext)context).setDeletionService(del);
  102.  
  103.     //创建NodeStatusUpdaterImpl
  104.     //节点状态更新，比如与ResourceManager的通信，对容器进行管理
  105.     nodeStatusUpdater =
  106.         createNodeStatusUpdater(context, dispatcher, nodeHealthChecker);
  107.  
  108.     //创建NodeLabelsProvider 负责获取节点标签
  109.     nodeLabelsProvider = createNodeLabelsProvider(conf);
  110.     if (nodeLabelsProvider != null) {
  111.       addIfService(nodeLabelsProvider);
  112.       nodeStatusUpdater.setNodeLabelsProvider(nodeLabelsProvider);
  113.     }
  114.  
  115.     //创建NodeAttributesProvider 负责获取节点属性
  116.     nodeAttributesProvider = createNodeAttributesProvider(conf);
  117.     if (nodeAttributesProvider != null) {
  118.       addIfService(nodeAttributesProvider);
  119.       nodeStatusUpdater.setNodeAttributesProvider(nodeAttributesProvider);
  120.     }
  121.     
  122.     //创建NodeResourceMonitorImpl 
  123.     //节点资源监视器的实现。它定期跟踪节点的资源利用情况，并将其报告给NM。
  124.     nodeResourceMonitor = createNodeResourceMonitor();
  125.     addService(nodeResourceMonitor);
  126.     ((NMContext) context).setNodeResourceMonitor(nodeResourceMonitor);
  127.  
  128.     //创建ContainerManagerImpl 管理容器生命周期的实体
  129.     containerManager =
  130.         createContainerManager(context, exec, del, nodeStatusUpdater,
  131.         this.aclsManager, dirsHandler);
  132.     addService(containerManager);
  133.     ((NMContext) context).setContainerManager(containerManager);
  134.  
  135.     //创建NMLogAggregationStatusTracker
  136.     //用于缓存已完成应用程序的日志聚合状态。它还将定期删除旧的缓存日志聚合状态
  137.     this.nmLogAggregationStatusTracker = createNMLogAggregationStatusTracker(
  138.         context);
  139.     addService(nmLogAggregationStatusTracker);
  140.     ((NMContext)context).setNMLogAggregationStatusTracker(
  141.         this.nmLogAggregationStatusTracker);
  142.  
  143.     WebServer webServer = createWebServer(context, containerManager
  144.         .getContainersMonitor(), this.aclsManager, dirsHandler);
  145.     addService(webServer);
  146.     ((NMContext) context).setWebServer(webServer);
  147.  
  148.     
  149.     ((NMContext) context).setQueueableContainerAllocator(
  150.         new OpportunisticContainerAllocator(
  151.             context.getContainerTokenSecretManager()));
  152.  
  153.     dispatcher.register(ContainerManagerEventType.class, containerManager);
  154.     dispatcher.register(NodeManagerEventType.class, this);
  155.     addService(dispatcher);
  156.  
  157.     //创建JvmPauseMonitor
  158.     //该类设置一个简单的线程，该线程在睡眠短时间间隔的循环中运行。
  159.     //如果睡眠时间明显长于其目标时间，则表示JVM或主机已暂停处理，
  160.     //这可能会导致其他问题。如果检测到这样的暂停，线程将记录一条消息。
  161.     pauseMonitor = new JvmPauseMonitor();
  162.     addService(pauseMonitor);
  163.     metrics.getJvmMetrics().setPauseMonitor(pauseMonitor);
  164.  
  165.     //初始化度量系统
  166.     DefaultMetricsSystem.initialize("NodeManager");
  167.  
  168.     //时间轴服务v.2是否通过配置启用
  169.     if (YarnConfiguration.timelineServiceV2Enabled(conf)) {
  170.       this.nmCollectorService = createNMCollectorService(context);
  171.       addService(nmCollectorService);
  172.     }
  173.  
  174.     //StatusUpdater应该最后添加，这样它才能最后启动，这样我们就可以在向RM注册之前确保一切正常
  175.     addService(nodeStatusUpdater);
  176.     ((NMContext) context).setNodeStatusUpdater(nodeStatusUpdater);
  177.     nmStore.setNodeStatusUpdater(nodeStatusUpdater);
  178.  
  179.     //在为添加的服务调用init之前进行安全登录。
  180.     try {
  181.       doSecureLogin();
  182.     } catch (IOException e) {
  183.       throw new YarnRuntimeException("Failed NodeManager login", e);
  184.     }
  185.  
  186.     //注册NodeManagerMXBean
  187.     //MBeans.register("NodeManager", "NodeManager", this);
  188.     registerMXBean();
  189.  
  190.     super.serviceInit(conf);
  191.     // TODO add local dirs to del
  192.   }
  193.  
  194.  
  195.   private void initAndStartRecoveryStore(Configuration conf)
  196.       throws IOException {
  197.     //是否启用节点管理器以在启动后恢复
  198.     //可以通过yarn.nodemanager.recovery.enabled设置，默认false
  199.     boolean recoveryEnabled = conf.getBoolean(
  200.         YarnConfiguration.NM_RECOVERY_ENABLED,
  201.         YarnConfiguration.DEFAULT_NM_RECOVERY_ENABLED);
  202.     if (recoveryEnabled) {
  203.       FileSystem recoveryFs = FileSystem.getLocal(conf);
  204.       String recoveryDirName = conf.get(YarnConfiguration.NM_RECOVERY_DIR);
  205.       if (recoveryDirName == null) {
  206.         throw new IllegalArgumentException("Recovery is enabled but " +
  207.             YarnConfiguration.NM_RECOVERY_DIR + " is not set.");
  208.       }
  209.       Path recoveryRoot = new Path(recoveryDirName);
  210.       recoveryFs.mkdirs(recoveryRoot, new FsPermission((short)0700));
  211.       nmStore = new NMLeveldbStateStoreService();
  212.     } else {
  213.       //未存储状态时要使用的状态存储
  214.       nmStore = new NMNullStateStoreService();
  215.     }
  216.     nmStore.init(conf);
  217.     nmStore.start();
  218.   }

2.4、serviceStart

  1.   protected void serviceStart() throws Exception {
  2.     //获取serviceInit()时添加到服务列表，并依次启动它们
  3.     List<Service> services = getServices();
  4.     if (LOG.isDebugEnabled()) {
  5.       LOG.debug(getName() + ": starting services, size=" + services.size());
  6.     }
  7.     for (Service service : services) {
  8.       service.start();
  9.     }
  10.     super.serviceStart();
  11.   }

3、WebAppProxyServer

ProxyServer将位于最终用户和AppMaster web界面之间。

3.1、main

  1.   public static void main(String[] args) {
  2.     Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
  3.     StringUtils.startupShutdownMessage(WebAppProxyServer.class, args, LOG);
  4.     try {
  5.       YarnConfiguration configuration = new YarnConfiguration();
  6.       new GenericOptionsParser(configuration, args);
  7.       //启动代理服务器
  8.       WebAppProxyServer proxyServer = startServer(configuration);
  9.       proxyServer.proxy.join();
  10.     } catch (Throwable t) {
  11.       ExitUtil.terminate(-1, t);
  12.     }
  13.   }
  14.  
  15.   protected static WebAppProxyServer startServer(Configuration configuration)
  16.       throws Exception {
  17.     WebAppProxyServer proxy = new WebAppProxyServer();
  18.     ShutdownHookManager.get().addShutdownHook(
  19.         new CompositeServiceShutdownHook(proxy), SHUTDOWN_HOOK_PRIORITY);
  20.     //实际调用本类的serviceInit()，详看第3.2步
  21.     proxy.init(configuration);
  22.     //初始化时只添加了两个服务 WebAppProxy 和 JvmPauseMonitor
  23.     //我们看下WebAppProxy 的启动，详见第3.3步
  24.     proxy.start();
  25.     return proxy;
  26.   }

3.2、serviceInit

  1. //只添加了两个服务 WebAppProxy 和 JvmPauseMonitor
  2. protected void serviceInit(Configuration conf) throws Exception {
  3.     Configuration config = new YarnConfiguration(conf);
  4.     //以指定给代理的Kerberose主体身份登录
  5.     doSecureLogin(conf);
  6.     //构建WebAppProxy
  7.     proxy = new WebAppProxy();
  8.     addService(proxy);
  9.  
  10.     //这些在RM、NM中都讲过
  11.     DefaultMetricsSystem.initialize("WebAppProxyServer");
  12.     JvmMetrics jm = JvmMetrics.initSingleton("WebAppProxyServer", null);
  13.     pauseMonitor = new JvmPauseMonitor();
  14.     addService(pauseMonitor);
  15.     jm.setPauseMonitor(pauseMonitor);
  16.  
  17.     super.serviceInit(config);
  18.   }

3.3、WebAppProxy

  1. protected void serviceInit(Configuration conf) throws Exception {
  2.     //core-default.xml中的 hadoop.security.authentication 值由kerberos和simple，默认simple
  3.     String auth =  conf.get(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION);
  4.     if (auth == null || "simple".equals(auth)) {
  5.       isSecurityEnabled = false;
  6.     } else if ("kerberos".equals(auth)) {
  7.       isSecurityEnabled = true;
  8.     } else {
  9.       LOG.warn("Unrecognized attribute value for " +
  10.           CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION +
  11.           " of " + auth);
  12.     }
  13.     //web代理的地址
  14.     //web代理的地址网络代理的地址为HOST:PORT，如果未给定，则代理将作为RM的一部分运行
  15.     //先看yarn.web-proxy.address是否设置，如果没有设置再看
  16.     //yarn.resourcemanager.webapp.address（RM web应用程序的http地址。如果只提供一个主机作为值，则网络应用程序将在随机端口上提供服务。）默认值为${yarn.resourcemanager.hostname}:8088
  17.     String proxy = WebAppUtils.getProxyHostAndPort(conf);
  18.     String[] proxyParts = proxy.split(":");
  19.     proxyHost = proxyParts[0];
  20.  
  21.     //创建到RM/Application History Server的新连接以获取应用程序报告。
  22.     fetcher = new AppReportFetcher(conf);
  23.     //获取yarn.web-proxy.address的值
  24.     bindAddress = conf.get(YarnConfiguration.PROXY_ADDRESS);
  25.     if(bindAddress == null || bindAddress.isEmpty()) {
  26.       throw new YarnRuntimeException(YarnConfiguration.PROXY_ADDRESS + 
  27.           " is not set so the proxy will not run.");
  28.     }
  29.     LOG.info("Instantiating Proxy at " + bindAddress);
  30.     String[] parts = StringUtils.split(bindAddress, ':');
  31.     port = 0;
  32.     if (parts.length == 2) {
  33.       bindAddress = parts[0];
  34.       port = Integer.parseInt(parts[1]);
  35.     }
  36.     //从相同ACL的字符串表示构建一个新ACL。
  37.     //字符串是一个以逗号分隔的用户和组列表。用户列表位于第一位，并由组列表后面的空格分隔。例如“user1，user2 group1，group2”
  38.     acl = new AccessControlList(conf.get(YarnConfiguration.YARN_ADMIN_ACL, 
  39.         YarnConfiguration.DEFAULT_YARN_ADMIN_ACL));
  40.     super.serviceInit(conf);
  41.   }
  42.   
  43.   @Override
  44.   protected void serviceStart() throws Exception {
  45.     try {
  46.       Configuration conf = getConfig();
  47.       HttpServer2.Builder b = new HttpServer2.Builder()
  48.           .setName("proxy")
  49.           .addEndpoint(
  50.               URI.create(WebAppUtils.getHttpSchemePrefix(conf) + bindAddress
  51.                   + ":" + port)).setFindPort(port == 0).setConf(getConfig())
  52.           .setACL(acl);
  53.       if (YarnConfiguration.useHttps(conf)) {
  54.         WebAppUtils.loadSslConfiguration(b);
  55.       }
  56.       proxyServer = b.build();
  57.       proxyServer.addServlet(ProxyUriUtils.PROXY_SERVLET_NAME,
  58.           ProxyUriUtils.PROXY_PATH_SPEC, WebAppProxyServlet.class);
  59.       proxyServer.setAttribute(FETCHER_ATTRIBUTE, fetcher);
  60.       proxyServer
  61.           .setAttribute(IS_SECURITY_ENABLED_ATTRIBUTE, isSecurityEnabled);
  62.       proxyServer.setAttribute(PROXY_HOST_ATTRIBUTE, proxyHost);
  63.       //调用HttpServer2.start()
  64.       //最终调用的时org.eclipse.jetty.server.start()
  65.       //启动服务器。不等待服务器启动
  66.       proxyServer.start();
  67.     } catch (IOException e) {
  68.       LOG.error("Could not start proxy web server",e);
  69.       throw e;
  70.     }
  71.     super.serviceStart();
  72.   }
  73.   
  74.   @Override
  75.   protected void serviceStop() throws Exception {
  76.     if(proxyServer != null) {
  77.       try {
  78.         proxyServer.stop();
  79.       } catch (Exception e) {
  80.         LOG.error("Error stopping proxy web server", e);
  81.         throw new YarnRuntimeException("Error stopping proxy web server",e);
  82.       }
  83.     }
  84.     if(this.fetcher != null) {
  85.       this.fetcher.stop();
  86.     }
  87.     super.serviceStop();
  88.   }

四、总结

1、用户执行./start-yarn.sh

2、start-yarn.sh中依次启动resourceManager、nodemanager、proxyserver

3、根据yarn命令和hadoop-functions.sh找到三个角色的启动类，并且在本地或者远程（通过ssh的方式）启动各自的java进程

4、resourceManager、nodemanager、proxyserver初始化各自的服务列表

5、resourceManager、nodemanager、proxyserver按照服务列表依次启动服务

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/429302
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号