Yarn中RMApp、RMAppAttempt、RMContainer和RMNode状态机及其状态转移

概述

        YARN中，如果一个对象由若干个状态以及触发这些状态发生转移的事件构成，它将被抽象成一个状态机，在YARN的ResourceManager内部，共有四类状态机，分别是RMApp，RMAppAttempt，RMContainer和RMNode。其中，前2类状态机维护了一个应用程序相关的生命周期，包括Application生命周期，一次进行尝试的生命周期；RMContainer则维护了分配出去的各个资源的使用状态；RMNode维护了一个NodeManager的生命周期。以上四个状态机，继承了EventHandler的Interface形式存在于Yarn源码的org.apache.hadoop.yarn.server.resourcemanager中；其具体实现类，则是对应的xxxImpl类。

        YARN中的Application生命周期由状态机RMAppImpl维护，每个Application可能会尝试运行多次，每次成为一次"运行尝试"，而每次运行尝试的生命周期则由状态机RMAppAttemptImpl维护，如果一次运行尝试运行失败，RMApp会创建另外一个运行尝试，知道某次运行尝试运行成功或者达到运行尝试上限。对于每次运行尝试，ResourceManager将为它分配一个Container，Container是运行环境的抽象，内部封装了任务的运行环境和资源等信息，而一个应用程序的ApplicationMaster就运行在这个Container中。ApplicationMaster启动之后，会不断向ResourceManager申请Container以运行各类任务。Container的生命周期由状态机RMContainerImpl维护。具体组织结构如下：

        Application Attempt的生命周期与ApplicationMaster的生命周期基本上是一致的 ： 一个Application内部所有任务均由ApplicationMaster维护和管理，ApplicationMaster本身需要占用一个Container，而这个Container由ResourceManager为其申请和启动。一旦ApplicationMaster成功启动，他就会与ResourceManager通信，为它内部的任务申请Container。如果ApplicationMaster重新启动，则意味着一个新的Application Attempt被启动，换句话说，一个Application Attempt的"生死存亡"与ApplicationMaster的"命运"紧紧绑定在一起。

 

RMApp状态机

         在RM内部，维护着所有Application的状态。对于每个Application，都有一个RMApp对象与之对应。在RMApp的实现类RMAppImpl中，维护着对象的基本信息，包括起始时间，名字，用户，组等信息，其中最复杂的部分莫过于其维护的状态机。

// RMAppImpl
private static final StateMachineFactory<RMAppImpl,
                                           RMAppState,
                                           RMAppEventType,
                                           RMAppEvent> stateMachineFactory
                               = new StateMachineFactory<RMAppImpl,
                                           RMAppState,
                                           RMAppEventType,
                                           RMAppEvent>(RMAppState.NEW)

    // Transitions from NEW state
    .addTransition(RMAppState.NEW, RMAppState.NEW,
        RMAppEventType.NODE_UPDATE, new RMAppNodeUpdateTransition())
    .addTransition(RMAppState.NEW, RMAppState.NEW_SAVING,
        RMAppEventType.START, new RMAppNewlySavingTransition())
    // ......

其状态机转换如下图所示：

状态与转换解释

• NEW：RMApp的初始状态；
  • 当客户端通过RPC调用RM的submitApplication方法后，RM会初始化RMAppImpl，此时状态机的状态被设置为NEW。
• NEW_SAVING：表示RM在处理客户端提交作业的请求期间状态为NEW_SAVING；
  • RM的submitApplication方法中，在返回给客户端前，RM会创建START事件，当NEW状态遇到START事件后，RMAppImpl的状态转换为NEW_SAVING。
• SUBMITTED：表示App已经提交成功，RM已经存下该App状态。两种情况下会转换为SUBMITTED状态：
  • 当RMAppImpl的状态由NEW转换为NEW_SAVING期间，会触发RMAppNewlySavingTransition的transition方法，在次方法中会调用RMStateStore存储RMAppImpl，事实上是在RMStateStore.ForwardingEventHandler中调用handleStoreEvent方法存储RMAppImpl，完成存储后会调用notifyDoneStoringApplication方法，在此方法中创建RMAppNewSavedEvent事件并交给rmDispatcher。RMAppImpl遇到RMAppNewSavedEvent（对应APP_NEW_SAVED）事件后，状态转换为SUBMITTED。
  • 在NEW状态下，如果是Recover模式，且改App存储在RMStateStore中，则转换为SUBMITTED。
