【Yarn】Yarn的基本执行流程（二）AM Container的启动

本文主要是介绍【Yarn】Yarn的基本执行流程（二）AM Container的启动，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Yarn的基本执行流程之AM Container的启动

AM Container（第一个Container）的启动

NM RM心跳交互触发调度Container的启动流程

当有NM节点向RM发送心跳请求时，RM内部最终会以事件的形式通知到调度器，调度器则选择合适的应用为其分配资源。

1. NM上的节点NodeStatusUpdater服务的实现NodeStatusUpdaterImpl与ResourceTrackerService保持心跳联系，定时发送信息。
2. ResourceTrackerService验证相关信息后，会向调度器发送RMNodeStatusEvent事件。
3. RMNodeImpl向调度器发送NodeUpdateSchedulerEvent事件，由FairScheduler进行nodeUpdate处理，其中就包含了启动AM流程。

RM中调度启动AM流程

1. FairSchduler在更新节点时会调用attemptScheduling方法，尝试为Container分配资源
2. FSAppAttempt申请分配Container资源，实例化Container。
3. 申请到了Container资源，触发RMContainerEventType.START事件。在完成ContainerStartedTransition方法后，RMContainerState.NEW将变为RMContainerState.ALLOCATED。
4. ContainerStartedTransition方法发送了RMAppAttemptEventType.CONTAINER_ALLOCATED事件，触发AMContainerAllocatedTransition方法，RMAppAttemptState.SCHEDULED将变为RMAppAttemptState.ALLOCATED_SAVING。
5. 从调度器获取启动 AM 的 Container。
6. 把之前申请资源拿走。
7. 处理RMContainerEventType.ACQUIRED事件，RMContainerState.ALLOCATED将变为RMContainerState.ACQUIRED
8. 把当前Container所处的Node放入上下文，RMAppState.ACCEPTED保持不变
9. 当第5步开始的链路结束后，回到RMAppAttemptImpl，执行异步存储Attempt信息。
10. 处理RMStateStoreEventType.STORE_APP_ATTEMPT事件，RMStateStore.storeApplicationAttemptStateInternal持久化存储，是存ZK？？？
11. 存储成功时候，发送RMAppAttemptEventType.ATTEMPT_NEW_SAVED事件，RMAppAttemptState.ALLOCATED_SAVING将变为RMAppAttemptState.ALLOCATED。
12. 执行registerClientToken()注册客户端token，向处理器发送AMLauncherEventType.LAUNCH事件。处理器是将事件交由ApplicationMasterLauncher继续处理AMLauncher启动流程。

AMLauncher启动流程

ResourceManager启动的时候会将会创建ApplicationMasterLauncher服务，用来进行AM Container的启动与关闭。

1. ApplicationMasterLauncher回去创建一个AMLauncher实例，用来启动AM Container。

2. 将AMLauncher实例放入masterEvents队列。AMLauncher实例的处理是一种异步的处理方式。

3. 在实例化ApplicationMasterLauncher时，会创建一个LauncherThread线程实例，用来调度处理masterEvents。

4. LauncherThread从队列中获取AMLauncher调度执行。

5. 执行launch()方法，这里面包含了与NM交互，具体启动NM上Container的部分。

6. 向事件处理器发送RMAppAttemptEventType.LAUNCHED事件，调度执行RMAppAttemptImpl的AMLaunchedTransition()，此时认为AM container已经启动，等待AM container注册到RN上，将刚刚启动的 ApplicationMaster 注册到 AMLivelinessMonitor，启动心跳监控。RMAppAttemptState.ALLOCATED将变为 RMAppAttemptState.LAUNCHED。

NM上容器的启动流程

由ContainerManagerImpl被调用到startContainers()方法展开。遍历StartContainerRequests，处理每个StartContainerRequest。具体操作是在startContainerInternal() 方法。

下载资源

