ideaspark源码

‘壹’ Spark源码分析之SparkSubmit的流程

本文主要对SparkSubmit的任务提交流程源码进行分析。 Spark源码版本为2.3.1。

首先阅读一下启动脚本，看看首先加载的是哪个类，我们看一下 spark-submit 启动脚本中的具体内容。

可以看到这里加载的类是org.apache.spark.deploy.SparkSubmit，并且把启动相关的参数也带过去了。下面我们跟一下源码看看整个流程是如何运作的...

SparkSubmit的main方法如下

这里我们由于我们是提交作业，所有会走上面的submit(appArgs, uninitLog)方法

可以看到submit方法首先会准备任务提交的环境，调用了prepareSubmitEnvironment，该方法会返回四元组，该方法中会调用doPrepareSubmitEnvironment，这里我们重点注意 childMainClass类具体是什么 ，因为这里涉及到后面启动我们主类的过程。

以下是doPrepareSubmitEnvironment方法的源码...

可以看到该方法首先是解析相关的参数，如jar包，mainClass的全限定名，系统配置，校验一些参数，等等，之后的关键点就是根据我们 deploy-mode 参数来判断是如何运行我们的mainClass，这里主要是通过childMainClass这个参数来决定下一步首先启动哪个类。

childMainClass根据部署模型有不同的值:

之后该方法会把准备好的四元组返回，我们接着看之前的submit方法

可以看到这里最终会调用doRunMain()方法去进行下一步。

doRunMain的实现如下...

doRunMain方法中会判断是否需要一个代理用户，然后无论需不需要都会执行runMain方法，我们接下来看看runMain方法是如何实现的。

这里我们只假设以集群模式启动，首先会加载类，将我们的childMainClass加载为字节码对象mainClass ，然后将mainClass 映射成SparkApplication对象，因为我们以集群模式启动，那么上一步返回四元组中的childMainClass的参数为ClientApp的全限定名，而这里会调用app实例的start方法因此，这里最终调用的是ClientApp的start方法。

ClientApp的start方法如下...

可以看到这里和之前我们的master启动流程有些相似。
可以参考我上一篇文章 Spark源码分析之Master的启动流程 对这一流程加深理解。

首先是准备rpcEnv环境，之后通过master的地址获取masterEndpoints端点相关信息，因为这里运行start方法时会将之前配置的相关参数都传进来，之后就会通过rpcEnv注册相关clientEndPoint端点信息，同时需要注意，这里会把masterEndpoints端点信息也作为构造ClientEndpoint端点的参数，也就是说这个ClientEndpoint会和masterEndpoints通信。

而在我上一篇文章中说过，只要是setupEndpoint方法被调用，一定会调用相关端点的的onStart方法，而这会调用clientEndPoint的onStart方法。

ClientEndPoint类中的onStart方法会匹配launch事件。源码如下

onStart中匹配我们的launch的过程，这个过程是启动driverWrapper的过程，可以看到上面源码中封装了mainClass ，该参数对应DriverWrapper类的全限定名，之后将mainClass封装到command中，然后封装到driverDescription中，向Master申请启动Driver。

这个过程会向Mster发送消息，是通过rpcEnv来实现发射消息的，而这里就涉及到outbox信箱，会调用postToOutbox方法，向outbox信箱中添加消息，然后通过TransportClient的send或sendRpc方法发送消息。发件箱以及发送过程是在同一个线程中进行。

而细心的同学会注意到这里调用的方法名为SendToMasterAndForwardReply，见名之意，发送消息到master并且期待回应。

下面是rpcEnv来实现向远端发送消息的一个调用流程，最终会通过netty中的TransportClient来写出。

之后，Master端会触发receiveAndReply函数,匹配RequestSubmitDriver样例类，完成模式匹配执行后续流程。

可以看到这里首先将Driver信息封装成DriverInfo，然后添加待调度列表waitingDrivers中，然后调用通用的schele函数。

由于waitingDrivers不为空，则会走LaunchDriver的流程，当前的application申请资源，这时会向worker发送消息，触发Worker的receive方法。

Worker的receive方法中，当Worker遇到LaunchDriver指令时，创建并启动一个DriverRunner，DriverRunner启动一个线程，异步的处理Driver启动工作。这里说启动的Driver就是刚才说的org.apache.spark.deploy.worker.DriverWrapper

可以看到上面在DriverRunner中是开辟线程异步的处理Driver启动工作，不会阻塞主进程的执行，而prepareAndRunDriver方法中最终调用 runDriver..

runDriver中主要先做了一些初始化工作，接着就开始启动driver了。

上述Driver启动工作主要分为以下几步：

下面我们直接看DriverWrapper的实现

DriverWrapper，会创建了一个RpcEndpoint与RpcEnv，RpcEndpoint为WorkerWatcher，主要目的为监控Worker节点是否正常，如果出现异常就直接退出，然后当前的ClassLoader加载userJar，同时执行userMainClass，在执行用户的main方法后关闭workerWatcher。

以上就是SparkSubmit的流程，下一篇我会对SparkContext的源码进行解析。

欢迎关注...