ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > 数据分析 > Spark on Yarn 任务提交流程源码分析

Spark on Yarn 任务提交流程源码分析

原创 数据分析 作者：破棉袄 时间：2015-10-21 10:28:21 0 删除 编辑

流程分析，以作备忘：

shell调用org.apache.spark.deploy.SparkSubmit   >  org.apache.spark.deploy.yarn.Client


打开org.apache.spark.deploy.yarn.Client的main方法：
  1. def main(argStrings: Array[String]) {
  2.     if (!sys.props.contains("SPARK_SUBMIT")) {
  3.       logWarning("WARNING: This client is deprecated and will be removed in a " +
  4.         "future version of Spark. Use ./bin/spark-submit with \"--master yarn\"")
  5.     }

  7.     // Set an env variable indicating we are running in YARN mode.
  8.     // Note that any env variable with the SPARK_ prefix gets propagated to all (remote) processes
  9.     System.setProperty("SPARK_YARN_MODE", "true")
  10.     val sparkConf = new SparkConf

  12.     val args = new ClientArguments(argStrings, sparkConf)
  13.     // to maintain backwards-compatibility
  14.     if (!Utils.isDynamicAllocationEnabled(sparkConf)) {
  15.       sparkConf.setIfMissing("spark.executor.instances", args.numExecutors.toString)
  16.     }
  17.     new Client(args, sparkConf).run()
  18.   }
进入run（）方法：
  1. def run(): Unit = {
  2.     val appId = submitApplication()
  3.    //判断是否等待完成，根据提交模式判断（client、cluster两种模式）  
  4.    // private val fireAndForget = isClusterMode && !sparkConf.getBoolean("spark.yarn.submit.waitAppCompletion", true)
  5.    if (fireAndForget) {                             
  6.       val report = getApplicationReport(appId)                               //ApplicationReport是应用程序的报告（包括程序用户、程序队列、程序名称等等）
  7.       val state = report.getYarnApplicationState                             //得到应用程序的完成状态
  8.       logInfo(s"Application report for $appId (state: $state)")
  9.       logInfo(formatReportDetails(report)) 
  10.       if (state == YarnApplicationState.FAILED || state == YarnApplicationState.KILLED) {    //判断状态
  11.         throw new SparkException(s"Application $appId finished with status: $state")
  12.       }
  13.     } else {
  14.       val (yarnApplicationState, finalApplicationStatus) = monitorApplication(appId)               //client模式提交的处理 （博文后边细说此方法）  涉及两个对象YarnApplicationState（对于yarn来说任务的状态）、FinalApplicationStatus（对于任务来说任务的运行状态）
  15.       if (yarnApplicationState == YarnApplicationState.FAILED ||
  16.         finalApplicationStatus == FinalApplicationStatus.FAILED) {
  17.         throw new SparkException(s"Application $appId finished with failed status")
  18.       }
  19.       if (yarnApplicationState == YarnApplicationState.KILLED ||
  20.         finalApplicationStatus == FinalApplicationStatus.KILLED) {
  21.         throw new SparkException(s"Application $appId is killed")
  22.       }
  23.       if (finalApplicationStatus == FinalApplicationStatus.UNDEFINED) {
  24.         throw new SparkException(s"The final status of application $appId is undefined")
  25.       }
  26.     }
  27.   }
跟进submitApplication（）方法：
  1. def submitApplication(): ApplicationId = {
  2.     var appId: ApplicationId = null
  3.     try {
  4.       // Setup the credentials before doing anything else,
  5.       // so we have don't have issues at any point.
  6.       setupCredentials()
  7.       yarnClient.init(yarnConf)
  8.       yarnClient.start()

  10.       logInfo("Requesting a new application from cluster with %d NodeManagers"
  11.         .format(yarnClient.getYarnClusterMetrics.getNumNodeManagers))

  13.       // Get a new application from our RM
  14.       val newApp = yarnClient.createApplication()
  15.       val newAppResponse = newApp.getNewApplicationResponse()
  16.       appId = newAppResponse.getApplicationId()

  18.       // Verify whether the cluster has enough resources for our AM
  19.       verifyClusterResources(newAppResponse)                                  //内存判断   (博文后边细讲此方法)

  21.       // Set up the appropriate contexts to launch our AM
  22.       val containerContext = createContainerLaunchContext(newAppResponse)      // 构建ApplicationMaster的container(包括jar包路径 userClas等)
  23.       val appContext = createApplicationSubmissionContext(newApp, containerContext)

  25.       // Finally, submit and monitor the application
  26.       logInfo(s"Submitting application ${appId.getId} to ResourceManager")
  27.       yarnClient.submitApplication(appContext)
  28.       appId
  29.     } catch {
  30.       case e: Throwable =>
  31.         if (appId != null) {
  32.           cleanupStagingDir(appId)
  33.         }
  34.         throw e
  35.     }
  36.   }