• ACCEPTED：表示该App已经提交给调度器。
  • 在NEW_SAVING转换为SUBMITTED状态的时候，RMAppImpl会触发StartAppAttemptTransition，这时会创建一个新的RMAppAttempt，然后新建RMAppAttemptEventType.START事件给处理器，经过RMAppAttempt处理机。当该RMAppAttempt交给调度器（的某个组）后，状态改为ACCEPTED。
• RUNNING：AM已经启动并注册到RM上。两种情况下会转换为RUNNING状态：
  • AM启动后会向RM注册，这时候会触发RMAppImpl状态转换为RUNNING状态。
  • RMAppImpl也有可能在Recovery模式下转换为RUNNING。
• FINAL_SAVING：FINAL_SAVING状态表示正在保存RMAppImpl到存储器，目的是保证RMAppImpl的状态已经存储下来，当RMStateStore在完成App状态更新到存储器后会根据App的状态转换为最终状态，包括FAILED，FINISHED，FINISHING，KILLED。
• FINISHING：FINISHING状态表示RM上相应的App状态已经完成存储工作，在等待RMAppEventType.ATTEMPT_FINISHED事件。因为只有RMAppAttempt结束后RMApp才能结束。在RMAppState.RUNNING状态遇到RMAppEventType.ATTEMPT_UNREGISTERED事件时，RMAppImpl转换为FINAL_SAVING，并存储targetedFinalState为RMAppState.FINISHING，遇到RMAppEventType.APP_UPDATE_SAVED事件后RMAppImpl转换为FINISHING状态。
• FINISHED：RMAppImpl的结束状态（另外两个个结束状态是KILLED和FAILED），正常情况下，处于RUNNING的RMAppImpl成功结束后状态就是FINISHED，另外RM收到AM的REJECTED请求后最终状态也是FINISHED，即FINISHED状态是AM主动通知RM自己结束后的状态。
• FAILED：处于FINAL_SAVING的RMAppImpl遇到RMAppEventType. FAILED事件后RMAppImpl转换为FAILED状态。
• KILLING：RMAppImpl遇到客户端执行KILL操作后会转换为FINAL_SAVING状态，另外会设置RMAppImpl的targetedFinalState为RMAppEventType.KILL。
• KILLED：处于FINAL_SAVING的RMAppImpl遇到RMAppEventType.APP_UPDATE_SAVED事件后RMAppImpl转换为KILLED状态。

RMAppAttempt状态机

        在RM中，一个RMApp可能对于一到多个RMAppAttempt，即假如RMApp的第一个RMAppAttempt失败后，RM会根据配置启动新的RMAppAttempt。RMAppAttempt内部维护了进度，运行的机器信息，起始时间，URL等信息。其中最复杂的信息是RMAppAttempt的状态机信息。

// RMAppAttemptImpl
private static final StateMachineFactory<RMAppAttemptImpl,
                                           RMAppAttemptState,
                                           RMAppAttemptEventType,
                                           RMAppAttemptEvent>
       stateMachineFactory  = new StateMachineFactory<RMAppAttemptImpl,
                                            RMAppAttemptState,
                                            RMAppAttemptEventType,
                                     RMAppAttemptEvent>(RMAppAttemptState.NEW)

       // Transitions from NEW State
      .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.SUBMITTED,
          RMAppAttemptEventType.START, new AttemptStartedTransition())
      .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.FINAL_SAVING,
          RMAppAttemptEventType.KILL,
          new FinalSavingTransition(new BaseFinalTransition(
            RMAppAttemptState.KILLED), RMAppAttemptState.KILLED))
      // ......

RMAppAttempt状态与转换解释

• NEW：和RMApp的状态机一样，RMAppAttempt的初始状态也是NEW。
  • 在RMApp由NEW_SAVING转换为RMAppState.SUBMITTED时，RMApp会创建RMAppAttemptImpl对象，这时RMAppAttempt状态为NEW。
