通过源码呈现 Spark Streaming 的底层机制。
1. 初始化与接收数据
Spark Streaming 通过分布在各个节点上的接收器，缓存接收到的流数据，并将流数 据 包 装 成 Spark 能 够 处 理 的 RDD的格式， 输入到Spark Streaming， 之 后由Spark Streaming将作业提交到Spark集群进行执行，如图1所示。
图 1 Spark Streaming 执行模型
初始化的过程主要可以概括为两点：
1）调度器的初始化。
调度器调度 Spark Streaming 的运行，用户可以通过配置相关参数进行调优。
2）将输入流的接收器转化为 RDD 在集群进行分布式分配，然后启动接收器集合中的每个接收器。
针对不同的数据源， Spark Streaming 提供了不同的数据接收器，分布在各个节点上的每个接收器可以认为是一个特定的进程，接收一部分流数据作为输入。
用户也可以针对自身生产环境状况，自定义开发相应的数据接收器。
如图 2 所示，接收器分布在各个节点上。通过下面代码，创建并行的、在不同Worker 节点分布的 receiver 集合。
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) … }
图 2 Spark Streaming 接收器
2. 数据接收与转化
在上面的“初始化与接收数据”部分中已经介绍过， receiver 集合转换为 RDD，在集群上分布式地接收数据流。那么每个 receiver 是怎样接收并处理数据流的呢？读者可以通过图 3，对输入流的处理有一个全面的了解。图 3为 Spark Streaming 数据接收与转化的示意图。
图 3 的主要流程如下。
1）数据缓冲：在 receiver 的 receive 函数中接收流数据，将接收到的数据源源不断地放入到 BlockGenerator.currentBuffer。
2）缓冲数据转化为数据块：在 BlockGenerator 中有一个定时器（RecurringTimer），将 当 前 缓 冲 区 中 的 数 据 以 用 户 定 义 的 时 间 间 隔 封 装 为 一 个 数 据 块 Block， 放 入 到
BlockGenerator 的 blocksForPush 队列中（这个队列）。
3）数据块转化为 Spark 数据块：在 BlockGenerator 中有一个 BlockPushingThread线程，不断地将 blocksForPush 队列中的块传递给 BlockManager，让 BlockManager 将
数据存储为块。 BlockManager 负责 Spark 中的块管理。
4）元数据存储：在 pushArrayBuffer 方法中还会将已经由 BlockManager 存储的元数据信息（例如： Block 的 id 号）传递给 ReceiverTracker， ReceiverTracker 会将存储的
blockId 放到对应 StreamId 的队列中。
图 3 Spark Streaming 数据接收与转化
图中部分组件的作用如下：
KeepPushingBlocks：调用此方法持续写入和保持数据块。
pushArrayBuffer：调用 pushArrayBuffer 方法将数据块存储到 BlockManager 中。
reportPushedBlock：存储完成后汇报数据块信息到主节点。
receivedBlockInfo（ Meta Data）：已经接收到的数据块元数据记录。
streamId：数据流 Id。
BlockInfo：数据块元数据信息。
BlockManager.put：数据块存储器写入备份数据块到其他节点。
Receiver ：数据块接收器，接收数据块。
BlockGenerator：数据块生成器，将数据缓存生成 Spark 能处理的数据块。
BlockGenerator.currentBuffer ：缓存网络接收的数据记录，等待之后转换为 Spark的数据块。
BlockGenerator.blocksForPushing ：将一块连续数据记录暂存为数据块，待后续转换为 Spark 能够处理的 BlockManager 中的数据块（A Block As a BlockManager’s Block)。
BlockGenerator.blockPushingThread：守护线程负责将数据块转换为 BlockManager中数据块。
ReceiveTracker：输入数据块的元数据管理器，负责管理和记录数据块。
BlockManager： Spark 数据块管理器，负责数据块在内存或磁盘的管理。
RecurringTimer：时间触发器，每隔一定时间进行缓存数据的转换。
上面的过程中涉及最多的类就是 BlockGenerator，在数据转化的过程中其扮演者不可或缺的角色。
private[streaming] class BlockGenerator( listener: BlockGeneratorListener, receiverId: Int, conf: SparkConf ) extends Logging
3. 生成 RDD 与提交 Spark Job
Spark Streaming 根据时间段，将数据切分为 RDD，然后触发 RDD 的 Action 提交 Job， Job 被 提 交 到 Job Manager 中 的 Job Queue 中 由 Job Scheduler 调 度， 之 后Job Scheduler 将 Job 提交到 Spark 的 Job 调度器，然后将 Job 转换为大量的任务分发给 Spark 集群执行，如图 4 所示。
图 4 Spark Streaming 调度模型
Job generator 中通过下面的方法生成 Job 进行调度和执行。
从下面的代码可以看出 job 是从 outputStream 中生成的，然后再触发反向回溯执行
整个 DStream DAG，类似 RDD 的机制。
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]())) } }
Spark Streaming 在保证实时处理的要求下还能够保证高吞吐与容错性。用户的数据分析中很多情况下也存在需要分析图数据，运行图算法，通过 GraphX 可以简便地开发分布式图分析算法。
本文转自大数据躺过的坑博客园博客，原文链接：http://www.cnblogs.com/zlslch/p/5725374.html，如需转载请自行联系原作者