看一下yarnClient什么来路：
  1. import org.apache.hadoop.yarn.client.api.{YarnClient, YarnClientApplication}

  2. private     val yarnClient = YarnClient.createYarnClient
找一下hadoop源码：
  1. @InterfaceAudience.Public
  2. @InterfaceStability.Stable
  3. public abstract class YarnClient extends AbstractService
  4. {
  5.   @InterfaceAudience.Public
  6.   public static YarnClient createYarnClient()
  7.   {
  8.     YarnClient client = new YarnClientImpl();
  9.     return client;
  10.   }

  12.   @InterfaceAudience.Private
  13.   protected YarnClient(String name) {
  14.     super(name);
  15.   }
  16.  ......
  17. }

看一下createApplication（）方法：
  1. public YarnClientApplication createApplication()
  2.     throws YarnException, IOException
  3.   {
  4.     ApplicationSubmissionContext context = (ApplicationSubmissionContext)Records.newRecord(ApplicationSubmissionContext.class);

  6.     GetNewApplicationResponse newApp = getNewApplication();
  7.     ApplicationId appId = newApp.getApplicationId();
  8.     context.setApplicationId(appId);
  9.     return new YarnClientApplication(newApp, context);
  10.   }

   private GetNewApplicationResponse getNewApplication() throws YarnException, IOException
  {
    GetNewApplicationRequest request = (GetNewApplicationRequest)Records.newRecord(GetNewApplicationRequest.class);

    return this.rmClient.getNewApplication(request);
  }

  看一下ApplicationClientProtocol的getNewApplication(request)方法：
 
  @InterfaceAudience.Public
  @InterfaceStability.Stable
  @Idempotent
  public abstract GetNewApplicationResponse getNewApplication(GetNewApplicationRequest paramGetNewApplicationRequest)
    throws YarnException, IOException;

最后再回到run（）方法看一下如果为client模式提交的处理逻辑 进入monitorApplication（）方法：

  2.   def monitorApplication(
  3.       appId: ApplicationId,
  4.       returnOnRunning: Boolean = false,
  5.       logApplicationReport: Boolean = true): (YarnApplicationState, FinalApplicationStatus) = {
  6.     val interval = sparkConf.getLong("spark.yarn.report.interval", 1000)    //app运行监控的间隔时间ms
  7.     var lastState: YarnApplicationState = null
  8.     while (true) {                                                          //写死 一直等到程序完成才返回
  9.       Thread.sleep(interval)
  10.       val report: ApplicationReport =
  11.         try {
  12.           getApplicationReport(appId)
  13.         } catch {
  14.           case e: ApplicationNotFoundException =>
  15.             logError(s"Application $appId not found.")
  16.             return (YarnApplicationState.KILLED, FinalApplicationStatus.KILLED)
  17.           case NonFatal(e) =>
  18.             logError(s"Failed to contact YARN for application $appId.", e)
  19.             return (YarnApplicationState.FAILED, FinalApplicationStatus.FAILED)
  20.         }
  21.       val state = report.getYarnApplicationState                        

  23.       if (logApplicationReport) {
  24.         logInfo(s"Application report for $appId (state: $state)")

  26.         // If DEBUG is enabled, log report details every iteration
  27.         // Otherwise, log them every time the application changes state
  28.         if (log.isDebugEnabled) {
  29.           logDebug(formatReportDetails(report))
  30.         } else if (lastState != state) {
  31.           logInfo(formatReportDetails(report))
  32.         }
  33.       }

  35.       if (state == YarnApplicationState.FINISHED ||
  36.         state == YarnApplicationState.FAILED ||
  37.         state == YarnApplicationState.KILLED) {
  38.         cleanupStagingDir(appId)
  39.         return (state, report.getFinalApplicationStatus)   //返回运行结果
  40.       }

  42.       if (returnOnRunning && state == YarnApplicationState.RUNNING) {
  43.         return (state, report.getFinalApplicationStatus)
  44.       }

  46.       lastState = state
  47.     }

  49.     // Never reached, but keeps compiler happy
  50.     throw new SparkException("While loop is depleted! This should never happen...")
  51.   }