• SUBMITTED：表示RMAppAttempt已经创建，并准备给调度器。
  • 在RMApp创建RMAppAttempt时会创建一个RMAppAttemptEventType.START事件，交给处理器，RMAppAttempt在NEW的状态下遇到RMAppAttemptEventType.START事件时，转化为SUBMITTED状态，并调用AttemptStartedTransition。
• SCHEDULED：表示该RMAppAttempt已经交给调度器，做好了调度的准备。
  • AttemptStartedTransition中，RMAppAttempt会创建AppAttemptAddedSchedulerEvent事件，该事件最终会传递给调度器；调度器把相应的RMAppAttempt放到待调度的组中，然后创建RMAppAttemptEventType.APP_ACCEPTED事件，这个事件会使RMAppAttempt调用ScheduleTransition，在ScheduleTransition中，根据该App是否为UnmanagedAM（AM是否被RM管理，UnmanagedAM由客户端负责启动，然后向RM注册；managedAM由RM命令NM启动，并负责该AM的生命周期）决定返回LAUNCHED_UNMANAGED_SAVING或SCHEDULED状态，如果是managedAM（大多数情况）返回SCHEDULED，否则返回LAUNCHED_UNMANAGED_SAVING状态。
• ALLOCATED_SAVING：表示正在保存RMAppAttemptImpl信息到RMStateStore中。
  • 当某个NM(NodeManager)发送心跳到RM，RM会给改RM分配相应的container，这时候会创建RMContainerEventType.START事件，在RMContainerImpl状态机（后续将会介绍）遇到这个事件时会调用ContainerStartedTransition，在其转换方法中会创建RMAppAttemptEventType.CONTAINER_ALLOCATED事件，而RMAppAttemptImpl（SCHEDULED状态）遇到SCHEDULED事件后会调用AMContainerAllocatedTransition，这时会把RMAppAttemptImpl信息存储到制定的存储器中。RMAppAttempt状态转换为ALLOCATED_SAVING。
• ALLOCATED：表示RMAppAttempt已经完成调度到某台NM上的工作。
  • 当RMStateStore完成RMAppAttemptImpl信息的存储后，会创建RMAppAttemptEventType.ATTEMPT_NEW_SAVED事件，RMAppAttemptImpl遇到此事件后会调用AttemptStoredTransition，检查存储是否出错，并创建AMLauncherEventType.LAUNCH事件，然后把状态转换为ALLOCATED。
• LAUNCHED：表示RM已经发送命令让某台NM启动相应的AM。
  • 在AMLauncher中，处理AMLauncherEventType.LAUNCH事件，实际的方法lanch()方法中，AM首先连接上相应的NM，然后通过相应的协议让NM启动AM。完成后生成RMAppAttemptEventType.LAUNCHED事件。RMAppAttemptImpl处理此事件，调用AMLaunchedTransition，然后转换状态为LAUNCHED。
• RUNNING：AM成功向RM注册后的状态。
  • AM启动后会向RM注册，注册的时候RM会生成RMAppAttemptEventType.REGISTERED事件，RMAppAttemptImpl处理这个事件，调用AMRegisteredTransition，获取AM的host，port，URL等信息，创建RMAppEventType.ATTEMPT_REGISTERED事件，然后把RMAppAttemptImpl状态转换为RUNNING。
• LAUNCHED_UNMANAGED_SAVING：Unmanaged AM正在保存状态，见SCHEDULED状态。
• FINAL_SAVING：表示正在保存RMAppAttemptImpl的最终信息。
  • 这个状态类似于RMApp的FINAL_SAVING状态，即为了保证存RMAppAttemptImpl信息，在转换为FINISHED，FAILED，FINISHING，KILLED前增加这个状态，在转换为FINAL_SAVING前，需要保存RMAppAttemptImpl的 targetedFinalState状态，在RMStateStore完成相应存储后根据targetedFinalState转化为后续的状态。
• FINISHING： 表示RMAppAttemptImpl正在结束。
  • 遇到RMAppAttemptEventType.EXPIRE或AM的RMAppAttemptEventType.CONTAINER_FINISHED事件时，RMAppAttemptImpl状态转换为FINISHED
