Spark Streaming运行流程及源码解析（二）

Spark Streaming源码流程解析。

写在前面

以下是我自己梳理了一遍Spark Streaming程序运行的流程，过程可能有点细、有点乱。

大家可以一边看我写的流程、一边跟着步骤点进去看源码，这样就不会太乱了。

跟着源码走一遍以后，对Spark Streaming的理解也就很清晰了。

这篇文章是自己看源码过程的记录，如果有理解偏差的部分，欢迎交流指正。

开干

以如下的WordCount代码展开叙述：

// 创建SparkConf，配置master为local
val conf = new SparkConf()
    .setMaster("local[2]")
    .setAppName("socket-streaming")
// 实例化StreamingContext    
val ssc = new StreamingContext(conf, Seconds(2))
// 创建一个ReceiverInputDStream对象
val lines = ssc socketTextStream("localhost", 1234)
// 进行逻辑处理、输出
lines
    .flatMap(_.split(" "))
    .map((_, 1))
    .reduceByKey(_ + _)
    .print()
// 启动
ssc.start()
// 等待执行停止
ssc.awaitTermination()

以上代码启动后，可以接受1234端口收到的消息，然后按空格将句子切分成单词，之后对单词进行计数，每隔两秒计算输出一次结果。

接下来以我们写的WordCount代码为辅，从启动流处理引擎、接收并存储数据、处理数据、输出数据依次走一遍源码。

启动流处理引擎

StreamingContext的创建

从val ssc = new StreamingContext(conf, Seconds(2))开始，这里会实例化StreamingContext对象。

先看一下StreamingContext中的一些重要的变量。

// SparkContext实例，Spark上下文，可以通过直接传参获得，
// 也可以通过sparkConf创建，或从checkpoint中取到
private[streaming] val sc: SparkContext = {
    if (_sc != null) {
        _sc
    } else if (isCheckpointPresent) {
        SparkContext.getOrCreate(_cp.createSparkConf())
    } else {
        throw new SparkException("Cannot create StreamingContext without a SparkContext")
    }
}

// DStreamGraph用来管理DStream的依赖, 
// 创建时将StreamingContext实例绑定到DStreamGraph上
private[streaming] val graph: DStreamGraph = {
    if (isCheckpointPresent) {
        _cp.graph.setContext(this)
        _cp.graph.restoreCheckpointData()
        _cp.graph
    } else {
        require(_batchDur != null, "Batch duration for StreamingContext cannot be null")
        val newGraph = new DStreamGraph()
        newGraph.setBatchDuration(_batchDur)
        newGraph
    }
}

// JobScheduler用来生成和调度任务,
// 也会将StreamingContext实例绑定到自己身上
private[streaming] val scheduler = new JobScheduler(this)

// 批处理间隔
batchDuration

实例化StreamingContext时，这些变量都将会被实例化。

既然这样，就顺势也看一下DStreamGraph和JobScheduler中一些重要的变量。

先看一下DStreamGraph中的重要变量：

// inputStreams是输入数据源的集合，
// 输入数据源中有对应的receive方法用来接收数据
private val inputStreams = new ArrayBuffer[InputDStream[_]]()

// outputStreams就是DStream的集合，
// 我们调用的各个算子最终都会根据依赖生成的DStream，
// outputOperator型的算子都会注册到这里来
private val outputStreams = new ArrayBuffer[DStream[_]]()

再看看JobScheduler中的重要变量：

// 生成的job集合，以time为key，jobset为value的Map
private val jobSets: java.util.Map[Time, JobSet] = new ConcurrentHashMap[Time, JobSet]

// 一个线程池，用来执行job
private val jobExecutor =
    ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs, "streaming-job-executor")

// JobGenerator用来生成job
private val jobGenerator = new JobGenerator(this)

// Driver端用于管理Receiver的总管家
var receiverTracker: ReceiverTracker = null

// 事件循环，用来处理JobScheduler相关的事件
// 本质是以LinkedBlockingDeque一个队列
private var eventLoop: EventLoop[JobSchedulerEvent] = null

