[spark streaming]Job动态生成和深度思考

你的名字 2022-05-15 10:44 160阅读 0赞

**本期内容：**

**1、Spark Streaming Job生成深度思考**

**2、Spark Streaming Job生成源码解析**

# **一、Spark Streaming Job生成深度思考** #

**源码解析：**

**1. 在大数据处理场景中，如果不是流处理的话，一般会有定时任务。例如10分钟触发一次，1个小时触发一次，这就是做流处理的感觉，一切不是流处理，或者与流处理无关的数据都将是没有价值的数据，以前做批处理的时候其实也是隐形的在做流处理，一切处理终将被流处理统一！！**

DStreams其实有三种类型：

**第一种是输入的DStrams，可以有各种不同的数据来源构建的Stream**，例如来自Socket，Kafka，Flume等；

**第二种是输出的DStrams，outputStreams **是逻辑级别的Action，由于还是Spark Streaming框架级别的，底层还是会翻译成物理级别的Action，就是ADD的Action；

**第三种是Transforms操作从一种DStream转变为另一种DStream**，即基于其他DStream产生的。其中DStreamGraph 类记录了数据来源的DStream，和输出类型的DStream

产生DStreams有两种方式：

DSTreams要么基于数据源产生，要么基于其它的DStreams产生；

SparkStreaming流处理基于时间作为触发器，Storm基于事件作为触发器，基于一个又一个的Record!!

# 二**、Spark Streaming Job生成源码解析** #

Spark 作业动态生成三大核心：   
JobGenerator: 负责Job生成。 基于源数据生成；  
JobSheduler： 负责Job调度。 基于源数据调度；  
ReceiverTracker: 获取元数据。

**JobGenerator****和ReceiverTracker是JobScheduler的成员，从JobScheduler的start方法可以看出！！**

**跟踪源码得到如下运行流程：**

streamingcontext.start-->jobscheduler.start-->receiverTracker.start()-->JobGenterator.start()-->EventLoop-->processEvent()-->generateJobs()-->jobScheduler.receiverTracker.allocateBlocksToBatch(time)-->graph.generateJobs(time)-->jobScheduler.inputInfoTracker.getInfo(time)-->jobScheduler.submitJobSet-->startFirstTime()-->graph.start()-->timer.start()

**具体逻辑图如下：**

