首页 > 网络 > 云计算 >

Spark定制版:007~SparkStreaming源码解读之JobScheduler内幕实现和深度思考

2016-05-25

本讲内容:a JobScheduler内幕实现;b JobScheduler深度思考

本讲内容:

a. JobScheduler内幕实现
b. JobScheduler深度思考

注:本讲内容基于Spark 1.6.1版本(在2016年5月来说是Spark最新版本)讲解。

上节回顾

上节课,我们以JobGenerator类为重心,为大家左右延伸,解密Job之动态生成;并总结出了Job之动态生成的三大核心:

a. JobGenerator: 负责Job生成

b. JobSheduler:负责Job调度

c. ReceiverTracker:获取元数据

如Job动态生成图:

\

开讲

由上节课我们可知:

JobScheduler是SparkStreaming 所有Job调度的中心,内部有两个重要的成员:JobGenerator负责Job的生成,ReceiverTracker负责记录输入的数据源信息。

JobScheduler的启动会导致ReceiverTracker和JobGenerator的启动。ReceiverTracker的启动导致运行在Executor端的Receiver启动并且接收数据,ReceiverTracker会记录Receiver接收到的数据meta信息。

JobGenerator的启动导致每隔BatchDuration,就调用DStreamGraph生成RDD Graph,并生成Job。

JobScheduler中的线程池来提交封装的JobSet对象(时间值,Job,数据源的meta)。Job中封装了业务逻辑,导致最后一个RDD的action被触发,被DAGScheduler真正调度在Spark集群上执行该Job。

因此可以说JobScheduler是整个Spark Streming的调度的核心,其地位相当于Spark Core中的DAGScheduler。因此,我们务必彻底掌握JobScheduler。

首先让我们来看看JobScheduler重要方法的源码跟踪步骤图:

\

下面我们来逐步解密JobScheduler的内幕吧。

上几节课中,我们在进行Spark Streaming开发的时候,会对DStream进行各种transform和action级别的操作,这些操作就构成DStream graph,也就是DStream 之间的依赖关系,随着时间的流逝,DStream graph会根据batchintaval时间间隔,产生RDD的DAG,然后进行job的执行。DStream的DStream graph是逻辑级别的,RDD的DAG是物理执行级别的。DStream是空间维度的层面,空间维度加上时间构成时空维度。

至此,JobScheduler是将逻辑级别的job物理的运行在Spark Core上。JobGenerator是产生逻辑级别的Job,使用JobScheduler将Job在线程池中运行。JobScheduler是在StreamingContext中进行实例化的,并在StreamingContext的start方法中开辟一条新的线程启动的。

\

大括号中的代码作为一个匿名函数在新的线程中执行。Sparkstreaming运行时至少需要两条线程,其中一条用于一直循环接收数据,现在所说的至少两条线程和上边开辟一条新线程运行scheduler.start()并没有关系。Sparkstreaming运行时至少需要两条线程是用于作业处理的,上边的代码开辟新的线程是在调度层面的中,不论Sparkstreaming程序运行时指定多少线程,这里都会开辟一条新线程,之间没有一点关系。

每一条线程都有自己私有的属性,在这里给新的线程设置私有的属性,这些属性不会影响主线程中的。

源码中代码的书写模式非常值得学习,以后看源码的时候就把它当做是一个普通的应用程序,从JVM的角度看,Spark就是一个分布式的应用程序。在以后的项目中要将调度和作业处理线程分开,也便于维护和优化。

JobScheduler在实例化的时候会实例化JobGenerator和线程池。

\

线程池中默认是有一条线程,当然可以在spark配置文件中配置或者使用代码在sparkconf中修改默认的线程数,在一定程度上增加默认线程数可以提高执行Job的效率,这也是一个性能调优的方法(尤其是在一个程序中有多个Job时)。

ReceiverTracker、JobGenerator在JobScheduler实例化的时候实例化。

来看一下JobScheduler.start的代码:

\

Eventloop是在调用JobGenerator的start方法时实例化。

\

在EventLoop的start方法中会回调onStart方法,一般在onStart方法中会执行一些准备性的代码,在JobSchedule中虽然并没有复写onStart方法,不过Spark Streaming框架在这里显然是为了代码的可扩展性考虑的,这是开发项目时需要学习的。

\

DStream的action级别的操作转过来还是会调用foreachRDD这个方法,生动的说明在对DStream操作的时候其实还是对RDD的操作。

\

上边代码中foreachFunc这个方法是对DStream action级别的方法的进一步封装;foreachRDD方法,转过来new ForEachDstream

\

注释中说的:将这个函数作用于这个DStream中的每一个RDD,这是一个输出操作,因此这个DStream会被注册成Outputstream,并进行物化。

ForEachDStream中很重要的一个函数generateJob。考虑时间维度和action级别,每个Duration都基于generateJob来生成作业。foreachFunc(rdd, time)这个方法就是对Dstream最后的操作 ,new Job(time, jobFunc)只是在RDD的基础上,加上时间维度的封装而已。这里的Job只是一个普通的对象,代表了一个spark的计算,调用Job的run方法时,真正的作业就触发了。foreachFunc(rdd, time)中的rdd其实就是通过DStreamGraph中最后一个DStream来决定的。

来看一下ForEachDStream.generateJob的代码:

\

Job是通过ForEachDstream的generateJob来生成的,值得注意的是在DStream的子类中,只有ForEachDstream重写了generateJob方法。

现在考虑一下ForEachDStream的generateJob方法是谁调用的?当然是JobGenerator。ForEachDstream的generateJob方法是静态的逻辑级别,他如果想要真正运行起来变成物理级别的这时候就需要JobGenerator。

现在就来看看JobGenerator的代码,JobGenerator中有一个定时器timer和消息循环体eventLoop,timer会基于batchInterval,一直向eventLoop中发送generateJobs的消息,进而导致processEvent方法->generateJobs方法的执行。

\

generateJobs方法的代码:

\

graph.generateJobs(time)这个方法的代码:

\

其中的outputStream.generateJob(time)中的outputStream就是前面说ForEachDstream,generateJob(time)方法就是ForEachDstream中的generateJob(time)方法。

这是从时间维度调用空间维度的东西,所以时空结合就转变成物理的执行了。

JobGenerator的generateJobs方法的代码:

\

基于graph.generateJobs产生job后,会封装成JobSet并提交给JobScheduler,JobSet(time, jobs, streamIdToInputInfos),其中streamIdToInputInfos就是接收的数据的元数据。

JobSet代表了一个batch duration中的一批jobs。就是一个普通对象,包含了未提交的jobs,提交的时间,执行开始和结束时间等信息。

JobSet提交给JobScheduler后,会放入jobSets数据结构中,jobSets.put(jobSet.time, jobSet) ,所以JobScheduler就拥有了每个batch中的jobSet.在线程池中进行执行。

\

在把job放入线程池中时,采用JobHandler进行封装。JobHandler是一个Runable接口的实例。

其中主要的代码就是job.run(),前面说过job.run()调用的就是Dstream的action级别的方法。

在job.run()前后会发送JobStarted和JobCompleted的消息,JobScheduler接收到这两个消息只是记录一下时间,通知一下job要开始执行或者执行完成,并没有过多的操作。

\

相关文章
最新文章
热点推荐