1. 创建一个ContainerImpl实例，初始化相关信息。
2. 为新的appc创建一个Application实例，即app的AM Container来初始化。
3. 获取日志聚合上下文，将Application信息，存储下来。
4. 向调度器发送ApplicationInitEvent(ApplicationEventType.INIT_APPLICATION)事件，调用AppInitTransition()方法，执行成功后会将ApplicationState.INITING变成ApplicationState.RUNNING
5. 发送LogHandlerAppStartedEvent(LogHandlerEventType.APPLICATION_STARTED)事件，由LogAggregationService来处理。会为APP创建一个AppLogAggregatorImpl实例，并由线程池调度它。
6. 向调度器发送ApplicationEvent(ApplicationEventType.APPLICATION_LOG_HANDLING_INITED)事件，触发AppLogInitDoneTransition()方法。
7. 向调度器发送ApplicationLocalizationEvent(LocalizationEventType.INIT_APPLICATION_RESOURCES)事件，由ResourceLocalizationService来处理。开始对app的创建应用程序资源跟踪。
8. 向调度器发送ApplicationInitedEvent(ApplicationEventType.APPLICATION_INITED)事件，调用ApplicationImpl中的AppInitDoneTransition方法处理。
9. 遍历此APP中的所有Container（需要建立的），向调度器发送ContainerInitEvent(ContainerEventType.INIT_CONTAINER)事件，由ContainerImpl的RequestResourcesTransition方法处理，ContainerState.LOCALIZING.NEW将变为ContainerState.LOCALIZING
10. 会发送辅助服务的相关事件AuxServicesEvent(AuxServicesEventType.CONTAINER_INIT)和AuxServicesEvent(AuxServicesEventType.APPLICATION_INIT)暂时展开，之后发送ContainerLocalizationRequestEvent事件，进行容器资源的本地化。
11. 资源本地服务ResourceLocalizationService为每个资源请求构建LocalResourcesTracker（LocalResourcesTrackerImpl）。
12. 由LocalResourcesTrackerImpl发送ResourceRequestEvent(ResourceEventType.REQUEST)事件，ResourceState.INIT将变为ResourceState.DOWNLOADING
13. 转发LocalizerResourceRequestEvent<br>(LocalizerEventType.REQUEST_RESOURCE_LOCALIZATION)事件给ResourceLocalizationService
14. 按照资源的分类交交由PublicLocalizer和LocalizerRunner两个线程进行处理，此时该容器应该是没有对应的LocalizerRunner的，所以先进行LocalizerRunner实例化之后再启动。将PRIVATE和APPLICATION资源交给LocalizerRunner。
15. 请求公共资源下载交给PublicLocalizer。
16. 发送ResourceLocalizedEvent(ResourceEventType.LOCALIZED)事件，交由LocalizedResource处理。
17. 没当一个下载完成之后，发送ContainerEventType.RESOURCE_LOCALIZED事件，触发该容器的全部资源的检查，如果还有资源没下载，就保持ContainerState.LOCALIZING，等待下次事件触发。
18. 如果全部资源下载完成，则发送ContainerLocalizationEvent(LocalizationEventType.CONTAINER_RESOURCES_LOCALIZED)事件给ResourceLocalizationService，去销毁该容器的LocalizerRunner线程，
19. 发送ContainerSchedulerEvent<br>(ContainerSchedulerEventType.SCHEDULE_CONTAINER)事件给ContainerScheduler。

此时ContainerState.LOCALIZING变为ContainerState.SCHEDULED。

AM Container 启动与运行

1. 尝试启动pending的容器们，在资源满足情况的前提下，遍历容器，按队列顺序向ContainerImpl发送启动事件，直到无法满足。

2. 向容器启动器发送ContainersLauncherEvent<br>(ContainersLauncherEventType.LAUNCH_CONTAINER)事件

3. 创建ContainerLaunch实例，并由线程池启动该实例线程

4. 更新容器信息到对应的日志目录和工作目录，将待运行的 Container 所需的环境和运行命令写到 Shell 脚本中launch_container.sh 脚本中，并将启动该脚本的命令写入 default_container_executro.sh 中，然后通过该脚本启动 Container。在这里主要对执行的环境和内容进行相关配置。ContainerState.SCHEDULED将变为ContainerState.RUNNING

5. 发送ContainerStartMonitoringEvent事件，开启容器监控。此刻以及认为容器以及启动了。

6. 将容器的启动信息更新进持久化存储