至此，流程结束  yarn利用分布式缓存机制将application部署到各个计算节点

深入看一下verifyClusterResources(newAppResponse)方法：
  1. private def verifyClusterResources(newAppResponse: GetNewApplicationResponse): Unit = {
  2.     val maxMem = newAppResponse.getMaximumResourceCapability().getMemory()              //每个task最多可申请的内存 container的最大值
  3.     logInfo("Verifying our application has not requested more than the maximum " +
  4.       s"memory capability of the cluster ($maxMem MB per container)") 
  5.     val executorMem = args.executorMemory + executorMemoryOverhead     //1024(M) +设置的executor的值    args.executorMemory在1.5版本中为写死1024单位M 
  6.     if (executorMem > maxMem) {                   //如果executor所需要的内存大于container的最大值
  7.       throw new IllegalArgumentException(s"Required executor memory (${args.executorMemory}" +
  8.         s"+$executorMemoryOverhead MB) is above the max threshold ($maxMem MB) of this cluster! " +
  9.         "Please increase the value of 'yarn.scheduler.maximum-allocation-mb'.")
  10.     }
  11.     val amMem = args.amMemory + amMemoryOverhead         
  12.   //args.amMemory在1.5版本中为写死512单位M args.amMemoryOverhead:
  13.   //                                                            if (isClusterMode) driverMemOverheadKey else amMemOverheadKey
  14.   //                                                               val driverMemOverheadKey = "spark.yarn.driver.memoryOverhead"
      //                                                               val amMemOverheadKey = "spark.yarn.am.memoryOverhead"
     

  15.     if (amMem > maxMem) {
  16.       throw new IllegalArgumentException(s"Required AM memory (${args.amMemory}" +
  17.         s"+$amMemoryOverhead MB) is above the max threshold ($maxMem MB) of this cluster! " +
  18.         "Please increase the value of 'yarn.scheduler.maximum-allocation-mb'.")
  19.     }
  20.     logInfo("Will allocate AM container, with %d MB memory including %d MB overhead".format(
  21.       amMem,
  22.       amMemoryOverhead))

  24.     // We could add checks to make sure the entire cluster has enough resources but that involves
  25.     // getting all the node reports and computing ourselves.
  26.   }


   总结：

  Cluster模式:

    客户端操作：
　  1、SparkSubmit中根据yarnConf来初始化yarnClient，并启动yarnClient 　　
     2、创建客户端Application，并获取Application的ID，进一步判断集群中的资源是否满足executor和ApplicationMaster申请的资源，如果不满足则抛出IllegalArgumentException；
　  3、设置资源、环境变量：其中包括了设置Application的Staging目录、准备本地资源（jar文件、log4j.properties）、设置Application其中的环境变量、创建Container启动的Context等；
　  4、设置Application提交的Context，包括设置应用的名字、队列、AM的申请的Container、标记该作业的类型为spark；
　  5、申请Memory，并最终通过submitApplication方法向ResourceManager提交该Application。　　
       当作业提交到YARN上之后，客户端就没事了，会关闭此进程，因为整个作业运行在YARN集群上进行，运行的结果将会保存到HDFS或者日志中。
    Yarn操作：
      1、运行ApplicationMaster的run方法；
　　2、设置好相关的环境变量。
　　3、创建amClient，并启动；
　　4、在Spark UI启动之前设置Spark UI的AmIpFilter；
　　5、在startUserClass函数专门启动了一个线程（名称为Driver的线程）来启动用户提交的Application，也就是启动了Driver。在Driver中将会初始化SparkContext；
　　6、等待SparkContext初始化完成，最多等待spark.yarn.applicationMaster.waitTries次数（默认为10），如果等待了的次数超过了配置的，程序将会退出；否则用SparkContext初始化yarnAllocator；
               怎么知道SparkContext初始化完成？