接下来执行val lines = ssc.socketTextStream("localhost", 1234)

如下所示，socketTextStream()会调用socketStream()，socketStream方法中会new一个SocketInputDStream，SocketInputDStream用于接收数据

def socketTextStream(
    hostname: String,
    port: Int,
    storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String] = withNamedScope("socket text stream") {
    socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)
}

def socketStream[T: ClassTag](
    hostname: String,
    port: Int,
    converter: (InputStream) => Iterator[T],
    storageLevel: StorageLevel
): ReceiverInputDStream[T] = {
    new SocketInputDStream[T](this, hostname, port, converter, storageLevel)
}

追踪一下SocketInputDStream的继承关系，发现它继承于ReceiverInputDStream，ReceiverInputDStream又继承于InputDStream。

InputDStream中有ssc.graph.addInputStream(this)这么一行代码，将InputDStream添加到DStreamGraph中的inputStreams中。

所以在new SocketInputDStream时，InputDStream就添加到DStreamGraph中了。（这个找了挺久才找见的，之前一直不知道InputDStream什么时候添加进去的）

outputOperator算子注册

接着执行如下几行代码

lines
    .flatMap(_.split(" "))
    .map((_, 1))
    .reduceByKey(_ + _)
    .print()

上面每个算子的调用会生成相互依赖的DStream: FlatMappedDStream、MappedDStream、ShuffledDStream。

只有到print()(outputOperator类算子)调用的时候，才会将DStream注册到DStreamGraph中的outputStreams中，之后DStreamGraph才能根据依赖关系生成job。

接下来跟进一下print()

// 以下的方法是依次调用的
def print(): Unit = ssc.withScope {
    print(10)
}

def print(num: Int): Unit = ssc.withScope {
    def foreachFunc: (RDD[T], Time) => Unit = {
        (rdd: RDD[T], time: Time) => {
            val firstNum = rdd.take(num + 1)
            // scalastyle:off println
            println("-------------------------------------------")
            println(s"Time: $time")
            println("-------------------------------------------")
            firstNum.take(num).foreach(println)
            if (firstNum.length > num) println("...")
            println()
            // scalastyle:on println
        }
    }
    foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps = false)
}

