yarn的主要功能是什么（【深入浅出 Yarn 架构与实现】4-2 RM 管理 Application Master）

上一篇文章对 ResourceManager 整体架构和功能进行了讲述    
    
    
。本篇将对 RM 中管理 Application Master 的部分进行深入的讲解  


     


     


    。

下面将会介绍 RM 与 AM 整体通信执行流程 
    
    
    ，并对 RM 中涉及的对应服务进行具体讲解


 




 




 


。

为了更好的学习本篇知识
     


     


     


，建议先熟悉以下知识点 




 




 




，不了解的部分可翻到前面对应的文章进行学习： RPC（2-2 Yarn 基础库 - 底层通信库 RPC） 事件处理器（2-3 Yarn 基础库 - 服务库与事件库） AM 程序执行流程（3-3 Yarn Application Master 编写）

一 
    
    
    、AM 执行流程

客户端提交任务到 RM 后 

   

   

   ，启动 AM 到任务完成的流程如下所示：

各个步骤具体执行操作请对应下面各服务讲解   

   

   

。

二
     


     


     


、AM 管理主要组成

ApplictionMaster 管理部分主要由三个服务构成
     

     

     

，它们共同管理应用程序的 AM 的生存周期   

     

     

   。

（以下服务均能根据名称找到源码中对应的类  




  




  




，可以看其具体的实现逻辑）

一）ApplicationMasterLauncher

「服务&事件处理器」处理 AM 的 LAUNCH 和 CLEANUP 事件 从源码中可以看到：EventHandler 的 handle 方法收到 AM 事件后创建 Runnable 对象 


  


  


  ，之后会放到 masterEvents 阻塞队列中
     
     
     
，launcherHandlingThread 不断从队列中取出事件   




   




   




，提交到线程池 launcherPool 中处理



  




  




  

。（流程图如下所示）

二）AMLivelinessMonitor

检查服务活性（是否有心跳） 继承自抽象类 AbstractLivelinessMonitor 



 



 



 ，在抽象类中已经实现好 live 检查逻辑               ，在一段时间内未汇报心跳信息
    




    




    




，则任务其挂了  


  


  


。AMLivelinessMonitor 只需定义当 AM 被认为挂了（expire）时














，需要处理的逻辑    
     
     
  。 当失败时会发一个 RMAppAttemptEvent EXPIRE 事件




   




   




   
。

抽象类 AbstractLivelinessMonitor 简要介绍：

public abstract class AbstractLivelinessMonitor<O> extends AbstractService { // 里面最重要的检查函数 // 定期遍历记录的 list                ，看是否有超时的 // 检查周期默认为超时时间的 1/3 private class PingChecker implements Runnable { @Override public void run() { while (!stopped && !Thread.currentThread().isInterrupted()) { synchronized (AbstractLivelinessMonitor.this) { Iterator<Map.Entry<O, Long>> iterator = running.entrySet().iterator(); //avoid calculating current time everytime in loop long currentTime = clock.getTime(); while (iterator.hasNext()) { Map.Entry<O, Long> entry = iterator.next(); if (currentTime > entry.getValue() + expireInterval) { iterator.remove(); expire(entry.getKey()); LOG.info("Expired:" + entry.getKey().toString() + " Timed out after " + expireInterval/1000 + " secs"); } } } try { Thread.sleep(monitorInterval); } catch (InterruptedException e) { LOG.info(getName() + " thread interrupted"); break; } } } }

三）ApplicationMasterService

是 RM RPC 服务端 ApplicationMasterProtocol 的实现类 



 



 



。 接收处理来自 AM 的请求：主要包括注册 




 




 




、心跳 

   

   

   、清理三类                。 心跳通过 ApplicationMasterProtocol#allocate 方法定期调用实现
     



     



     



，主要作用： 请求资源 获取新分配的资源 定期告诉 RM 其还活着（心跳）

三
     

     

     

、小结

本篇主要介绍了 RM 中对 AM 的管理部分




    




    




    。首先介绍了 RM 相关组件与 AM 交互流程














，之后对各服务执行逻辑  




  




  




、RPC 调用等进行了详细的介绍















。本篇中仅对 ApplicationMasterLauncher 组件进行了详细讲解 
    
    
    ，并绘图说明
     


     


     


，其余部分各位同学感兴趣可自行梳理               。

在学习这部分知识时 




 




 




，建议对照源码进行梳理 

   

   

   ，可以更好的了解其中的流程 



     



     



     。