Spark Streaming 原理剖析

 val tempRDD =
if (hasLocationPreferences) {
val receiversWithPreferences = receivers.map(r => (r,
Seq(r.preferredLocation.get)))
ssc.sc.makeRDD[Receiver[_]](receiversWithPreferences)
} else {
// 在这里创造 RDD 相当于进入 SparkContext.makeRDD
// 此处将 receivers 的集合作为一个 RDD 进行分区 RDD[Receiver]
// 即使是只有一个输入流按照这个分布式也是流的输入端在 worker 而不再 Master
…

// 将 receivers 的集合打散然后启动它们
…
ssc.sparkContext.runJob(tempRDD, startReceiver)
…
}

private[streaming] class BlockGenerator(
listener: BlockGeneratorListener,
receiverId: Int,
conf: SparkConf
) extends Logging

private def generateJobs(time: Time) {
SparkEnv.set(ssc.env)
Try(graph.generateJobs(time)) match {
case Success(jobs) =>
// 获取输入数据块的元数据信息
val receivedBlockInfo = graph.getReceiverInputStreams.map { stream =>
. . .
}.toMap
jobScheduler.submitJobSet(JobSet(time, jobs, receivedBlockInfo))
case Failure(e) =>
jobScheduler.reportError("Error generating jobs for time " + time, e)
}
eventActor !DoCheckpoint(time)
}
// 下 面 进 入 JobScheduler 的 submitJobSet 方 法 一 探 究 竟 JobScheduler 是 整 个 Spark
Streaming 调度的核心组件
def submitJobSet(jobSet: JobSet) {
. . .
jobSets.put(jobSet.time, jobSet)
jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
. . .
}
// 进入 Graph 生成 job 的方法 Graph 本质是 DStreamGraph 类生成的对象
final private[streaming] class DStreamGraph extends Serializable with

Logging {
def generateJobs(time: Time): Seq[Job] = {
. . .
private val inputStreams = new ArrayBuffer[InputDStream[_]]()
private val outputStreams = new ArrayBuffer[DStream[_]]()
. . .
val jobs = this.synchronized {
outputStreams.flatMap(outputStream => outputStream.generateJob(time))
. . .
}
// outputStreams 中的对象是 DStream下面进入 DStream 的 generateJob 一探究竟
private[streaming] def generateJob(time: Time): Option[Job] = {
getOrCompute(time) match {
case Some(rdd) => {
val jobFunc = () => {
val emptyFunc = { (iterator: Iterator[T]) => {} }
// 此处相当于针对每个时间段生成的一个 RDD会调用 SparkContext 的方法 runJob 提交 Spark 的一
个 Job
context.sparkContext.runJob(rdd, emptyFunc)
}
Some(new Job(time, jobFunc))
}
case None => None
}
}
// 在 DStream 算是父类一些具体的 DStream 例如 SocketInputStream 等的类的父类可以通过
SocketInputDStream 看是如何通过上面的 getOrCompute 生成 RDD 的
private[streaming] def getOrCompute(time: Time): Option[RDD[T]] = {
generatedRDDs.get(time) match {
. . .
case None => {
if (isTimeValid(time)) {
// Dstream 是个父类这里代表的是子类的 compute 方法 DStream 通过 compute 调用用户自定
义函数。当任务执行时同一个 stage 中的 DStream 函数会串联依次执行
compute(time) match {
. . .
generatedRDDs.put(time, newRDD)
. . .
}
在 SocketInputDStream 的 compute 方法中生成了对应时间片的 RDD
override def compute(validTime: Time): Option[RDD[T]] = {
if (validTime >= graph.startTime) {
val blockInfo = ssc.scheduler.receiverTracker.getReceivedBlockInfo(id)

receivedBlockInfo(validTime) = blockInfo
val blockIds = blockInfo.map(_.blockId.asInstanceOf[BlockId])
Some(new BlockRDD[T](ssc.sc, blockIds))
} else {
Some(new BlockRDD[T](ssc.sc, Array[BlockId]()))
}
}