Spark源码（7）SparkContext初始化源码分析

一、SparkContext 初始化

上次阅读到 Master 通知 Worker 启动了一个 Driver&＃xff0c;就是启动了一个 JVM&＃xff0c;并且开始使用反射的方式执行 DriverWrapper 的 main 方法。

假设我们提交了一个 JavaWordCount 程序到集群上&＃xff0c;其实接下来就开始执行这个方法&＃xff1a;

然后开始创建 SparkContext

SparkSession spark &＃61; SparkSession.builder().appName("JavaWordCount").getOrCreate();


点进去&＃xff0c;可以看到&＃xff1a;

SparkContext.getOrCreate(sparkConf)


这里 new 了一个 SparkContext()&＃xff0c;然后开始看 SparkContext 的构造方法&＃xff0c;这个方法尤其的长&＃xff0c;但我们得抓住重点。

重点就在 500 多行的这里&＃xff0c;创建了 SchedulerBackend&＃xff0c;TaskScheduler&＃xff0c;DAGScheduler

下面简单的介绍下这三个核心组件的作用&＃xff1a;

&＃xff08;1&＃xff09;TaskScheduler

TaskScheduler 是一个 trait&＃xff0c;它的核心任务是提交 TaskSet 到集群并汇报结果&＃xff0c;并且向 DAGScheduler 汇报任务的执行情况。

主要实现类是 TaskSchedulerImpl

&＃xff08;2&＃xff09;SchedulerBackend

SchedulerBackend 也是一个 trait&＃xff0c;不同的部署模式有不同的实现&＃xff0c;比如在 standalone 模式下&＃xff0c;实现类是&＃xff1a;StandaloneSchedulerBackend。

它内部有两个 Endpoint 组件&＃xff0c;一个是 DriverEndpoint &＃xff0c;用来和 Worker 打交道&＃xff0c;比如创建 Executor、提交 Task 等&＃xff1b;一个是 ClientEndpoint 用来和 master 打交道。

SchedulerBackend 专门负责收集 Worker 上的资源信息&＃xff0c;它知道每个任务在提交时&＃xff0c;自己拥有多少资源&＃xff0c;然后去具体运行 Task。

&＃xff08;3&＃xff09;DAGScheduler

DAGScheduler 的主要作用是&＃xff1a;当执行 action 算子的时候&＃xff0c;DAGScheduler 会从最后一个 rdd 开始往前递归寻找 ShuffleDependency&＃xff0c;每找到一个就会划分一个 stage&＃xff0c;最终达到的目的是&＃xff0c;根据业务逻辑算子依赖关系&＃xff0c;划分成一个个 stage&＃xff0c;并且根据 stage 先后关系&＃xff0c;把 stage 转化成 ShuffleMapTask 或者 ResultTask&＃xff0c;封装成 TaskSet&＃xff0c;交给 TaskScheduler 提交。

最后要注意的是&＃xff0c;DAGScheduler 和 SchedulerBackend 都是 TaskSchedulerImpl 的成员变量。

二、TaskScheduler 的创建

下面回到创建 TaskScheduler 的地方&＃xff1a;

val (sched, ts) &＃61; SparkContext.createTaskScheduler(this, master, deployMode)


这个方法中&＃xff0c;主要是根据不同的部署模式&＃xff0c;分别创建不同的 TaskScheduler 和 SchedulerBackend。

在 Standalone 模式下&＃xff1a;

创建了 TaskSchedulerImpl 对象和 StandaloneSchedulerBackend 对象。

在 TaskSchedulerImpl 的构造方法中&＃xff0c;看到了两个重要的属性&＃xff0c;dagScheduler 和 backend&＃xff0c;其他倒没有什么逻辑了。

然后看到 StandaloneSchedulerBackend 的构造方法

这里要注意一下继承关系&＃xff0c;继承了 CoarseGrainedSchedulerBackend

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GwvcTFpp-1637895026068)(C:/Users/wangkai/AppData/Roaming/Typora/typora-user-images/image-20210927222927252.png)]

然后父类里有一个很重要的属性&＃xff0c;在执行构造的方法的时候&＃xff0c;就会执行这个

也就是说&＃xff0c;在这句代码执行的时候&＃xff1a;

val backend &＃61; new StandaloneSchedulerBackend(scheduler, sc, masterUrls)


其实&＃xff0c;就创建了一个 DriverEndpoint 组件了&＃xff0c;那么紧接着就需要看一下 DriverEndpoint 类的 onStart() 方法&＃xff1a;

每间隔一段时间&＃xff0c;就给自己发送了一个 ReviveOffers 消息

这个消息的处理逻辑就很长了&＃xff0c;需要单独一次来阅读了。可以一句话总结它干的事情&＃xff1a;过滤出所有的资源&＃xff0c;找到所有需要启动的 TaskSet&＃xff0c;分配资源&＃xff0c;启动任务。

然后再回来看 StandaloneSchedulerBackend 的构造方法&＃xff0c;构造方法里没有逻辑&＃xff0c;接着回到 SparkContext 往下看。

接下来&＃xff0c;scheduler 调用了 initialize 方法&＃xff1a;

scheduler.initialize(backend)


这个方法就把当前的 SchedulerBackend 绑定到自己的成员属性上&＃xff1a;

接着 new 了 DAGScheduler

_dagScheduler &＃61; new DAGScheduler(this)


然后启动 TaskScheduler()

_taskScheduler.start()


三、TaskScheduler 的启动

首先看到实现类 TaskSchedulerImpl 类的 start() 方法

调动了内部的 SchedulerBackend 的启动方法&＃xff1a;

这个方法中&＃xff0c;首先封装了一个 CoarseGrainedExecutorBackend 类的启动命令

把这个 Command 封装到 ApplicationDescription&＃xff0c;应用程序的描述信息中

然后把 ApplicationDescription 作为成员变量&＃xff0c;并且还创建了一个 StandaloneAppClient 对象

然后给启动了起来&＃xff0c;new 了一个 ClientEndpoint() &＃xff0c;注意这是一个 Endpoint

看一下它的 onStart() 方法&＃xff0c;向 Master 注册

最终是给 Master 发送了一个 RegisterApplication 消息&＃xff0c;消息里有 ApplicationDescription&＃xff08;应用程序的描述信息&＃xff09;

四、总结

这一切都做完之后&＃xff0c;SparkContext() 最核心的地方就阅读完了&＃xff0c;总结一下就是创建了 DAGScheduler、TaskScheduler、SchedulerBackend 三大组件。并且 SchedulerBackend 内部有两个 Endpoint&＃xff0c;DriverEndpoint 和 ClientEndpoint&＃xff0c;分别从来和 Worker 通信 、和 Master 通信。

下次我们来阅读 Master 处理 Driver 的注册应用程序的消息。

下面是目前的流程图&＃xff1a;