[spark streaming]JobScheduler内幕实现和深度思考

小鱼儿 2022-05-15 10:44 142阅读 0赞

**本期内容：**

**1，JobScheduler内幕实现**

**2，JobScheduler深度思考**

**摘要：JobScheduler是Spark Streaming整个调度的核心，其地位相当于Spark Core上的调度中心中的DAGScheduler！**

**         **

# **一、JobScheduler内幕实现** #

**首先看看源码中对jobScheduler的解释：**

/**
     * This class schedules jobs to be run on Spark. It uses the JobGenerator to generate
     * the jobs and runs them using a thread pool.
     */

我们在看看源码对JobGenerator的解释：

/**
     * This class generates jobs from DStreams as well as drives checkpointing and cleaning
     * up DStream metadata.
     */

**问：JobScheduler是在什么地方生成的？**

**答：JobScheduler是在StreamingContext实例化时产生的，从StreamingContext的源码第183行中可以看出：**

**      **private\[streaming\] val scheduler = new JobScheduler(this)

**问：Spark Streaming为啥要设置两条线程？   
答：setMaster指定的两条线程是指程序运行的时候至少需要两条线程。一条线程用于接收数据，需要不断的循环。另一条是处理线程，是我们自己指定的线程数用于作业处理。如StreamingContext的start()方法所示：**

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 </td> 
 <td> &nbsp; <code>def</code>&nbsp;<code>start()</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>synchronized { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>state&nbsp;</code><code>match</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>INITIALIZED&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>startSite.set(DStream.getCreationSite())</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>StreamingContext.ACTIVATION</code><code>_</code><code>LOCK.synchronized { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>StreamingContext.assertNoOtherContextIsActive()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>validate()</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// Start the streaming scheduler in a new thread, so that thread local properties</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// like call sites and job groups can be reset without affecting those of the</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// current thread.&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //Spark Streaming内部启动的线程，用于整个作业的调度</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ThreadUtils.runInNewThread(</code><code>"streaming-start"</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>sparkContext.setCallSite(startSite.get)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>sparkContext.clearJobGroup()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>sparkContext.setLocalProperty(SparkContext.SPARK</code><code>_</code><code>JOB</code><code>_</code><code>INTERRUPT</code><code>_</code><code>ON</code><code>_</code><code>CANCEL,&nbsp;</code><code>"false"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>scheduler.start()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>state&nbsp;</code><code>=</code>&nbsp;<code>StreamingContextState.ACTIVE</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>NonFatal(e)&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logError(</code><code>"Error starting the context, marking it as stopped"</code><code>, e)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>scheduler.stop(</code><code>false</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>state&nbsp;</code><code>=</code>&nbsp;<code>StreamingContextState.STOPPED</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code>&nbsp;<code>e</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>StreamingContext.setActiveContext(</code><code>this</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>shutdownHookRef&nbsp;</code><code>=</code>&nbsp;<code>ShutdownHookManager.addShutdownHook(</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>StreamingContext.SHUTDOWN</code><code>_</code><code>HOOK</code><code>_</code><code>PRIORITY)(stopOnShutdown)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// Registering Streaming Metrics at the start of the StreamingContext</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>assert(env.metricsSystem !</code><code>=</code>&nbsp;<code>null</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>env.metricsSystem.registerSource(streamingSource)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>uiTab.foreach(</code><code>_</code><code>.attach())</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"StreamingContext started"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>ACTIVE&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logWarning(</code><code>"StreamingContext has already been started"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>STOPPED&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code>&nbsp;<code>new</code>&nbsp;<code>IllegalStateException(</code><code>"StreamingContext has already been stopped"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;}</code> </td> 
 </tr> 
 </tbody> 
</table>

进入JobScheduler源码：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 </td> 
 <td> <code>/**&nbsp;&nbsp; JobScheduler负责逻辑层面的Job，并将其物理级别的运行在Spark之上</code> <code>&nbsp;</code><code>* This class schedules jobs to be run on Spark. It uses the JobGenerator to generate</code> <code>&nbsp;</code><code>* the jobs and runs them using a thread pool.</code> <code>&nbsp;</code><code>*/</code> <code>private</code><code>[streaming]</code> <code>class</code>&nbsp;<code>JobScheduler(</code><code>val</code>&nbsp;<code>ssc</code><code>:</code>&nbsp;<code>StreamingContext)&nbsp;</code><code>extends</code>&nbsp;<code>Logging { </code> &nbsp; <code>&nbsp;&nbsp;</code><code>//通过JobSet集合，不断地存放接收到的Job</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>jobSets</code><code>:</code>&nbsp;<code>java.util.Map[Time, JobSet]&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>ConcurrentHashMap[Time, JobSet]&nbsp;&nbsp;</code><code>//设置并行度，默认为1，想要修改作业运行的并行度在spark-conf或者应用程序中修改此值就中&nbsp; 为什么要修改并发度呢？&nbsp; 答：有时候应用程序中有多个输出，会导致多个job的执行，都是在一个batchDurations里面，job之间执行无需互相等待，所以可以通过设置此值并发执行！&nbsp;&nbsp;&nbsp;&nbsp; 不同的Batch，线程池中有很多的线程，也可以并发运行！</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>numConcurrentJobs&nbsp;</code><code>=</code>&nbsp;<code>ssc.conf.getInt(</code><code>"spark.streaming.concurrentJobs"</code><code>,&nbsp;</code><code>1</code><code>)&nbsp;&nbsp;</code><code>//将逻辑级别的Job转化为物理级别的job就是通过newDaemonFixedThreadPool线程实现的</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>jobExecutor&nbsp;</code><code>=</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs,&nbsp;</code><code>"streaming-job-executor"</code><code>)&nbsp;&nbsp;</code><code>//实例化JobGenerator</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>jobGenerator&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>JobGenerator(</code><code>this</code><code>)</code> <code>&nbsp;&nbsp;</code><code>val</code>&nbsp;<code>clock&nbsp;</code><code>=</code>&nbsp;<code>jobGenerator.clock</code> <code>&nbsp;&nbsp;</code><code>val</code>&nbsp;<code>listenerBus&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>StreamingListenerBus()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//下面三个是说在JobScheduler启动时实例化</code> <code>&nbsp;&nbsp;</code><code>// These two are created only when scheduler starts.</code> <code>&nbsp;&nbsp;</code><code>// eventLoop not being null means the scheduler has been started and not stopped</code> <code>&nbsp;&nbsp;</code><code>var</code>&nbsp;<code>receiverTracker</code><code>:</code>&nbsp;<code>ReceiverTracker&nbsp;</code><code>=</code>&nbsp;<code>null</code> <code>&nbsp;&nbsp;</code><code>// A tracker to track all the input stream information as well as processed record number</code> <code>&nbsp;&nbsp;</code><code>var</code>&nbsp;<code>inputInfoTracker</code><code>:</code>&nbsp;<code>InputInfoTracker&nbsp;</code><code>=</code>&nbsp;<code>null</code> &nbsp; <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>var</code>&nbsp;<code>eventLoop</code><code>:</code>&nbsp;<code>EventLoop[JobSchedulerEvent]&nbsp;</code><code>=</code>&nbsp;<code>null</code> &nbsp; <code>&nbsp;&nbsp;</code><code>def</code>&nbsp;<code>start()</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>synchronized { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(eventLoop !</code><code>=</code>&nbsp;<code>null</code><code>)&nbsp;</code><code>return</code>&nbsp;<code>// scheduler has already been started</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logDebug(</code><code>"Starting JobScheduler"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>eventLoop&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>EventLoop[JobSchedulerEvent](</code><code>"JobScheduler"</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>override</code>&nbsp;<code>protected</code>&nbsp;<code>def</code>&nbsp;<code>onReceive(event</code><code>:</code>&nbsp;<code>JobSchedulerEvent)</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>processEvent(event)</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>override</code>&nbsp;<code>protected</code>&nbsp;<code>def</code>&nbsp;<code>onError(e</code><code>:</code>&nbsp;<code>Throwable)</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>reportError(</code><code>"Error in job scheduler"</code><code>, e)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>eventLoop.start()</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// attach rate controllers of input streams to receive batch completion updates</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>for</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>inputDStream &lt;- ssc.graph.getInputStreams</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>rateController &lt;- inputDStream.rateController</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>} ssc.addStreamingListener(rateController)</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>listenerBus.start(ssc.sparkContext)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>receiverTracker&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>ReceiverTracker(ssc)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>inputInfoTracker&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>InputInfoTracker(ssc)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>receiverTracker.start()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobGenerator.start()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"Started JobScheduler"</code><code>)</code> <code>&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

# **二、JobScheduler深度思考** #

**下面从应用程序的输出方法print()入手，反推Job的生成过程：**

**1.点击应用程序中的print()方法后，跳入DStream的print():**

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 </td> 
 <td> <code>/**</code> <code>&nbsp;</code><code>* Print the first ten elements of each RDD generated in this DStream. This is an output</code> <code>&nbsp;</code><code>* operator, so this DStream will be registered as an output stream and there materialized.</code> <code>&nbsp;</code><code>*/</code> <code>def</code>&nbsp;<code>print()</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>ssc.withScope { </code> <code>&nbsp;&nbsp;</code><code>print(</code><code>10</code><code>)</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

2.再次点击上面红线标记的print()方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 </td> 
 <td> <code>/**</code> <code>&nbsp;</code><code>* Print the first num elements of each RDD generated in this DStream. This is an output</code> <code>&nbsp;</code><code>* operator, so this DStream will be registered as an output stream and there materialized.</code> <code>&nbsp;</code><code>*/</code> <code>def</code>&nbsp;<code>print(num</code><code>:</code>&nbsp;<code>Int)</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>ssc.withScope { </code> <code>&nbsp;&nbsp;</code><code>def</code>&nbsp;<code>foreachFunc</code><code>:</code>&nbsp;<code>(RDD[T], Time)&nbsp;</code><code>=</code><code>&gt; Unit&nbsp;</code><code>=</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>(rdd</code><code>:</code>&nbsp;<code>RDD[T], time</code><code>:</code>&nbsp;<code>Time)&nbsp;</code><code>=</code><code>&gt; { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code>&nbsp;<code>firstNum&nbsp;</code><code>=</code>&nbsp;<code>rdd.take(num +&nbsp;</code><code>1</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// scalastyle:off println</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(</code><code>"-------------------------------------------"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(</code><code>"Time: "</code>&nbsp;<code>+ time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(</code><code>"-------------------------------------------"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>firstNum.take(num).foreach(println)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(firstNum.length &gt; num) println(</code><code>"..."</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// scalastyle:on println</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps&nbsp;</code><code>=</code>&nbsp;<code>false</code><code>)</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

从图中红色标记的代码可以得出：SparkStreaming最终执行的时候还是对RDD进行各种逻辑级别的操作！

3.再次点击图上的foreachRDD进入foreachRDD方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 </td> 
 <td> <code>/**</code> <code>&nbsp;</code><code>* Apply a function to each RDD in this DStream. This is an output operator, so</code> <code>&nbsp;</code><code>* 'this' DStream will be registered as an output stream and therefore materialized.</code> <code>&nbsp;</code><code>* @param foreachFunc foreachRDD function</code> <code>&nbsp;</code><code>* @param displayInnerRDDOps Whether the detailed callsites and scopes of the RDDs generated</code> <code>&nbsp;</code><code>*&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; in the `foreachFunc` to be displayed in the UI. If `false`, then</code> <code>&nbsp;</code><code>*&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; only the scopes and callsites of `foreachRDD` will override those</code> <code>&nbsp;</code><code>*&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; of the RDDs on the display.</code> <code>&nbsp;</code><code>*/</code> <code>private</code>&nbsp;<code>def</code>&nbsp;<code>foreachRDD(</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>foreachFunc</code><code>:</code>&nbsp;<code>(RDD[T], Time)&nbsp;</code><code>=</code><code>&gt; Unit,</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>displayInnerRDDOps</code><code>:</code>&nbsp;<code>Boolean)</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;</code><code>new</code>&nbsp;<code>ForEachDStream(</code><code>this</code><code>,</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>context.sparkContext.clean(foreachFunc,&nbsp;</code><code>false</code><code>), displayInnerRDDOps).register()</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

4.点击上图的ForEachDStream进入ForEachDStream类并找到了generateJob方法：

5.上一讲中我们得出了如下的流程：

streamingcontext.start-->jobscheduler.start-->receiverTracker.start()-->JobGenterator.start()-->EventLoop-->processEvent()-->generateJobs()-->jobScheduler.receiverTracker.allocateBlocksToBatch(time)-->**graph.generateJobs(time)**

其中最后的graph.generateJobs是DSTreamGraph的方法，进入之：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>def</code>&nbsp;<code>generateJobs(time</code><code>:</code>&nbsp;<code>Time)</code><code>:</code>&nbsp;<code>Seq[Job]&nbsp;</code><code>=</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;</code><code>logDebug(</code><code>"Generating jobs for time "</code>&nbsp;<code>+ time)</code> <code>&nbsp;&nbsp;</code><code>val</code>&nbsp;<code>jobs&nbsp;</code><code>=</code>&nbsp;<code>this</code><code>.synchronized { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//此时的outputStream就是forEachDStream</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>outputStreams.flatMap { outputStream&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code>&nbsp;<code>jobOption&nbsp;</code><code>=</code>&nbsp;<code>outputStream.generateJob(time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobOption.foreach(</code><code>_</code><code>.setCallSite(outputStream.creationSite))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobOption</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>logDebug(</code><code>"Generated "</code>&nbsp;<code>+ jobs.length +&nbsp;</code><code>" jobs for time "</code>&nbsp;<code>+ time)</code> <code>&nbsp;&nbsp;</code><code>jobs</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

private val outputStreams = new ArrayBuffer[DStream[_]]()通过查看DStream的子类继承结构和上面的ForEachDStream的generateJob方法，得出DStream的子类中只有ForEachDStream override了DStream的generateJob!最终得出结论：

真正Job的生成是通过ForeachDStream的generateJob来生成的，此时的job是逻辑级别的，真正被物理级别的调用是在JobGenerator中generateJob方法中：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 </td> 
 <td> <code>/** Generate jobs and perform checkpoint for the given `time`.&nbsp; */</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>def</code>&nbsp;<code>generateJobs(time</code><code>:</code>&nbsp;<code>Time) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// Set the SparkEnv in this thread, so that job generation code can access the environment</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// Example: BlockRDDs are created in this thread, and it needs to access BlockManager</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>SparkEnv.set(ssc.env)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Try { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobScheduler.receiverTracker.allocateBlocksToBatch(time)&nbsp;</code><code>// allocate received blocks to batch</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>graph.generateJobs(time)&nbsp;</code><code>// generate jobs using allocated block</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>match</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>Success(jobs)&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>val</code>&nbsp;<code>streamIdToInputInfos&nbsp;</code><code>=</code>&nbsp;<code>jobScheduler.inputInfoTracker.getInfo(time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>Failure(e)&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobScheduler.reportError(</code><code>"Error generating jobs for time "</code>&nbsp;<code>+ time, e)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater&nbsp;</code><code>=</code>&nbsp;<code>false</code><code>))</code> <code>&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

进入jobScheduler.submitJobSet方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> <code>&lt;</code><code>strong</code><code>&gt;</code><code>//将逻辑级别的Job转化为物理级别的job就是通过newDaemonFixedThreadPool线程实现的&lt;/strong&gt;</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>jobExecutor&nbsp;</code><code>=</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs,&nbsp;</code><code>"streaming-job-executor"</code><code>)</code> </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 </td> 
 <td> <code>def</code>&nbsp;<code>submitJobSet(jobSet</code><code>:</code>&nbsp;<code>JobSet) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(jobSet.jobs.isEmpty) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"No jobs added for time "</code>&nbsp;<code>+ jobSet.time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>else</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobSets.put(jobSet.time, jobSet)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobSet.jobs.foreach(job&nbsp;</code><code>=</code><code>&gt; jobExecutor.execute(</code><code>new</code>&nbsp;<code>JobHandler(job)))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"Added jobs for time "</code>&nbsp;<code>+ jobSet.time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

至此，整个job的生成、执行就非常清晰了，最后总结如下：

**ReceiverTracker启动后会导致运行在Executor端的Receiver启动并且接收数据，ReceiverTracker会记录Receiver接收到的数据meta信息，  **

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

**JobScheduler中的线程池来提交封装的JobSet对象(时间值，Job，数据源的meta)。Job中封装了业务逻辑，导致最后一个RDD的action被触发，**

**被DAGScheduler真正调度在Spark集群上执行该Job。**