private def foreachRDD(
    foreachFunc: (RDD[T], Time) => Unit,
    displayInnerRDDOps: Boolean): Unit = {
    new ForEachDStream(this,
                       context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()
}

private[streaming] def register(): DStream[T] = {
    ssc.graph.addOutputStream(this)
    this
}

这里调用了register()将DStream注册到DStreamGraph的outputStreams中

到这里就将我们的业务逻辑什么的都封装到DStream中了

StreamingContext的启动

接下来走ssc.start()启动StreamingContext

StreamingContext的start方法中主要就是调用scheduler.start()，启动了JobScheduler

接下来在看看JobScheduler的start方法

def start(): Unit = synchronized {
    if (eventLoop != null) return // scheduler has already been started

    logDebug("Starting JobScheduler")
    // 事件环主要接收调度JobSchedulerEvent事件
    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()
    // 添加监听
    for {
        inputDStream <- ssc.graph.getInputStreams
        rateController <- inputDStream.rateController
    } ssc.addStreamingListener(rateController)

    // 监听总线启动
    listenerBus.start()
    receiverTracker = new ReceiverTracker(ssc)
    inputInfoTracker = new InputInfoTracker(ssc)

    val executorAllocClient: ExecutorAllocationClient = ssc.sparkContext.schedulerBackend match {
        case b: ExecutorAllocationClient => b.asInstanceOf[ExecutorAllocationClient]
        case _ => null
    }
    // 管理分配Executor
    executorAllocationManager = ExecutorAllocationManager.createIfEnabled(
        executorAllocClient,
        receiverTracker,
        ssc.conf,
        ssc.graph.batchDuration.milliseconds,
        clock)
    executorAllocationManager.foreach(ssc.addStreamingListener)
    // 启动ReceiverTracker
    receiverTracker.start()
    // 启动JobGenerator
    jobGenerator.start()
    executorAllocationManager.foreach(_.start())
    logInfo("Started JobScheduler")
}

JobScheduler中主要启动了ReceiverTracker和JobGenerator。

ReceiverTracker通知Executor启动Receiver，管理Receiver的执行，与Receiver交互。

JobGenerator用于生成job，执行job。

这两个类分别代表了接收并存储数据 和 生成job、执行job

接下来先看接收并存储数据

接收并存储数据

Driver端ReceiverTracker的操作

先从ReceiverTracker.start()说起。

def start(): Unit = synchronized {
    if (isTrackerStarted) {
        throw new SparkException("ReceiverTracker already started")
    }

    if (!receiverInputStreams.isEmpty) {
        // 建立RPC终端
        endpoint = ssc.env.rpcEnv.setupEndpoint(
            "ReceiverTracker", new ReceiverTrackerEndpoint(ssc.env.rpcEnv))
        // 加载Receiver
        if (!skipReceiverLaunch) launchReceivers()
        logInfo("ReceiverTracker started")
        trackerState = Started
    }
}

// 加载Receiver
private def launchReceivers(): Unit = {
    // 从inputStreams中获取receivers
    val receivers = receiverInputStreams.map { nis =>
        val rcvr = nis.getReceiver()
        rcvr.setReceiverId(nis.id)
        rcvr
    }
        
    runDummySparkJob()
    // 发送StartAllReceivers的消息
    logInfo("Starting " + receivers.length + " receivers")
    endpoint.send(StartAllReceivers(receivers))
}

ReceiverTracker先建立RPC终端点准备通信，监听、回复与Receiver相关的信息。

然后调用launchReceivers()，launchReceivers中的receiverInputStreams是从DStreamGraph中获取的InputStream的集合。通过InputStream获取Receiver，然后发送StartAllReceivers消息。

这里的StartAllReceivers是发给endpoint的，也就是发给ReceiverTrackerEndpoint实例，也就相当于是发给自己的。

ReceiverTrackerEndpoint的receive方法通过模式匹配进行消息的接收，在收到StartAllReceivers后，会根据资源调度分配适合启动Receiver的位置，然后调用本类的startReceiver()

override def receive: PartialFunction[Any, Unit] = {
    // Local messages
    case StartAllReceivers(receivers) =>
    // 分配适合的位置
    val scheduledLocations = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
    for (receiver <- receivers) {
        val executors = scheduledLocations(receiver.streamId)
        updateReceiverScheduledExecutors(receiver.streamId, executors)
        receiverPreferredLocations(receiver.streamId) = receiver.preferredLocation
        startReceiver(receiver, executors)
    }
}

接下来看看startReceiver方法

private def startReceiver(
    receiver: Receiver[_],
    scheduledLocations: Seq[TaskLocation]): Unit = {
    def shouldStartReceiver: Boolean = {
        !(isTrackerStopping || isTrackerStopped)
    }
    val receiverId = receiver.streamId
    if (!shouldStartReceiver) {
        onReceiverJobFinish(receiverId)
        return
    }
    val checkpointDirOption = Option(ssc.checkpointDir)
    val serializableHadoopConf =
    new SerializableConfiguration(ssc.sparkContext.hadoopConfiguration)
    // 封装在worker节点启动receiver的方法
    val startReceiverFunc: Iterator[Receiver[_]] => Unit =
    (iterator: Iterator[Receiver[_]]) => {
        if (!iterator.hasNext) {
            throw new SparkException(
                "Could not start receiver as object not found.")
        }
        if (TaskContext.get().attemptNumber() == 0) {
            val receiver = iterator.next()
            assert(iterator.hasNext == false)
            val supervisor = new ReceiverSupervisorImpl(
                receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
            supervisor.start()
            supervisor.awaitTermination()
        } else {
        }
    }
    // 使用ScheduledLocations创建RDD以在Spark作业中运行接收器
    val receiverRDD: RDD[Receiver[_]] =
    if (scheduledLocations.isEmpty) {
        ssc.sc.makeRDD(Seq(receiver), 1)
    } else {
        val preferredLocations = scheduledLocations.map(_.toString).distinct
        ssc.sc.makeRDD(Seq(receiver -> preferredLocations))
    }
    // 提交启动receiver的job到spark核心进行启动
    val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
        receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
    // We will keep restarting the receiver job until ReceiverTracker is stopped
    future.onComplete {
        ...
    }(ThreadUtils.sameThread)
    logInfo(s"Receiver ${receiver.streamId} started")
}

stratReceiver方法先封装了启动receiver的方法和RDD，然后提交给spark核心进行执行。

上面代码startReceiverFunc中，封装了创建和启动ReceiverSupervisor的操作。

ReceiverSupervisor是Executor端Receiver的管理者，负责监督和管理Executor中的Receiver的运行

Executor端ReceiverSupervisor的操作

接下来追踪ReceiverSupervisor的start方法。

/** Start the supervisor */
def start() {
    onStart()
    startReceiver()
}

// ReceiverSupervisorImpl中的onStart方法
override protected def onStart() {
    registeredBlockGenerators.asScala.foreach { _.start() }
}

// ReceiverSupervisor的方法，用于启动Receiver
def startReceiver(): Unit = synchronized {
    try {
        if (onReceiverStart()) {
            receiverState = Started
            // 启动receiver，开始接收数据
            receiver.onStart()
        } else {
            ...
        }
    } catch {
    }
}

在onStart方法中，可以看到一个registeredBlockGenerators集合，它是BlockGenerator的集合。

BlockGenerator是Receiver中比较重要的一个类，用于将我们收到的单条数据写入buffer，然后定时将buffer封装为块，进行存储和汇报给Driver。

接下来详细看一下它的变量和方法

// listener创建BlockGenerator时传进来的监听器，
// 用来监听块相关事件：onAddData、onGenerateBlock、onPushBlock
listener: BlockGeneratorListener

// 是一个ArrayBuffer，用来暂存接收到的数据
@volatile private var currentBuffer = new ArrayBuffer[Any]

// 一个队列，用来存取封装好的Block块
private val blocksForPushing = new ArrayBlockingQueue[Block](blockQueueSize)

// 定时器，定时将currentBuffer中的数据封装为Block，然后推到blocksForPushing里面
private val blockIntervalTimer =
new RecurringTimer(clock, blockIntervalMs, updateCurrentBuffer, "BlockGenerator")

// blocksForPushing队列的大小
private val blockQueueSize = conf.getInt("spark.streaming.blockQueueSize", 10)

// 这是一个线程，用来从blocksForPushing中取出Block，然后进行存储，汇报ReceiverTracker
private val blockPushingThread = new Thread() { override def run() { keepPushingBlocks() } }

// 按照时间生成块，然后将块推到blocksForPushing中
private def updateCurrentBuffer(time: Long): Unit = {
    try {
        var newBlock: Block = null
        synchronized {
            if (currentBuffer.nonEmpty) {
                val newBlockBuffer = currentBuffer
                currentBuffer = new ArrayBuffer[Any]
                val blockId = StreamBlockId(receiverId, time - blockIntervalMs)
                listener.onGenerateBlock(blockId)
                newBlock = new Block(blockId, newBlockBuffer)
            }
        }
        if (newBlock != null) {
            blocksForPushing.put(newBlock)  // put is blocking when queue is full
        }
    } catch {
    }
}

// 推送块
private def keepPushingBlocks() {
    ... 
    while (!blocksForPushing.isEmpty) {
        val block = blocksForPushing.take()
        logDebug(s"Pushing block $block")
        // 调用本类的pushBlock方法
        pushBlock(block)
        logInfo("Blocks left to push " + blocksForPushing.size())
    }
    logInfo("Stopped block pushing thread")
} catch {
    case ie: InterruptedException =>
    logInfo("Block pushing thread was interrupted")
    case e: Exception =>
    reportError("Error in block pushing thread", e)
}
}

// 推送块
private def pushBlock(block: Block) {
    listener.onPushBlock(block.id, block.buffer)
    logInfo("Pushed block " + block.id)
}

大体来说，BlockGenerator中使用了一个ArrayBuffer来不断的接收存储数据，然后会按时将ArrayBuffer中的数据封装为Block。另有一个队列ArrayBlockingQueue来存取Block，一边存一边取，这样实现了单条数据的接收与存储。

再接着看pushBlock的操作。其中调用了listener.onPushBlock()。

listener是构造BlockGenerator时传进来的，使用的是ReceiverSupervisorImpl中的defaultBlockGeneratorListener。

private val defaultBlockGeneratorListener = new BlockGeneratorListener {
    def onAddData(data: Any, metadata: Any): Unit = { }

    def onGenerateBlock(blockId: StreamBlockId): Unit = { }

    def onError(message: String, throwable: Throwable) {
        reportError(message, throwable)
    }
    // 推块的时候调用，它又会调用ReceiverSupervisorImpl.pushArrayBuffer()
    def onPushBlock(blockId: StreamBlockId, arrayBuffer: ArrayBuffer[_]) {
        pushArrayBuffer(arrayBuffer, None, Some(blockId))
    }
}

// 将接收到的数据的ArrayBuffer作为数据块存储到Spark的内存中
def pushArrayBuffer(
    arrayBuffer: ArrayBuffer[_],
    metadataOption: Option[Any],
    blockIdOption: Option[StreamBlockId]
) {
    // 调用pushAndReportBlock()
    pushAndReportBlock(ArrayBufferBlock(arrayBuffer), metadataOption, blockIdOption)
}

// 将块数据进行存储，然后汇报给Driver
def pushAndReportBlock(
    receivedBlock: ReceivedBlock,
    metadataOption: Option[Any],
    blockIdOption: Option[StreamBlockId]
) {
    val blockId = blockIdOption.getOrElse(nextBlockId)
    val time = System.currentTimeMillis
    // 这步会真正的存储数据
    val blockStoreResult = receivedBlockHandler.storeBlock(blockId, receivedBlock)
    logDebug(s"Pushed block $blockId in ${(System.currentTimeMillis - time)} ms")
    val numRecords = blockStoreResult.numRecords
    val blockInfo = ReceivedBlockInfo(streamId, numRecords, metadataOption, blockStoreResult)
    // 将存储结果报告Driver
    if (!trackerEndpoint.askSync[Boolean](AddBlock(blockInfo))) {
        throw new SparkException("Failed to add block to receiver tracker.")
    }
    logDebug(s"Reported block $blockId")
}

listener.onPushBlock会调用pushArrayBuffer()，pushArrayBuffer方法会调用pushAndReportBlock()将数据进行存储，然后汇报给Driver。

这里需要注意一下：BlockGenerator负责单条数据的接收与生成快。这个一会会再说。

开始接收数据、存储数据

BlockGenerator的内部看完以后，接着回到ReceiverSupervisor.start()中来

def start() {
    onStart()
    startReceiver()
}

onStart()方法中启动BlockGenerator，启动块生成的定时器和推送块的线程

def start(): Unit = synchronized {
    if (state == Initialized) {
        state = Active
        blockIntervalTimer.start()
        blockPushingThread.start()
        logInfo("Started BlockGenerator")
    } else {
        throw new SparkException(
            s"Cannot start BlockGenerator as its not in the Initialized state [state = $state]")
    }
}

startReceiver()方法中，调用receiver.onStart()，开始接收数据

def startReceiver(): Unit = synchronized {
    try {
        if (onReceiverStart()) {
            receiverState = Started
            // 启动receiver开始接收数据
            receiver.onStart()
        } else {
            stop("Registered unsuccessfully because Driver refused to start receiver " + streamId, None)
        }
    } catch {
    }
}

以我们一开写的demo中的SocketInputDStream为例，它会生成一个SocketReceiver实例，以下是SocketReceiver的onStart方法。

def onStart() {
    try {
        // 启动socket，开始监听
        socket = new Socket(host, port)
    } catch {
    }
    new Thread("Socket Receiver") {
        setDaemon(true)
        override def run() { receive() }
    }.start()
}

def receive() {
    try {
        // 接收数据
        val iterator = bytesToObjects(socket.getInputStream())
        while(!isStopped && iterator.hasNext) {
            // 将接收到的数据进行存储
            store(iterator.next())
        }
    } catch {
        ...
    } finally {
        onStop()
    }
}

可以看到，onStart中启动了一个线程，开始不断的接收数据，之后会调用store()将接收到的数据进行存储。

这里的store()方法是Receiver中定义的，我们跟进一下。

def store(dataItem: T) {
    supervisor.pushSingle(dataItem)
}

/** Store an ArrayBuffer of received data as a data block into Spark's memory. */
def store(dataBuffer: ArrayBuffer[T]) {
    supervisor.pushArrayBuffer(dataBuffer, None, None)
}

/**
* Store an ArrayBuffer of received data as a data block into Spark's memory.
* The metadata will be associated with this block of data
* for being used in the corresponding InputDStream.
*/
def store(dataBuffer: ArrayBuffer[T], metadata: Any) {
    supervisor.pushArrayBuffer(dataBuffer, Some(metadata), None)
}

/** Store an iterator of received data as a data block into Spark's memory. */
def store(dataIterator: Iterator[T]) {
    supervisor.pushIterator(dataIterator, None, None)
}

会发现有好几个重载的方法，参数不尽相同。

SocketReceiver中调用的是store(dataItem: T)这个方法，它会调用pushSingle将数据添加到BlockGenerator中的currentBuffer中。BlockGenerator再定时将currentBuffer封装为Block，然后调用pushBlock、pushArrayBuffer、pushAndReportBlock对数据进行存储、汇报Driver。

store(dataItem: T)就相当于之前说的接收单条数据进行存储的操作。

另外几个重载方法也都会最终也都会调用pushAndReportBlock数据进行存储，然后报告Driver。这里就不再跟下去了。

数据的接收与存储到这里就结束了。接下来我们在回到JobGenerator解析一下job的生成和执行。

生成job、执行job

JobGenerator介绍

视线在跳回到JobGenerator这边来，先看看JobGenerator中几个重要变量

// job生成消息的事件环
private var eventLoop: EventLoop[JobGeneratorEvent] 

// 定时器，按照批处理间隔定时向eventLoop发送生成job的消息
private val timer = new RecurringTimer(
    clock, ssc.graph.batchDuration.milliseconds,
    longTime => eventLoop.post(GenerateJobs(new Time(longTime))), 
    "JobGenerator"
)

接下来看看JobGenerator的start方法

def start(): Unit = synchronized {
    if (eventLoop != null) return 
    checkpointWriter
    // eventLoop的回调方法onReceive会调用processEvent(event)进行事件的处理
    eventLoop = new EventLoop[JobGeneratorEvent]("JobGenerator") {
        override protected def onReceive(event: JobGeneratorEvent): Unit = processEvent(event)
        override protected def onError(e: Throwable): Unit = {
            jobScheduler.reportError("Error in job generator", e)
        }
    }
    // 启动事件环
    eventLoop.start()

    if (ssc.isCheckpointPresent) {
        restart()
    } else {
        startFirstTime()
    }
}

start方法中会启动eventLoop和调用startFirstTime()。

eventLoop启动后，会启动一个线程来不断的接收消息，根据接收到的消息作出相应的操作

看一下startFirstTime()，startFirstTime中启动了DStreamGraph 和 用于定时发送生成job消息的定时器

/** Starts the generator for the first time */
private def startFirstTime() {
    val startTime = new Time(timer.getStartTime())
    // 启动DStreamGraph
    graph.start(startTime - graph.batchDuration)
    // 启动定时器timer
    timer.start(startTime.milliseconds)
    logInfo("Started JobGenerator at " + startTime)
}

DStreamGraph的start方法就不跟进了，没有很重要的东西。

timer启动后，会定时发送GenerateJobs(new Time(longTime))的消息。eventLoop在收到消息后，调用processEvent方法进行处理，如下：

private def processEvent(event: JobGeneratorEvent) {
    logDebug("Got event " + event)
    event match {
        case GenerateJobs(time) => generateJobs(time)
        case ClearMetadata(time) => clearMetadata(time)
        case DoCheckpoint(time, clearCheckpointDataLater) =>
        doCheckpoint(time, clearCheckpointDataLater)
        case ClearCheckpointData(time) => clearCheckpointData(time)
    }
}

生成job

接下来就开始generateJobs的旅程了。

首先processEvent会将GenerateJobs消息通过调用JobGenerator.generateJobs()进行处理。

以下是JobGenerator的generateJobs方法：

// 根据时间生成job
private def generateJobs(time: Time) {

    ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")
    Try {
        // 调用receiverTracker给批分配数据
        jobScheduler.receiverTracker.allocateBlocksToBatch(time)
        // 在DStreamGraph中根据分配的块生成job
        graph.generateJobs(time) 
    } match {
        // 如果job生成成功，调用jobScheduler.submitJobSet提交job
        case Success(jobs) =>
        val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
        jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
        // 失败则打报告
        case Failure(e) =>
        jobScheduler.reportError("Error generating jobs for time " + time, e)
        PythonDStream.stopStreamingContextIfPythonProcessIsDead(e)
    }
    // 完成后进行checkpoint 
    eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
}

首先会调用receiverTracker.allocateBlocksToBatch()给当前批分配需要处理的数据，之后调用DStreamGraph.generateJobs()生成job序列，如果生成成功，调用jobScheduler.submitJobSet提交job。

先跟进一下DStreamGraph.generateJobs()：

def generateJobs(time: Time): Seq[Job] = {
    logDebug("Generating jobs for time " + time)
    val jobs = this.synchronized {
        // 根据outputStream生成job
        outputStreams.flatMap { outputStream =>
            val jobOption = outputStream.generateJob(time)
            jobOption.foreach(_.setCallSite(outputStream.creationSite))
            jobOption
        }
    }
    logDebug("Generated " + jobs.length + " jobs for time " + time)
    jobs
}

发现这里会遍历outputStreams生成job，outputStreams中存放的是我们调用的outputOperation算子对应的DStream，也就是之前说的调用outputOperation算子将DStream注册到DStreamGraph中的outputStreams中。

以我们最开始的WordCount代码为例，我们的代码最终会添加一个ForEachDStream到outputStreams中去。

所以就会调用这里就调用ForEachDStream.generateJob()来生成job。

以下是ForEachDStream的generateJob方法：

override def generateJob(time: Time): Option[Job] = {
    parent.getOrCompute(time) match {
        case Some(rdd) =>
        val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
            foreachFunc(rdd, time)
        }
        Some(new Job(time, jobFunc))
        case None => None
    }
}

