Java教程

spark源码之环境准备

本文主要是介绍spark源码之环境准备,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

我们使用yarn集群作为研究

spark环境准备

      • yarn环境准备
      • 启动Driver
      • 启动Executor

yarn环境准备

spark的入口类是SparkSubmit,在这里,我们开始提交参数

在这里插入图片描述

这里的args就是--class这些的。

在这里插入图片描述
解析好这些参数后,我们会返回一个SparkSubmitArguments的一个属性action的值并进行模式匹配。

在这里插入图片描述
我们可以看到,action默认就是SUBMIT

于是走submit分支:

在这里插入图片描述
在这里插入图片描述
假设我们没有使用--proxy-user参数,于是走runMain(args, uninitLog)

在这里插入图片描述
走进这个方法,首先他要准备提交的环境,并且我们关注返回元组中的childMainClass

进入prepareSubmitEnvironment(args)

在这里插入图片描述
因为我们是yarn集群模式,所以childMainClass = YARN_CLUSTER_SUBMIT_CLASS

YARN_CLUSTER_SUBMIT_CLASS是一个全类名:

在这里插入图片描述
在这里插入图片描述
拿到这个全类名后我们问他是不是SparkApplication的子类,如果是就用反射new出来,然后再转成SparkApplication

在这里插入图片描述
所以YarnClusterApplication就创建出来了。

在这里插入图片描述

创建出来后我们调用它的start方法。

在这里插入图片描述

start方法中我们要new一个客户端,进去:

在这里插入图片描述
他会创建一个YarnClient。这时候我们就要开始和RM建立连接了。

再走run方法:

在这里插入图片描述

在这里插入图片描述

现在要开始提交应用了。

在这里插入图片描述

yarnClient完成初始化并启动。

在这里插入图片描述

我们想知道它提交的是些什么。

所以,进到创建容器环境的代码:

      val containerContext = createContainerLaunchContext(newAppResponse)

在这里插入图片描述
它会将一些指令封装好提交给RM,所以我们关注这个amClass是个什么。

在这里插入图片描述

因为我们使用的是集群模式,所以amClass就是

org.apache.spark.deploy.yarn.ApplicationMaster

所以yarn客户端将启动ApplicationMaster的指令给RM,RM就会让其中一个NM来启动,因为命令是bin/java ApplicationMaster,所以起的是一个进程。

要执行ApplicationMaster这个进程,就是要执行main方法,进到他的main方法去:

在这里插入图片描述
在这里插入图片描述
在main方法中它new了一个ApplicationMaster,进去:

在这里插入图片描述
ApplicationMaster里面又new了一个YarnRMClient,这个东西就是ApplicationMaster用来和RM通信的。

在这里插入图片描述

再回到ApplicationMaster的main方法,这个AM创建出来后,去看它的run方法。

在这里插入图片描述
在这个run方法中,它判断我们是不是集群模式(显然我们是),是的话就走runDriver()

启动Driver

进入runDriver()

在这里插入图片描述
先看

  val sc = ThreadUtils.awaitResult(sparkContextPromise.future,
        Duration(totalWaitTime, TimeUnit.MILLISECONDS))

他在等待一个SparkContext,没有等到就阻塞。

所以我们进startUserApplication()看看发生了什么。

在这里插入图片描述

这个--class就是我们命令行传的主类参数:

在这里插入图片描述
在这里插入图片描述
拿到我们的main方法之后就是用invoke调用。

当然这是跑在一个线程中的:

启动的这个线程,就是我们的Driver

在这里插入图片描述

调用我们的main后一定会有SparkContext,随便的一个spark程序,都会创建SparkContext

在这里插入图片描述
于是之前阻塞的代码就可以走通了:

在这里插入图片描述

启动Executor

我们接着往下走:

在这里插入图片描述
首先他要注册AM,注册的含义AM就是要向RM申请资源。

然后要创建分配器,相当于是要分配资源了。

在这里插入图片描述
这里的client就是之前的YarnRMClient,就是用来和RM通信的。

在这里插入图片描述
它创建了分配器之后就要开始分配资源了。

在这里插入图片描述
如果可使用的容器数量大于0(yarn中资源是以container的形式存在的),就要开始着手处理这些容器了。

在这里插入图片描述
然后我们要去跑分配好的容器。

跑的时候他有一个线程池,走到里面的run方法去。

在这里插入图片描述
run方法里面它初始化了NM,然后启动容器。

在这里插入图片描述

最终是由NM来启动容器,nmClient.startContainer(container.get, ctx)里面有一个上下文对象,这个ctx里面封装了一些指令。

在这里插入图片描述

我们进到prepareCommand()里面去查看是哪些指令。

在这里插入图片描述
显然,他又是用bin/java的形式启动了一个进程,这个进程就是YarnCoarseGrainedExecutorBackend

接下来我们需要知道这个YarnCoarseGrainedExecutorBackend是什么。

在这里插入图片描述
走进它的run方法。

在这里插入图片描述
他首先要创建一个RpcEndpointRef类型的driver变量,这个东西就是用来和真正的Driver做通信的。

在这里插入图片描述
然后将自己(YarnCoarseGrainedExecutorBackend)设置为一个终端。其实就是一个通信的终端,我们看他设置的时候做了什么。

在这里插入图片描述
一路跟下去。

在这里插入图片描述
他会创建一个DedicatedMessageLoop

在这里插入图片描述

在这个类中,他会创建一个Inbox对象,就是一个收件箱。

在这里插入图片描述

Inbox回给自己发一个OnStart的指令。

在这里插入图片描述

他自己匹配到OnStart之后,就会调用终端onStart()方法。我们跟进去:

在这里插入图片描述
终端会给Driver发一个RegisterExecutor对象,表示要注册Excutor。

那Driver那边谁接受呢?当然是Driver里面的SparkContext

在这里插入图片描述

SparkContext里面有一个SchedulerBackend,他就是用来接发消息的。

在这里插入图片描述

这个实现类相当于是Driver方面的终端。

在这里插入图片描述
当他匹配到RegisterExecutor对象时,

在这里插入图片描述
最后返回了true

在这里插入图片描述
此时Executor端的终端接收到成功的消息后,就给自己发一个消息,说去注册Executor吧。

在这里插入图片描述
然后它自己和自己匹配到RegisteredExecutor后,就完成Executor的注册。

我们最后回到ApplicationMaster

在这里插入图片描述

刚才是在createAllocator这里离开的。Executor创建完成后,调用resumeDriver()来恢复Driver的执行,就是我们写的那些RDD。

这篇关于spark源码之环境准备的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!