第五课：Spark Streamingf 源码运行架构图
作者：大数据技术研发人员：谢彪
一、运行架构图
二
二二、SparkStreming 运行流程
（一）、在StreamingContext调用start方法的内部其实是会启动JobScheduler的Start方法，进行消息循环，在JobScheduler 的start内部会构造JobGenerator和ReceiverTacker，并且调用JobGenerator和ReceiverTacker的start方法：
1、JobGenerator启动后会不断的根据batchDuration生成一个个的Job.
2、ReceiverTracker启动后首先在Spark Cluster中启动Receiver（其实是在Executor中先启动ReceiverSupervisor），在Receiver收到数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker，在ReceiverTracker内部会通过ReceivedBlockTracker来管理接受到的元数据信息
每个BatchInterval会产生一个具体的Job，其实这里的Job不是Spark Core中所指的Job，它只是基于DStreamGraph而生成的RDD 的DAG而已，从Java角度讲，相当于Runnable接口实例，此时要想运行Job需要提交给JobScheduler，在JobScheduler中通过线程池的方式找到一个单独的线程来提交Job到集群运行（其实是在线程中基于RDD的Action触发真正的作业的运行），为什么使用线程池呢？
1、作业不断生成，所以为了提升效率，我们需要线程池；这和在Executor中通过线程池执行Task有异曲同工之妙；
2、有可能设置了Job的FAIR公平调度的方式，这个时候也需要多线程的支持。
运行源码流程解读：
1.创建StreamingContext
StreamingContext是Spark Streaming是运行基础，也是负责管理和其运行的重要组件,其代码如下：
可以看到，StreamingContext内部包涵了一个SparkContext，这个可以告诉我们Streaming就是Spark Core上的一个应用程序。
2.创建SocketInputDStream
首先进入StreamingContext的初始化
其次进入SocketTextStream类，可知它继承一个SocketInputDstream
而SocketInputDStream又继承>DStream。
在 SocketInputDStream之中有一个关键的内部类SocketReceiver，
执行调用流程：
- 1、 ->启动线程调用SocketReceiver.onStart()
- 2、->SocketReceiver.receive()
receive方法是实现具体的逻辑代码(根据不同类型的receiver有不同的实现，建立Socket连接，然后一直循环store)
2.2 DStream，本身是RDD的模板，主要提供生成新RDD的功能
2.2.1 RDD的存储：
可以看到RDD是使用HashMap存储起来，DStream的本质就是存储的一系列的数据流。
2.2.2 RDD的生成：
getOrCompute方法完成了RDD的生成，是SparkStreaming的核心部分。
3.StreamingContext启动(start)
/**
* Start the execution of the streams.
*
* @throws IllegalStateException if the StreamingContext is already stopped.
*/
def start(): Unit = synchronized {
state match {
case INITIALIZED =>
startSite.set(DStream.getCreationSite())
StreamingContext.ACTIVATION_LOCK.synchronized {
StreamingContext.assertNoOtherContextIsActive()
try {
validate()
// Start the streaming scheduler in a new thread, so that thread local properties
// like call sites and job groups can be reset without affecting those of the
// current thread.
ThreadUtils.runInNewThread("streaming-start") {
sparkContext.setCallSite(startSite.get)
sparkContext.clearJobGroup()
sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false")
scheduler.start()
}
进入Scheduler.start:
def start(): Unit = synchronized {
if (eventLoop != null) return // scheduler has already been started
logDebug("Starting JobScheduler")
eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)
override protected def onError(e: Throwable): Unit = reportError("Error in job scheduler", e)
}
eventLoop.start()
// attach rate controllers of input streams to receive batch completion updates
for {
inputDStream <- ssc.graph.getInputStreams
rateController <- inputDStream.rateController
} ssc.addStreamingListener(rateController)
listenerBus.start(ssc.sparkContext)
receiverTracker = new ReceiverTracker(ssc)
inputInfoTracker = new InputInfoTracker(ssc)
receiverTracker.start()
jobGenerator.start()
logInfo("Started JobScheduler")
从以上可知：在JobScheduler 的start内部会构造JobGenerator和ReceiverTacker，并且调用JobGenerator和ReceiverTacker的start方法：
1、JobGenerator启动后会不断的根据batchDuration生成一个个的Job.
2、ReceiverTracker启动后首先在Spark Cluster中启动Receiver（其实是在Executor中先启动ReceiverSupervisor），在Receiver收到数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker，在ReceiverTracker内部会通过ReceivedBlockTracker来管理接受到的元数据信息
具体的内部源码运行在以后中分解！
作者：大数据技术研发人员：谢彪
- 资料来源于：DT_大数据梦工厂(Spark发行版本定制)
- DT大数据梦工厂微信公众号:DT_Spark
- 新浪微博：http://www.weibo.com/ilovepains
- 王家林老师每晚20:00免费大数据实战
YY直播：68917580