• FINISHED：表示RMAppAttemptImpl已经结束。
  • 只有AM主动汇报自己结束的RMAppAttempt才会标识为FINISHED状态。
• FAILED：表示RMAppAttemptImpl已经失败。
• KILLED：表示RMAppAttemptImpl已经被kill。
  • 客户端发起KILL会导致RMAppAttemptImpl被KILL。

RMContainer状态机

        RMContainer是RM内部维护的Container状态。事实上在RM的调度器中，会维护着一个liveContainers列表，保存着所有存活着的Container信息。RMContainer主要维护了containerID、ApplicationAttemptId、state、reserved等信息。作为RMContainer实现类，RMContainerImpl维护了一个状态机：

// RMContainerImpl
private static final StateMachineFactory<RMContainerImpl, RMContainerState, 
                                         RMContainerEventType, RMContainerEvent> 
   stateMachineFactory = new StateMachineFactory<RMContainerImpl, 
       RMContainerState, RMContainerEventType, RMContainerEvent>(
      RMContainerState.NEW)

    // Transitions from NEW state
    .addTransition(RMContainerState.NEW, RMContainerState.ALLOCATED,
        RMContainerEventType.START, new ContainerStartedTransition())
    .addTransition(RMContainerState.NEW, RMContainerState.KILLED,
        RMContainerEventType.KILL)
    // ......

RMContainer状态与转换解释

• NEW：RMContainerImpl的初始状态，当调度器初始化一个RMContainerImpl时，其状态被设置为new 。
• RESERVED：表示RMContainerImpl已经预订了某个NM的资源。
  • 当调度器准备把某个container分配给相应的NM，当时这时NM上的资源不能满足container的需求，这时调度器则会让container预订此NM，然后创建一个RMContainerEventType.RESERVED事件，RMContainerImpl会调用ContainerReservedTransition处理这个事件，把预订信息（资源，节点，优先级）保存下来，然后设置自己的状态为RESERVED。当然有可能出现多次预订，多次预订后还是RESERVED状态。
• ALLOCATED：表示RMContainerImpl处于已经分配的状态。
  • 同样的，无论RMContainerImpl处于NEW还是RESERVED，当调度器准备把某个container分配给相应的NM，且NM的资源能满足container的需求，则会把相应的container标记调度到NM上，并创建RMContainerEventType.START事件，RMContainerImpl会调用ContainerStartedTransition，创建RMAppAttemptEventType.CONTAINER_ALLOCATED（RMAppAttempt状态机详解中有介绍）事件，然后RMContainerImpl状态被设置为ALLOCATED。
• ACQUIRED：表示已经分配资源的Container已经被通知到AM。
  • AM（ApplicationMaster）通过ApplicationMasterProtocol.Allocate()向RM发起资源请求，RM会调用调度器处理AM的请求，在调度器中首先会把请求资源保存下来，然后把已经分配的资源（AM上次请求的资源已经得到了分配）返回给AM，这期间调度器会生成RMContainerEventType.ACQUIRED事件，RMContainerImpl调用AcquiredTransition处理这个事件，生成RMAppAttemptEventType.CONTAINER_ACQUIRED事件，然后RMContainerImpl状态改为ACQUIRED状态。
• RUNNING：表示RMContainerImpl已经处于运行状态。
  • 当NM发送心跳给RM，NM会把自己节点上运行的container列表汇报给RM，RM让调度器负责处理，在NM汇报的container列表中包含刚刚运行的container，调度器处理这些container的时候会生成RMContainerEventType.LAUNCHED事件，RMContainerImpl会调用LaunchedTransition处理此事件，然后RMContainerImpl状态改为RUNNING。
• COMPLETED：表示RMContainerImpl已经运行结束。
  • 当NM发送心跳给RM，包含了已经结束了container，这时候调度器处理这些container的时候会生成RMContainerEventType.FINISHED事件，RMContainerImpl会调用FinishedTransition处理此事件，生成RMAppAttemptEventType.CONTAINER_FINISHED事件，然后RMContainerImpl改为COMPLETED。