generateJob方法会调用parent.getOrCompute()生成RDD，如果生成成功，以RDD和我们定义的逻辑处理函数构造Job，并返回job。

需要注意一下这里的parent，parent其实就是它所依赖的上一个DStream的引用，

lines
    .flatMap(_.split(" "))
    .map((_, 1))
    .reduceByKey(_ + _)
    .print()

以我们写的代码为例，这里的parent就是由reduceByKey算子生成的ShuffledDStream的引用，ShuffledDStream中的parent是map生成的MappedDStream的引用，MappedDStream中的parent是flatMap生成的FlatMappedDStream的引用。

FlatMappedDStream中的parent就是SocketInputDStream的引用

跟进一下parent.getOrCompute()，现在的parent是ShuffledDStream的引用

private[streaming] final def getOrCompute(time: Time): Option[RDD[T]] = {
    // 已经生成的RDD集合，是以时间为key，rdd为value的HashMap
    generatedRDDs.get(time).orElse {
        if (isTimeValid(time)) {
            val rddOption = createRDDWithLocalProperties(time, displayInnerRDDOps = false) {
                // 执行compute方法，生成rdd，几乎每个DStream子类都会实现这个方法
                SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
                    compute(time)
                }
            }
            // 对生成的rdd缓存或checkpoint，添加到已经生成的RDD集合中
            rddOption.foreach { case newRDD =>
                if (storageLevel != StorageLevel.NONE) {
                    newRDD.persist(storageLevel)
                }
                if (checkpointDuration != null && (time - zeroTime).isMultipleOf(checkpointDuration)) {
                    newRDD.checkpoint()
                }
                generatedRDDs.put(time, newRDD)
            }
            rddOption
        } else {
            None
        }
    }
}

