我们使用yarn集群作为研究
spark环境准备
yarn环境准备
spark的入口类是SparkSubmit，在这里，我们开始提交参数
这里的args就是--class这些的。
SparkSubmitArguments的一个属性action的值并进行模式匹配。
action默认就是SUBMIT。
于是走submit分支：
--proxy-user参数，于是走runMain(args, uninitLog)。
childMainClass。
进入prepareSubmitEnvironment(args)。
childMainClass = YARN_CLUSTER_SUBMIT_CLASS。
而YARN_CLUSTER_SUBMIT_CLASS是一个全类名：
拿到这个全类名后我们问他是不是
SparkApplication的子类，如果是就用反射new出来，然后再转成SparkApplication。
所以
YarnClusterapplication就创建出来了。
他会创建一个
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方法里面它初始化了NM，然后启动容器。
最终是由NM来启动容器，nmClient.startContainer(container.get, ctx)里面有一个上下文对象，这个ctx里面封装了一些指令。
我们进到prepareCommand()里面去查看是哪些指令。
显然，他又是用
bin/java的形式启动了一个进程，这个进程就是YarnCoarseGrainedExecutorBackend。
接下来我们需要知道这个YarnCoarseGrainedExecutorBackend是什么。
@H_404_430@
走进它的
run方法。
他首先要创建一个
RpcEndpointRef类型的driver变量，这个东西就是用来和真正的Driver做通信的。
然后将自己（
YarnCoarseGrainedExecutorBackend）设置为一个终端。其实就是一个通信的终端，我们看他设置的时候做了什么。
一路跟下去。
他会创建一个
DedicatedMessageLoop。
他自己匹配到OnStart之后，就会调用终端onStart()方法。我们跟进去：
终端会给Driver发一个
RegisterExecutor对象，表示要注册Excutor。
那Driver那边谁接受呢？当然是Driver里面的SparkContext。
SparkContext里面有一个SchedulerBackend，他就是用来接发消息的。
这个实现类相当于是Driver方面的终端。
当他匹配到
RegisterExecutor对象时，
最后返回了
true。
此时Executor端的终端接收到成功的消息后，就给自己发一个消息，说去注册Executor吧。
然后它自己和自己匹配到
RegisteredExecutor后，就完成Executor的注册。
我们最后回到ApplicationMaster。
刚才是在createallocator这里离开的。Executor创建完成后，调用resumeDriver()来恢复Driver的执行，就是我们写的那些RDD。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