• EXPIRED：表示RMContainerImpl由于超时
  • 当RMContainerImpl由ALLOCATED转换ACQUIRED期间，RMContainerImpl会调用AcquiredTransition，在AcquiredTransition中，container注册到containerAllocationExpirer，当container超过一定时间（默认600000ms）没有成为RUNNING状态，则会创建SchedulerEventType.CONTAINER_EXPIRED事件，调度器负责处理此事件，并创建RMContainerEventType.EXPIRE事件，RMContainerImpl调用FinishedTransition事件处理，状态改为EXPIRED。
• RELEASED：表示RMContainerImpl对应的资源被AM主动释放。
  • AM通过allocate发送心跳给RM，其中包含了需要释放的资源（通过container对应），这时调度器会创建RMContainerEventType.RELEASED事件，RMContainerImpl处理此事件，然后状态改为RELEASED。
• KILLED：表示RMContainerImpl被KILL。
  • 当container对应的Application已经结束，或者客户端或者AM发起kill ，相应的RMContainerImpl状态会转换为KILLED。

RMNode状态机

        RMNode状态机是ResourceManager的四个状态机（RMApp，RMAppAttempt，RMContainer，RMNode）中最简单的一个，其状态机定义如下所示。RMNode是ResourceManager用于保存NM信息的类，包括NM的机器名，地址，端口，状态等信息。

// RMNodeImpl
private static final StateMachineFactory<RMNodeImpl,
                                           NodeState,
                                           RMNodeEventType,
                                           RMNodeEvent> stateMachineFactory 
                 = new StateMachineFactory<RMNodeImpl,
                                           NodeState,
                                           RMNodeEventType,
                                           RMNodeEvent>(NodeState.NEW)
  
     //Transitions from NEW state
     .addTransition(NodeState.NEW, NodeState.RUNNING, 
         RMNodeEventType.STARTED, new AddNodeTransition())
     .addTransition(NodeState.NEW, NodeState.NEW,
         RMNodeEventType.RESOURCE_UPDATE, 
         new UpdateNodeResourceWhenUnusableTransition())
     // ......

RMNode状态与转换详解

• NEW：每个NM启动的时候都会通过ResourceTracker协议向RM注册，RM便会创建RMNodeImpl对象，这时RMNodeImpl的状态被初始化为NEW。
• RUNNING：在RM处理NM注册期间，新建RMNodeImpl对象后，会创建RMNodeEventType.STARTED事件，RMNodeImpl调用AddNodeTransition处理此事件，这时会创建SchedulerEventType.NODE_ADDED和NodesListManagerEventType.NODE_USABLE事件，然后RMNodeImpl状态改为RUNNING。
• UNHEALTHY：NM每次汇报心跳给RM的时候都需要传递其状态给RM，RM处理NM心跳的时候会创建RMNodeEventType.STATUS_UPDATE事件，RMNodeImpl会调用StatusUpdateWhenHealthyTransition处理此事件，如果NM传递过来的状态是UNHEALTHY，则RMNodeImpl状态被设置为UNHEALTHY，否则设置为RUNNING。
  另外如果RMNodeImpl处于UNHEALTHY状态，当收到RMNodeEventType.STATUS_UPDATE事件后，RMNodeImpl调用StatusUpdateWhenUnHealthyTransition处理，并根据NM传递过来的状态设置RMNodeImpl的状态。
• DECOMMISSIONED：当NM发送心跳给RM的时候，RM会检查这个节点是否属于合法节点（如在exclude文件中的节点就属于不合法节点），如果不合法，则创建RMNodeEventType.DECOMMISSION事件，RMNodeImpl调用DeactivateNodeTransition处理此事件，RMNodeImpl被置为DECOMMISSIONED。
• LOST：与RMContainerImpl的LOST状态类似，当Node超过一段时间（默认600000ms）没有发送心跳则RMNodeImpl会被标记为LOST。
• REBOOTED：当RM处理NM心跳时，如果NM传递过来的心跳号码与RM保存的心跳号码不一致（相差大于1），则RM会认定这个NM重启了，这时RM会创建RMNodeEventType.REBOOTING事件，RMNodeImpl会设置自己状态为REBOOTED。