　　         其实在5步骤中启动Application的过程中会初始化SparkContext，在初始化SparkContext的时候将会创建YarnClusterScheduler，在SparkContext初始化完成的时候，
               会调用YarnClusterScheduler类中的postStartHook方法，而该方法会通知ApplicationMaster已经初始化好了SparkContext
          
              为何要等待SparkContext初始化完成？
                 CoarseGrainedExecutorBackend启动后需要向CoarseGrainedSchedulerBackend注册
　　7、当SparkContext初始化完成的时候，通过amClient向ResourceManager注册ApplicationMaster    
　　8、分配并启动Executeors。在启动Executeors之前，先要通过yarnAllocator获取到numExecutors个Container，然后在Container中启动Executeors。如果在启动Executeors的过程中失败的次数达到了maxNumExecutorFailures的次数，
　　那么这个Application将失败，将Application Status标明为FAILED，并将关闭SparkContext。其实，启动Executeors是通过ExecutorRunnable实现的，而ExecutorRunnable内部是启动CoarseGrainedExecutorBackend的，
      CoarseGrainedExecutorBackend启动后会向SchedulerBackend注册。(resourceManager是如何决定该分配几个container？  在shell提交时跟参数 默认启动两个executor)
　　9、最后，Task将在CoarseGrainedExecutorBackend里面运行，然后运行状况会通过Akka通知CoarseGrainedScheduler，直到作业运行完成。



     Client模式:

      客户端操作：
       1、通过SparkSubmit类的launch的函数直接调用作业的main函数（通过反射机制实现），如果是集群模式就会调用Client的main函数。
 　　2、而应用程序的main函数一定都有个SparkContent，并对其进行初始化；
 　　3、在SparkContent初始化中将会依次做如下的事情：设置相关的配置、注册MapOutputTracker、BlockManagerMaster、BlockManager，创建taskScheduler和dagScheduler；其中比较重要的是创建taskScheduler和dagScheduler。在创建taskScheduler的时候会根据我们传进来的master来选择Scheduler和SchedulerBackend。由于我们选择的是yarn-client模式，程序会选择YarnClientClusterScheduler和YarnClientSchedulerBackend，并将YarnClientSchedulerBackend的实例初始化YarnClientClusterScheduler，上面两个实例的获取都是通过反射机制实现的，YarnClientSchedulerBackend类是CoarseGrainedSchedulerBackend类的子类，YarnClientClusterScheduler是TaskSchedulerImpl的子类，仅仅重写了TaskSchedulerImpl中的getRackForHost方法。
 　　4、初始化完taskScheduler后，将创建dagScheduler，然后通过taskScheduler.start()启动taskScheduler，而在taskScheduler启动的过程中也会调用SchedulerBackend的start方法。在SchedulerBackend启动的过程中将会初始化一些参数，封装在ClientArguments中，并将封装好的ClientArguments传进Client类中，并client.submitApplication()方法获取Application ID。

 　　
      Yarn操作：
     1、运行ApplicationMaster的run方法（runExecutorLauncher）；
     2、无需等待SparkContext初始化完成（因为YarnClientClusterScheduler已启动完成），向sparkYarnAM注册该Application
 　  3、分配Executors，这里面的分配逻辑和yarn-cluster里面类似，就不再说了。
 　  4、最后，Task将在CoarseGrainedExecutorBackend里面运行，然后运行状况会通过Akka通知CoarseGrainedScheduler，直到作业运行完成。
 　  5、在作业运行的时候，YarnClientSchedulerBackend会每隔1秒通过client获取到作业的运行状况，并打印出相应的运行信息，当Application的状态是FINISHED、FAILED和KILLED中的一种，那么程序将退出等待。
 　  6、最后有个线程会再次确认Application的状态，当Application的状态是FINISHED、FAILED和KILLED中的一种，程序就运行完成，并停止SparkContext。整个过程就结束了。
 
 

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/29754888/viewspace-1815323/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： scala class和object的区别
下一篇： Option[T] 作用
请登录后发表评论 登录
全部评论
破棉袄

注册时间：2014-07-16

  • 博文量
    180
  • 访问量
    1159109

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us 编辑准则

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593