DStream中定义了一个generatedRDDs用来存储已经生成的RDD。

会先去generatedRDDs中获取当前批的RDD，如果不存在则执行compute()生成RDD。

按我们写的代码来走的话，调用的是ShuffledDStream的compute方法。

override def compute(validTime: Time): Option[RDD[(K, C)]] = {
    parent.getOrCompute(validTime) match {
        case Some(rdd) => Some(rdd.combineByKey[C](
            createCombiner, mergeValue, mergeCombiner, partitioner, mapSideCombine))
        case None => None
    }
}

发现又调用了parent.getOrCompute生成RDD。

我们就可以发现它是根据依赖关系，循环的去调用getOrCompute和compute，直到最开始的DStream。

我们代码中最开始的是SocketInputDStream，会调用SocketInputDStream实例的compute方法，SocketInputDStream没有compute方法，这里调用的是他的父类ReceiverInputDStream的compute方法。

override def compute(validTime: Time): Option[RDD[T]] = {
    val blockRDD = {
        if (validTime < graph.startTime) {
            new BlockRDD[T](ssc.sc, Array.empty)
        } else {
            // 获取当前分配给当前批的块信息 
            val receiverTracker = ssc.scheduler.receiverTracker
            val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)
            val inputInfo = StreamInputInfo(id, blockInfos.flatMap(_.numRecords).sum)
            ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)
            // 根据批时间和块信息创建RDD，并返回
            createBlockRDD(validTime, blockInfos)
        }
    }
    Some(blockRDD)
}

一系列操作生成RDD完成后，回到ForEachDStream的generateJob方法，

override def generateJob(time: Time): Option[Job] = {
    parent.getOrCompute(time) match {
        case Some(rdd) =>
        val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
            foreachFunc(rdd, time)
        }
        Some(new Job(time, jobFunc))
        case None => None
    }
}

根据生成的RDD和业务处理函数封装成job，返回job到DStream.generateJobs()

DStream.generateJobs()再将job返回到JobGenerator.generateJobs()中来

此刻，我们的job就生成完成了。

提交执行job

接下来JobGenerator.generateJobs()中会执行jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))，将job进行提交。

def submitJobSet(jobSet: JobSet) {
    if (jobSet.jobs.isEmpty) {
      logInfo("No jobs added for time " + jobSet.time)
    } else {
      listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))
      jobSets.put(jobSet.time, jobSet)
      jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
      logInfo("Added jobs for time " + jobSet.time)
    }
  }

这里会将job封装到JobHandler中进行处理，JobHandler是一个线程类，其中会执行job.run运行job。

以下是Job的run方法，其中的func()就是我们封装进来的业务处理函数。

def run() {
    _result = Try(func())
}

将JobHandler扔到线程池中执行，我们的job就跑起来了。

输出数据

job跑起来后，会根据我们封装的func()，执行对应的输出。

end...

至此，Spark Streaming源码流程解析就over了。

多敲、多看、多搬砖、加油。

个人公众号：码农峰，定时推送行业资讯，持续发布原创技术文章，欢迎大家关注。