7. 发送启动命令给默认容器执行器（DefaultContainerExecutor），真正启动 Container。

   在DefaultContainerExecutor的launchContainer方法中会执行bash default_container_executor.sh命令，default_container_executor.sh脚本的包含部分内容有:，其中huatuo是队列名称,/data10/yarn是配置yarn.nodemanager.local-dirs的值。

   /bin/bash "/data10/yarn/usercache/huatuo/appcache/application_1723033197835_26906/container_e87_1723033197835_26906_01_000001/default_container_executor.sh"
   

   调用了default_container_executor_session.sh脚本

   #!/bin/bash
   ...
   exec /bin/bash "/data10/yarn/usercache/huatuo/appcache/appcache/application_1723033197835_26906/container_e87_1723033197835_26906_01_000001/launch_container.sh"
   

   最后调用了launch_container.sh脚本，内容如下:

   ...
   echo "Launching container"
   exec /bin/bash -c "$JAVA_HOME/bin/java -server -Djava.net.preferIPv6Addresses=false -XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED -Djdk.reflect.useDirectMethodHandle=false -Xmx2048m -Djava.io.tmpdir=$PWD/tmp '-XX:MaxMetaspaceSize=512m' '-XX:+PrintGCDetails' '-XX:+PrintGCDateStamps' '-XX:+UseParNewGC' '-XX:+UseConcMarkSweepGC' '-XX:CMSInitiatingOccupancyFraction=70' '-XX:+UseCMSInitiatingOccupancyOnly' '-XX:+ExplicitGCInvokesConcurrent' '-XX:ParallelGCThreads=4' '-XX:ConcGCThreads=2' '-XX:GCTimeLimit=90' '-XX:GCHeapFreeLimit=10' '-XX:OnOutOfMemoryError=kill %p' '-Dfastjson.parser.safeMode=true' -Xloggc:/data10/logs/application_1723033197835_26906/container_e87_1723033197835_26906_01_000001/gclog -Dspark.yarn.app.container.log.dir=/data10/logs/application_1723033197835_26906/container_e87_1723033197835_26906_01_000001 org.apache.spark.deploy.yarn.ApplicationMaster --class 'com.suning.bigquery.worker.BigqueryWorker' --jar file:/home/bigquery/software/bigquery-3.0.2/lib/worker/bigquery-worker-3.0.2.jar --arg 'huatuo' --arg '0' --arg 'namenode1-sit.cnsuning.com:2015,namenode2-sit.cnsuning.com:2015,slave01-sit.cnsuning.com:2015' --properties-file $PWD/__spark_conf__/__spark_conf__.properties --dist-cache-conf $PWD/__spark_conf__/__spark_dist_cache__.properties 1> /data10/logs/application_1723033197835_26906/container_e87_1723033197835_26906_01_000001/stdout 2> /data10/logs/application_1723033197835_26906/container_e87_1723033197835_26906_01_000001/stderr"
   

   这个container是个Spark任务，所以这里调用的是org.apache.spark.deploy.yarn.ApplicationMaster，并将标准输出写到stdout中，将标准错误输出写到stderr中。这也就是container的log目录里有三个文件的原因。至此，AM Container启动完毕。

NM RM心跳交互触发调度更新Container的状态

RMContainerState.ACQUIRED, RMContainerState.RUNNING

AM启动之后怎么注册到RM上

不同的任务类型在AM Container 启动的类不同，由这个l类管理进行AM注册到RM上

• mapreduce任务的AM上启动类：org.apache.hadoop.mapreduce.v2.app.MRAppMaster

• spark任务的AM上启动类：org.apache.spark.deploy.yarn.ApplicationMaster

• flink任务的AM上启动类：org.apache.flink.yarn.entrypoint.YarnApplicationClusterEntryPoint

以MRAppMaster为例：

1. Container启动的时候会去加载实例化MRAppMaster类，由MRAppMaster类中去创建RMContainerAllocator。
2. 通过服务框架进行服务的初始化
3. 通过服务框架进行服务的启动
4. 与RM基于ApplicationMasterProtocol协议建立连接，调用执行ApplicationMasterService的registerApplicationMaster方法，
5. 通过AMS处理链进行转发
6. 获取节点上的需要更新的Container，即是获取nodeUpdateQueue队列中的元素，其中包含了新启动的Container。
7. 触发 RMAppAttemptRegistrationEvent (RMAppAttemptEventType.REGISTERED) 事件，执行AMRegisteredTransition方法，RMAppAttemptState.ALLOCATED将变成RMAppAttemptState.RUNNING,
8. 触发 RMAppEvent (RMAppEventType.ATTEMPT_REGISTERED) 事件，执行方法AMRegisteredTransition()，RMAppState.ACCEPTED将变成RMAppState.RUNNING

AM怎么申请运行其他Container

以MRAppMaster为例：

在MRAppMaster启动中会创建一个JobImpl，由JobImpl计算出需要的MapTasks和ReduceTasks的个数。在RMCommunicator中会独立有一个AllocatorRunnable线程，在注册AM之后，会启动该线程，通过心跳机制，请求与RM中ApplicationMasterService服务的allocate方法，进行资源申请的调度。

1~4为AM注册到RM上流程（同上）

5. 在注册AM的之后，会启动分配器线程
6. AllocatorRunnable与RMContainerAllocator之前有心跳交互
7. 开始触发向RM申请接下来任务需要容器的资源流程
8. 请求与RM中ApplicationMasterService服务的allocate方法，获取分配结果，去开始对task进行分配和信息填充。
9. 检查分配的Container信息情况
10. 为task构建启动Context信息
11. 触发ContainerRemoteLaunchEvent(ContainerLauncher.EventType.CONTAINER_REMOTE_LAUNCH)事件，准备向分配的目标NodeManager发送启动容器的请求
12. 向分配的目标NodeManager发送启动容器请求
13. 从之前一些注册的心跳回调方法的队列heartbeatCallbacks中获取Runnable并执行

NM上的APP相关的任务会通过心跳机制来进行交互，更新运行情况。

【Yarn】Yarn的基本执行流程（三） 应用运行结束流程

这篇关于【Yarn】Yarn的基本执行流程（二）AM Container的启动的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---