**说明：此图引用了spark版本定制班成员[http://lqding.blog.51cto.com/9123978/1772958][http_lqding.blog.51cto.com_9123978_1772958]的成果，表示深深的感谢**

![235186-20160521143038498-824916772.png][]

**下面结合调式过程和job生成结构图，逐步追踪源码：**

1.val ssc = new StreamingContext(conf, Seconds(5))

ssc.start()程序运行的入口；

2.进入jobscheduler.start()

3.进入receiverTracker.start()方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 </td> 
 <td> <code>/** Start the endpoint and receiver execution thread. */</code> <code>def</code>&nbsp;<code>start()</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>synchronized { </code> <code>&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(isTrackerStarted) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code>&nbsp;<code>new</code>&nbsp;<code>SparkException(</code><code>"ReceiverTracker already started"</code><code>)</code> <code>&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(!receiverInputStreams.isEmpty) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>endpoint&nbsp;</code><code>=</code>&nbsp;<code>ssc.env.rpcEnv.setupEndpoint(&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//ReceiverTracker接收到源数据后保存在ReceiverTrackerEndpoint中</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>"ReceiverTracker"</code><code>,&nbsp;</code><code>new</code>&nbsp;<code>ReceiverTrackerEndpoint(ssc.env.rpcEnv))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(!skipReceiverLaunch) launchReceivers()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"ReceiverTracker started"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>trackerState&nbsp;</code><code>=</code>&nbsp;<code>Started</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

4.进入jobGenerator.start()方法：

首先看EventLoop类

<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>/**</code> <code>&nbsp;</code><code>* An event loop to receive events from the caller and process all events in the event thread. It</code> <code>&nbsp;</code><code>* will start an exclusive event thread to process all events.</code> <code>&nbsp;</code><code>*</code> <code>&nbsp;</code><code>* Note: The event queue will grow indefinitely. So subclasses should make sure `onReceive` can</code> <code>&nbsp;</code><code>* handle events in time to avoid the potential OOM. * EvenLoop类中有，后台线程从队列中获取消息，然后调用onReceive方法对该消息进行处理，这里的onReceive方法即匿名内部类中重写onReceive方法的processEvent方法。</code> <code>&nbsp;</code><code>*/</code> <code>private</code><code>[spark]&nbsp;</code><code>abstract</code>&nbsp;<code>class</code>&nbsp;<code>EventLoop[E](name</code><code>:</code>&nbsp;<code>String)&nbsp;</code><code>extends</code>&nbsp;<code>Logging { </code> <code>&nbsp;&nbsp;&nbsp;</code><code>//存储消息的LinkedBlockingDeque和后台线程</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>eventQueue</code><code>:</code>&nbsp;<code>BlockingQueue[E]&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>LinkedBlockingDeque[E]()</code> &nbsp; <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>stopped&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>AtomicBoolean(</code><code>false</code><code>)</code> &nbsp; <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>val</code>&nbsp;<code>eventThread&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>Thread(name) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>setDaemon(</code><code>true</code><code>)</code><code>//后台线程</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>override</code>&nbsp;<code>def</code>&nbsp;<code>run()</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>while</code>&nbsp;<code>(!stopped.get) {&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>val</code>&nbsp;<code>event&nbsp;</code><code>=</code>&nbsp;<code>eventQueue.take()</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>onReceive(event)</code><code>//对消息进行处理，这里的onReceive方法即匿名内部类中重写onReceive方法的processEvent方法</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>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>onError(e)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&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;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>NonFatal(e)&nbsp;</code><code>=</code><code>&gt; logError(</code><code>"Unexpected error in "</code>&nbsp;<code>+ name, e)</code> <code>&nbsp;&nbsp;&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>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&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;</code><code>case</code>&nbsp;<code>ie</code><code>:</code>&nbsp;<code>InterruptedException&nbsp;</code><code>=</code><code>&gt;&nbsp;</code><code>// exit even if eventQueue is not empty</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>NonFatal(e)&nbsp;</code><code>=</code><code>&gt; logError(</code><code>"Unexpected error in "</code>&nbsp;<code>+ name, e)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;</code><code>def</code>&nbsp;<code>start()</code><code>:</code>&nbsp;<code>Unit&nbsp;</code><code>=</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(stopped.get) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>throw</code>&nbsp;<code>new</code>&nbsp;<code>IllegalStateException(name +&nbsp;</code><code>" has already been stopped"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// Call onStart before starting the event thread to make sure it happens before onReceive</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>onStart()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>eventThread.start()</code> <code>&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

进入最重要的processEvent方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 </td> 
 <td> <code>//processEvent方法是对消息类型进行模式匹配，然后路由到对应处理该消息的方法中。消息的处理一般是发给另外一个线程来处理的，消息循环器不处理耗时的业务逻辑/** Processes all events */</code> <code>&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>def</code>&nbsp;<code>processEvent(event</code><code>:</code>&nbsp;<code>JobGeneratorEvent) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logDebug(</code><code>"Got event "</code>&nbsp;<code>+ event)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>event&nbsp;</code><code>match</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>GenerateJobs(time)&nbsp;</code><code>=</code><code>&gt; generateJobs(time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>ClearMetadata(time)&nbsp;</code><code>=</code><code>&gt; clearMetadata(time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>DoCheckpoint(time, clearCheckpointDataLater)&nbsp;</code><code>=</code><code>&gt;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>doCheckpoint(time, clearCheckpointDataLater)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>case</code>&nbsp;<code>ClearCheckpointData(time)&nbsp;</code><code>=</code><code>&gt; clearCheckpointData(time)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

进入generateJobs方法：

<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>在获取到数据后调用DStreamGraph的generateJobs方法来生成Job，具体如下步骤所示：&nbsp;&nbsp;</code><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 {&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//第一步：获取当前时间段里面的数据。根据分配的时间来分配具体要处理的数据。</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobScheduler.receiverTracker.allocateBlocksToBatch(time)&nbsp;</code><code>// allocate received blocks to batch&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //第二步：生成Job，获取RDD的DAG依赖关系。在此基于DStream生成了RDD实例</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;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//第三步：获取streamIdToInputInfos的信息。BacthDuractions要处理的数据，以及我们要处理的业务逻辑</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)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//第四步：将生成的Job交给jobScheduler</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>}&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//第五步：进行checkpoint</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>

进入DStreamGraph的generateJobs方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 </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>outputStreams.flatMap { outputStream&nbsp;</code><code>=</code><code>&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//这里的outputStreams是整个DStream中的最后一个DStream。outputStream.generateJob(time)类似于RDD中从后往前推</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>

进入onReceive方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 </td> 
 <td> <code>/**</code> <code>&nbsp;</code><code>* Invoked in the event thread when polling events from the event queue.</code> <code>&nbsp;</code><code>*</code> <code>&nbsp;</code><code>* Note: Should avoid calling blocking actions in `onReceive`, or the event thread will be blocked</code> <code>&nbsp;</code><code>* and cannot process events in time. If you want to call some blocking actions, run them in</code> <code>&nbsp;</code><code>* another thread.&lt;br&gt;不断的从消息队列中获得消息，一旦获得消息就会处理。&nbsp;&lt;br&gt;不要在onReceive中添加阻塞的消息，如果这样的话会不断的阻塞消息。&nbsp;&lt;br&gt;消息循环器一般都不会处理具体的业务逻辑，一般消息循环器发现消息以后都会将消息路由给其他的线程去处理</code> <code>&nbsp;</code><code>*/</code> <code>protected</code>&nbsp;<code>def</code>&nbsp;<code>onReceive(event</code><code>:</code>&nbsp;<code>E)</code><code>:</code>&nbsp;<code>Unit</code> </td> 
 </tr> 
 </tbody> 
</table>

其中submitJobSet方法，只是把JobSet放到ConcurrentHashMap中，把Job封装为JobHandler提交到jobExecutor线程池中

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>def</code>&nbsp;<code>submitJobSet(jobSet</code><code>:</code>&nbsp;<code>JobSet) { </code> <code>&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(jobSet.jobs.isEmpty) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"No jobs added for time "</code>&nbsp;<code>+ jobSet.time)</code> <code>&nbsp;&nbsp;</code><code>}&nbsp;</code><code>else</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobSets.put(jobSet.time, jobSet)</code> <code>&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;</code><code>logInfo(</code><code>"Added jobs for time "</code>&nbsp;<code>+ jobSet.time)</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>}</code> &nbsp; &nbsp; <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]</code> </td> 
 </tr> 
 </tbody> 
</table>

JobHandler对象为实现Runnable 接口，job的run方法导致了func的调用，即基于DStream的业务逻辑

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 </td> 
 <td> <code>def</code>&nbsp;<code>submitJobSet(jobSet</code><code>:</code>&nbsp;<code>JobSet) { </code> <code>&nbsp;&nbsp;</code><code>if</code>&nbsp;<code>(jobSet.jobs.isEmpty) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>logInfo(</code><code>"No jobs added for time "</code>&nbsp;<code>+ jobSet.time)</code> <code>&nbsp;&nbsp;</code><code>}&nbsp;</code><code>else</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>jobSets.put(jobSet.time, jobSet)</code> <code>&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;</code><code>logInfo(</code><code>"Added jobs for time "</code>&nbsp;<code>+ jobSet.time)</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>}</code> &nbsp; <code>&nbsp;</code>&nbsp; <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]</code> </td> 
 </tr> 
 </tbody> 
</table>

**最后总结两点：**

1.  **Action RDD触发作业的执行，这个时候作为Runnable接口封装，它会定义一个方法，方法里面是基于DStream的依赖关系生成的RDD。翻译的时候是将DStream的依赖关系翻译成RDD的依赖关系，由于DStream的依赖关系最后一个是action级别的，翻译成RDD的时候，RDD的最后一个操作也应该是action级别的，如果翻译的时候直接执行的话，就直接生成了Job，就没有所谓的队列，所以会将翻译的事件放到一个函数中或者一个方法中，因此，如果这个函数没有指定的action触发作业是执行不了的。**
2.  **Spark Streaming根据时间不断的去管理我们生成的作业，这个时候我们每个作业又有action级别的操作，这个action操作是对DStream进行逻辑级别的操作，它生成每个Job放到队列的时候，一定会被翻译为RDD的操作，那基于RDD操作的最后一个一定是action级别的，如果翻译的话直接就是触发action的话整个Spark Streaming的Job就不受管理了。因此我们既要保证它的翻译，又要保证对它的管理，把DStream之间的依赖关系转变为RDD之间的依赖关系，最后一个DStream使得action的操作，翻译成一个RDD之间的action操作，整个翻译后的内容它是一块内容，这一块内容是放在一个函数体中的，这个函数体，就是函数的定义，这个函数由于它只是定义还没有执行，所以它里面的RDD的action不会执行，不会触发Job,当我们的JobScheduler要调度Job的时候，转过来在线程池中拿出一条线程执行刚才的封装的方法。**

[http_lqding.blog.51cto.com_9123978_1772958]: http://lqding.blog.51cto.com/9123978/1772958
[235186-20160521143038498-824916772.png]: /images/20220515/942e302e2880477f93bb689165ba93b6.png