Spark作业资源调度源码解析.docx
Spark作业资源调度源码解析Spark作业调度分为资源调度和任务调度两个部分，作业提交之后首先由资源调度系统为作业分配所需的计算资源，并创建driver和所有的executor实例；然后由任务调度系统通过解析作业的计算逻辑进一步划分stage、创建任务集并根据一定的调度规则将任务集分发到合适的executor执行器上去运行。本文主要介绍Spark资源调度的相关原理。Spark资源调度方式Spark的资源调度主要有如下几种方式：1. SparkStanda1one：Standa1one是Spark的原生调度方式，适用于小型或者测试集群。当我们提交SPark作业时使用-masterspark：/IP：PORT参数时，SPark将使用StandaIone模式对作业进行资源调度;采用本模式调度需要用户自行搭建SPark集群，参与资源调度的角色主要为Master和Worker,Master用于作业调度和计算资源的分配，Worker则用于启动driver和executor实例。2. SparkonYarn：当提交作业时使用一masteryarn参数时，SPark作业将会被调度至Yarn集群上去运行，基于Yaci的调度可适用于大型高并发的企业集群；Yarn作为HadooP项目一部分，是业内应用非常广泛的资源调度器，可以支持包括SPark、MaPRedUCe和f1ink等多种计算引擎的调度；可以采用FIF0、Fair、CaPaCity多种调度模式，并进一步定制资源分配和调度策略，为用户提供了灵活的调度框架；同时还支持多租户资源隔离和资源争抢，可以达到资源隔离和集群资源使用率的平衡。Yarn通过AppIicationMaster来负责作业的资源申请以及任务的调度，而ResourceManager则负责集群资源的统一管理和分配。3. SparkonKubernetes:提交作业时使用一masterk8s:/apiserver：port参数则可以将应用调度到K8S集群运行；Spark从2.4开始支持使用kubernetes进行资源调度，Sparkonkubernetes可以方便地实现存算分离和计算资源的弹性伸缩，极大方便了业务的快速拓展，同时还可以提高硬件资源的利用率并节约成本;Kubernetes的调度器kube-schedu1er根据指定的调度算法在合适的node节点上启动driverpod和executorpodo由于Standa1one的是Spark的原生调度方式，并且实现方式较为简单，而Yarn和Kubernetes是独立于spark项目之外的资源调度系统，因此本文将以Standa1one模式为主线来解析Spark的资源调度原理，并以少量的Yarn和Kubernetes调度源码辅助说明。作业资源调度流程Spark作业通过SPark-SUbmit或者rest方式提交作业，然后由资源调度系统对作业进行调度，主要的流程如下：1 .SparkCIient提交作业，并向调度系统申请启动driver实例；2 .调度系统根据调度策略和driver的资源需求等信息选择在特定的节点上启动driver；3 .DriVer负责invoke用户的主类mainc1ass,在mainc1ass中初始化了SParkCOnteXt对象，SParkCOnteXt不仅是应用编程接口，同时也包含了整个作业运行时的诸多重要对象，包括作业运行时环境对象SParkEnV、作业调度器DAGSchedu1er任务调度器TaskSchedu1er以及负责与executor通信的RPC组件Schedu1erBackend等；4 .C1ient端向调度系统注册该作业;5 .调度系统根据已注册作业的资源需求信息，以及集群当前的资源使用情况将各个作业的executors调度到计算节点上去，executor启动后会初始化一个线程池等待执行driver分发的任务；6 .已经启动的executor向driver注册。Driver的创建和启动Stand1one模式下启动Driver：在StandaIone模式下，通过CIientAPP中CIientEndPOint对象向MaSter发起RequestSubniitDriver请求，Master接受到RequestSubmitDriver请求之后创建DriverInfo对象并将该对象放到一个HashSet对象中，然后开始调度driver的启动：IVaIdriver二CreaXeDriver(description)1J：>.rMaster遍历WaitingDrivers中等待调度的driver,针对每个driver随机从集群中获取一个状态正常的WOrker,如果WOrker的资源足够启动driver,则将driver调度到这个worker节点上，并调整WaitingDrivers:or(driver<-WaitingDrivers.to1ist),hi1e(numTVorkersVisited<numWorkersA1ive&&!1aunched)Iv/3,kc,hu:；、(！)rkrs(<p)IisC1uster1d1e二worker,drivers.isEmp1y&&worker,executors,isIif(Can1aUnChDriVer(Worker,CIriVer.desc)/检查WOrker叭IVa1a11ocated二WOrker.acquireResources(driver,desc.resourIi.-w-J：I"I1Z'I：,”；,',：I"1iI0)ri<'s-二di(.r.crPos=(crPos+1)%IiumWorkersA1iveMaster通过向worker发送1aunchDriver指令来启动driver,在standa1one模式下是在worker中启动了一个driver线程，然后对该worker的资源使用量进行调整：Va1driver=newDriverRunner(I.I.IdriverDesc.copy(command二WorkenmaybeUpdateSS1Sett：.！i,Ei.二一一:，I,.I111I11111I11j1111I11II11I1I11iIiI11mIII1111.11IaddResourcesUsed(resourcesJDriver实例启动之后会初始化SparkContext对象，在SparkContext中初始化了很多作业运行时的对象，主要包括作业运行时SparkEnv相关运行时对象:a1env1nstance二newSParkEnV(I,，：.(:IsSure'Seria1iz(?r,mSOriaIizor1anagor,map()utPUtTraCker,Khu:('V'm*"t'',.broadcastVanagor,bicd/('(.i11<：y>U'.('r,.¾I1-!.-ci()1、n，g。1、,OutputCommitCoordinator,以及作业资源调度和任务调度的相关对象:'/Createandstarttheschedu1e.Va1(SChed,ts)=SParkConteXt.CreateTaskSchedu1er(this,master,depIoyMode)：i一八'!,一.i，d<S(hrd：U'C：(.hear1beatReceiver.askBoo1ean(TaSkSehed1dCr1SSCI)OnYarn模式的实现：Spark通过Yarn调度的情况下，driver是运行在AppIicationMastercontainer中，通过YarnciusterApp1ication作为客户端向ResourceManager申请加载资源并启动driver,如下是runDriver()方法的相关代码片段：va1host=userConf.get(DRIVEI1HOST.ADDRESS)VaIPOrt二USerConf.get(DR1VER_PORT)registerAM(host,port,userConf,sc.ui.map(_.webUr1),appAttemptId)/向resourCUmanager注册app1icationnasfeVaIdriverRef二rpcEnv.SetupEndpointRef(RPCAddreSS(host,port).YarnSchedu1erBackend.ENDPOINT_NAME)CreateA11ocator(driverRef,userConf,rpcEnv,appttemptTd,distCacheConf)为driver加载启动drive.OnKubernetes模式的实现：在KUberneteS中,作业提交之后由SParkAPP1iCatiOn的子类KubernetesciientApp1ication作为客户端向kubernetes集群申请创建driverpod,并通过监听器WatCher来监听POd的状态变化:
