zoukankan      html  css  js  c++  java
  • 【Spark】部署流程的深度了解


    Spark核心组件

    关于Spark的基本架构模块,在【Spark】一起了解一下大数据必不可少的Spark吧! 这篇文章已经有所提及,这里主要要先明白两个主要运行角色DriverExecutor在整个部署流程的主要职责
     

    Driver

    是Spark驱动器节点,主要用于执行Spark任务中的main方法,负责实际代码的执行。
    详细职责:
    1.将用户程序转化为作业(job)
    2.在Executor之间调度任务(task)
    3.跟踪Executor的执行情况
    4.通过UI展示查询运行情况
     

    Executor

    其实Executor并不是一个进程,而是ExecutorBackend是一个进程,Executor是它的执行类,负责Spark作业中运行具体的任务,任务之间是彼此相互独立的
    Spark应用启动时,Executor节点同时启动,并且始终伴随整个Spark应用的生命周期而存在。如果Executor节点发生故障或者崩溃,Spark会将故障节点的任务调度到其他Executor节点上执行。
    详细职责:
    1.运行组成Spark应用的任务,并将结果返回给驱动器进程
    2.通过自身的块管理器(Block Manager) 为用户程序中要求缓存的RDD提供内存式存储。RDD 是直接缓存在Executor进程内的,因此任务可以在运行时充分利用缓存数据加速运算。


    Spark通用运行流程图

    在这里插入图片描述


    Standalone模式运行机制

    Client模式流程图

    在这里插入图片描述

    Cluster模式流程图

    在这里插入图片描述


    On-Yarn模式运行机制

    Client模式流程图

    在这里插入图片描述
    注:
    1.ApplicationMaster是Yarn集群中的核心概念,任何要在Yarn集群上启动的Job类型(包括MR和Spark)都必须有一个ApplicationMaster。
    2.每种计算框架(比如MR和Spark)如果想要在Yarn上执行自己的Application,就必须自己实现和提供一个ApplicationMaster,相当于自己实现了Yarn提供的接口,Spark自己开发的一个类。
    3.Spark-On-Yarn在Client模式下,Application向ResourceManager申请资源和task的调度是分离的。

    Cluster模式流程图

    在这里插入图片描述


    源码解读(多图版)

    Spark On-Yarn Cluster模式示例

    1.SparkSubmit

    通过spark-submit提交任务,会在后台启动一个进程
    在这里插入图片描述
    然后就会调用spark-submit中的main方法
    查看源码,在SparkSubmit的伴生对象中,我们可以看到其中要调用的main方法
    在这里插入图片描述
    在这里插入图片描述

    代码如下

     override def main(args: Array[String]): Unit = {
       val appArgs = new SparkSubmitArguments(args)
       if (appArgs.verbose) {
         // scalastyle:off println
         printStream.println(appArgs)
         // scalastyle:on println
       }
       appArgs.action match {
         case SparkSubmitAction.SUBMIT => submit(appArgs)
         case SparkSubmitAction.KILL => kill(appArgs)
         case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)
       }
     }
    

    从以上代码,可以看到val appArgs = new SparkSubmitArguments(args),这表示它将参数都封装到一起,而args指的就是提交任务时输入的--master--deploy-mode--executor-memory等这些参数。
    具体有哪些参数,我们也可以点进去查看,以下就是里面封装的参数
    在这里插入图片描述
    那么究竟是如何给这些参数,赋值的
    其实它是将这些参数解析转换成了java。
    在这里插入图片描述
    并且其中有一个handle方法,对我们输入的参数进行模式匹配。
    在这里插入图片描述
    再回到SparkSubmit中,我们可以看到一个关于action的模式匹配
    在这里插入图片描述
    action就和master一样,是我们上传jar包是需要配置的,如果不配置,它有方法默认为Submit
    在这里插入图片描述
    在这里插入图片描述
    那么就可以看一下submit方法里面具体是什么,首先就可以看到一个prepareSubmitEnvironment,也就是准备上传环境,又是采用了一个模式匹配,返回值就是括号里那四个childArgs、childClasspath、sysProps、childMainClass
    在这里插入图片描述
    接着往下看,可以看到一个判断是否为standalone的Cluster模式,如果不是就会执行doRunMain方法
    在这里插入图片描述
    doRunMain方法中主要是在判断代理用户参数是否为空,也就是有没有给代理用户的参数,不过无论是否给了这个参数,最终都会运行runMain方法,而runMain方法中需要的函数正是之前准备上传环境所返回的值。
    在这里插入图片描述
    点进去查看runMain方法,最主要的就是这个反射加载类
    在这里插入图片描述
    同样,还有这个查找提交的jar包中是否有main方法,并作了一个是否有static修饰符的判断,其实就是在查找main方法invoke就是在调用main方法
    在这里插入图片描述
    现在了解到这,就需要知道反射加载类中的childMainClass是什么,它的首次出现,是作为prepareSubmitEnvironment返回的元祖中的最后一个元素。
    在这里插入图片描述
    那么就要弄明白,childMainClass是如何进行赋值的。
    在prepareSubmitEnvironment方法中,是经过多个判断方法,判断当前是什么运行模式,来对childMainClass进行赋值,第一张图为yarn模式的Cluster模式,第二张是Client模式。Cluster给childMainClass赋值为 “org.apache.spark.deploy.yarn.Client”,Client则是args.mainClass,也就是启动时输入的--class的命令行参数(比如org.apache.spark.examples.SparkPi)。
    在这里插入图片描述
    在这里插入图片描述
    实际工作中,一般用的都是yarn-Cluster模式,所以现在要弄清楚 “org.apache.spark.deploy.yarn.Client” 是什么,在这之前,现在pom.xml中添加以下包

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-yarn_2.11</artifactId>
        <version>2.1.1</version>
    </dependency>
    

    查找类后,找到其中的main方法,那么我们发现,已经不再是SparkSubmit类了,已经到了Client类
    在这里插入图片描述

    2.Client

    Client类的第一步同样也是封装参数
    在这里插入图片描述
    在这里插入图片描述
    第二步就是new了一个Client对象。
    在这里插入图片描述
    我们需要看看这么做的原因,由下图就能知道,它首先就是想创建一个YarnClient
    在这里插入图片描述
    创建成功后,它立即进行了run方法,run方法点进去,第一件事就是submitApplication,提交应用,并且把最后返回的值赋值为appID,这就是在spark-shell界面登录时看到的appID。
    在这里插入图片描述
    submitApplication方法中,最重要的一步就是yarnClient.submitApplication(appContext),作用是向Yarn提交应用,它的参数是appContext,往上推,appContext的是createApplicationSubmissionContext方法返回的值,而此方法需要的参数containerContextcreateContainerLaunchContext方法返回的值,这两个方法的作用是设置适当的context启动ApplicationMaster,接下来的重点就是看这两个方法。
    在这里插入图片描述
    点进createContainerLaunchContext方法,可以看到的submitApplication提交的不是真实的应用,而是提交的command——指令。command中封装的指令,根据上面args的判断后赋值,可以确定有client和cluster两种模式,cluster模式的指令为 /bin/java org.apache.spark.deploy.yarn.ApplicationMaster,client模式的指令为org.apache.spark.deploy.yarn.ExecutorLauncher
    在这里插入图片描述

    3.ApplicationMaster

    现在我们就可以来看ApplicationMaster这个类,查看它的main方法。
    在这里插入图片描述
    在这之前,首先要搞明白,client模式下org.apache.spark.deploy.yarn.ExecutorLauncher和ApplicationMaster有什么区别,所以先看一下ExecutorLauncher类,发现它和ApplicationMaster其实没有区别,甚至直接调用了ApplicationMaster的Main方法。
    在这里插入图片描述
    好,现在回到ApplicationMaster的main方法,第一步还是对参数进行了封装第二步就是new了一个ApplicationMaster,并且可以看到里面有两个参数,第一个amArgs就是第一步封装的参数,第二个new YarnRMClient是创建了一个yarn的ResourceManager的客户端,这么做是因为ApplicationMaster会向ResourceManager申请资源进行交互。第三步就是让ApplicationMaster运行起来
    在这里插入图片描述
    先点进ApplicationMaster看一下,在这里,我们需要明白两个东西,一个是RPC,一个是Endpoint,其实很简单,RPC就是Remote Procedure Call——远程过程调用,主要是通过网络从远程计算机程序上请求服务,而Endpoint就是终端
    这里rpcEnv中的Env就是EnvironmentamEndpoint是ApplicationMaster的终端。
    在这里插入图片描述
    接下来,进入到run方法,里面最重要的就是判断是否是cluster模式,如果是cluster模式,执行runDriver方法,如果不是,执行runExecutorLauncher方法
    在这里插入图片描述
    那就该进入到runDriver方法,第一个就是startUserApplication方法,顾名思义,启动用户应用,点进去,可以看到,userClassLoader.loadClass(args.userClass).getMethod(“main”, classOf[Array[String]]),这一步主要就是要获取我们输入命令行参数时--class中类的main方法
    在这里插入图片描述
    在这里插入图片描述
    获取到用户类的main方法,下一步当然就是执行,它new了一个Thread,进行启动,我们可以看到它的名字就是Driver,所以Driver是一个线程,我们写的SparkContext、StreamingContext,就是Driver。
    在这里插入图片描述
    ApplicationMaster启动后,ResourceManager要和AM进行交互,所以要registerAM,它们两个都是进程,所以我们看到了参数中有RPC。
    点进去registerAM,可以看到有一个client的注册,这个client就是YarnRMClient,所以这个过程就是ApplicationMaster向ResourceManager申请资源的过程
    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述
    那么,究竟是怎么分配资源的呢?我们可以看allocateResources这个方法。
    点进去可以看到,主要就是分配RM中的container,allocatedContainers.size > 0就意味着RM中有资源可以分配,接下来就会进行资源的处理——handleAllocatedContainers
    在这里插入图片描述
    在这里插入图片描述
    点进去handleAllocatedContainers,首先进行一个节点分配的操作,然后开始运行——runAllocatedContainers
    点进去可以看到,最终的目的就是启动Executor——new ExecutorRunnable().run

    在这里插入图片描述
    在这里插入图片描述
    ApplicationMaster申请资源完毕后,通知NodeManager启动Container,而startContainer中又出现了prepareComand,command就是 /bin/java "org.apache.spark.executor.CoarseGrainedExecutorBackend"
    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述

    4.CoarseGrainedExecutorBackend

    打开CoarseGrainedExecutorBackend类,找到它的main方法,直接看其中的run方法,它又在run些什么呢?
    在这里插入图片描述
    点进去我们发现,Executor其实是ExecutorBackend的一个计算对象,真正运行的进程应该是ExecutorBackend
    在这里插入图片描述
    在这里插入图片描述
    这里继承了一个ThreadSafeRpcEndpoint,它又继承自RpcEndpoint,RpcEndpoint类中讲到,一个Endpoint的生命周期就是constructor -> onStart -> receive* -> onStop
    在这里插入图片描述

    在这里插入图片描述
    ExecutorBackend继承自它,那说明ExecutorBackend也是一个进程,它也有同样的生命周期。
    回到CoarseGrainedExecutorBackend类中,我们看到它确实有,在onStart阶段,它主要干了一件事,ref.ask[Boolean](RegisterExecutor——这就是前面一直说的Executor启动后会向Driver进行反向注册。在receive阶段,它干的就是上面说的new Executor,这么做就是创建了一个计算对象,来处理拿到的数据。
    在这里插入图片描述


    源码解读(无图版)

    1. SparkSubmit
        
        // 启动进程
        -- main
        
            // 封装参数
            -- new SparkSubmitArguments
            
            // 提交
            -- submit
            
                // 准备提交环境
                -- prepareSubmitEnvironment
                
                    // Cluster
                    -- childMainClass = "org.apache.spark.deploy.yarn.Client"
                    // Client
                    -- childMainClass = args.mainClass (SparkPi)
                
                -- doRunMain (runMain)
                
                    // 反射加载类
                    -- Utils.classForName(childMainClass)
                    // 查找main方法
                    -- mainClass.getMethod("main", new Array[String](0).getClass)
                    // 调用main方法
                    -- mainMethod.invoke
    
    2. Client
    
        -- main
        
            -- new ClientArguments(argStrings)
            
            -- new Client
            
                -- yarnClient = YarnClient.createYarnClient
            
            -- client.run
                    
                    -- submitApplication
                    
                        // 封装指令 command = bin/java org.apache.spark.deploy.yarn.ApplicationMaster (Cluster)
                        // command = bin/java org.apache.spark.deploy.yarn.ExecutorLauncher  (client)
                        -- createContainerLaunchContext
    
                        
                        -- createApplicationSubmissionContext
                    
                        // 向Yarn提交应用,提交指令
                        -- yarnClient.submitApplication(appContext)     
    
    3. ApplicationMaster
        
        // 启动进程
        -- main
        
            -- new ApplicationMasterArguments(args)
            
            // 创建应用管理器对象
            -- new ApplicationMaster(amArgs, new YarnRMClient)
            
            // 运行
            -- master.run
            
                // Cluster
                -- runDriver
                
                    // 启动用户应用
                    -- startUserApplication
                    
                        // 获取用户应用的类的main方法
                        -- userClassLoader.loadClass(args.userClass)
          .getMethod("main", classOf[Array[String]])
          
                        // 启动Driver线程,执行用户类的main方法,
                        -- new Thread().start()
                        
                    // 注册AM
                    -- registerAM
                    
                        // 获取yarn资源
                        -- client.register
                        
                        // 分配资源
                        -- allocator.allocateResources()
                        	// 处理资源
                            -- handleAllocatedContainers
                            	// 启动container
                                -- runAllocatedContainers
                                
                                    -- new ExecutorRunnable().run
                                    
                                        -- startContainer
                                        
                                            // command = bin/java org.apache.spark.executor.CoarseGrainedExecutorBackend
                                            -- prepareCommand      
    
    4. CoarseGrainedExecutorBackend
        // 启动进程
        -- main
        
            -- run
            
                -- onStart
                	// 反向注册
                    -- ref.ask[Boolean](RegisterExecutor
                
                -- receive
                
                    --  case RegisteredExecutor
                    	// 创建计算对象
                        -- new Executor          
    
  • 相关阅读:
    Spring整合JMS(四)——事务管理
    centos7.3部署django用uwsgi和nginx[亲测可用]
    ImportError: Couldn't import Django.或者提示Django 模块不存在
    ModuleNotFoundError: No module named '_sqlite3'
    django-admin:command not found的解决办法
    2个版本并存的python使用新的版本安装django的方法
    uwsgi的使用
    plugins/python/uwsgi_python.h:2:20: fatal error: Python.h: No such file or directory
    cento7.3下玩转sphinx
    一些常用的centos命令,记忆下,属于常用的
  • 原文地址:https://www.cnblogs.com/zzzsw0412/p/12772380.html
Copyright © 2011-2022 走看看