本文基于spark 2.11
1. 前言
spark使用RDD来抽象的表示数据，用户使用RDD提供的一些算子编写自己的spark application，使用RDD抽象表示数据要求对于输入数据是静态的，但是在流式数据处理中数据如同流水一样不停的在管道中产生，这不符合RDD的要求。Spark Streaming的处理方式是，从输入流中读区数据，将数据作为一个个batch保存起来，这样就有了静态的数据，就可以用RDD来表示这些数据，然后就可以基于RDD 创建任务了。
2. 基本原理
下面是一个从kafka读取数据处理的代码：
import org.apache.spark._
import org.apache.spark.streaming._
val conf = new SparkConf().setMaster("local[2]").setAppName("KafkaWordCount")
// batchDuration 设置为 1 秒，然后创建一个 streaming 入口
// 每1秒依据RDD中创建一次job，输入RDD就从已经已经收集的batch中取。
val ssc = new StreamingContext(conf, Seconds(1))
val kafkaParams: Map[String, String] = Map("group.id" -> "test",...)
val topics = Map("test" -> 1)
val lines = KafkaUtils.createStream(
ssc,
kafkaParams,
topics, StorageLevel.MEMORY_AND_DISK)
val words = lines.flatMap(_.split(" "))
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.print()
wordCounts.foreachRDD(...)
ssc.start()
ssc.awaitTermination()
和之前基于RDD的的wordcount程序不同：
- KafkaUtils.createStream(...)创建出来的不是RDD，和是一个DStream的类，
- DStream同样存在map、flatMap、reduceByKey这样的转换操作，但是它是从DStream到DStream的转换。
- print在RDD里表示一种action，会触发job的创建和提交，但是DStream的action操作不会，它的处理方式不同，后续会介绍。
- ssc.start会启动一下组建：
- JobScheduler, 调度和追踪job
- JobGenerator，由JobScheduler启动，定时（初始化StreamingContext指定的时间）从DStream创建出job
- ReceiverTracker， 运行在Driver上，收集 从各个receiver上报的流数据batch信息
- ReceiverSupervisor，由ReceiverTracker运行发送消息使其在executor上运行，接收Receiver汇报的batch数据，然后将数据信息汇报给ReceiverTracker。
- Receiver， 运行在executor上，由ReceiverSupervisor启动，负责着流中读区数据，分成batch，汇报给ReceiverSupervisor。
从上介绍可以看出，job是在ssc.start过程中创建的，而且在运行期间会根据用户设置的duration不断的创建。
下图表示了使用kafka作为输入源时的streaming工作期间的流程：
- Receiver1从kafka中读入数据，将数据转发给ReceiverSupervisor
- ReceiverSupervisor，使用BlockManager存储并管理数据信息。
- ReceiverSupervisor，将数据信息发送给运行在Driver上的ReceiverTracker。
- JobGenerator，假设上面wordCount代码中DStream之间的转换看作一张DAG，DStreamGraph保存了所有的DAG。JobGenerate每隔一段时间从DStreamGraph中的DStream DAG生成RDD的DAG，然后提交RDD的job。RDD的数据则来自Generator根据ReceiverTracker中收集的batch数据信息。
2.1 DStream 到RDD的转换
由于基于RDD计算的基于静态的数据，而数据是不断产生的，spark streaming将输入数据切成一个个batch，因此需要不断的产生job去计算batch中的数据。
上面wordCount程序，描述了DStream之间的转换，看起来几乎和RDD之间的转换是一样的，JobGenerator运行期间根据DStream不停创建RDD，再由RDD生成job 经SparkContext提交运行。DStream相当于模板，RDD相当于使用模板创造出的零件，而JobGenerator则相当于操作模板的工人了。
下图描述了DStream和RDD在运行期间的关系：
可以看到DStream的子类都有一个RDD的对应类，一句DStream生成的RDD DAG和DStream拥有一样的转换和依赖。采集输入流中的一段数据作为RDD的源数据。
RDD#compute方法中完成输入数据的计算，DStream也存在compute方法，但是其compute方法这是完成DStream到RDD的转换。
2.2 ReceiverInputDStream
所有继承DStream的类中，ReceiverInputDStream除了像其他DStream一样创建出RDD以外，还需要返回一个Receiver负责接收收据，例如ReceiverInputDStream的子类SocketInputDStream就能返回一个SocketReceiver的Receiver的实现类。
ReceiverInputDStream一般都是一个DStream DAG的源头。
当ReceiverTracker调用start启动时，它会从DStreamGraph持有的DStream DAG中获得所有的ReceiverInputDStream，然后取得Receiver，通过巧妙的方式将Reciver包装成Task，然后发送到executor上执行，然后在receiver端，Receiver和ReceiverSupervisor启动接收数据。
在SparkStreaming(3) ReceiverTracker和Receiver中，启动receiver时，receiver就是按上面方式获得的。
2.3 output 操作
DStream和RDD有着类似的操作，map这种使得RDD转换成新的RDD的操作称为Transformation，foreach这种触发job的创建和提交的操作称为Action， DStream类似，Dstream到DStream的称为Transformation, DStream的output操作有点类似rdd中的action操作，一个action意味着一个新的job被创建提交。DStream的output操作意味着一个DStream DAG模板的创建，也意味着到此处DStream转换成RDD应该触发job，DStream常见的output操作有：
- saveAsTextFiles
- saveAsObjectFiles
- foreachRDD
等
3 DStreamGraph
DStreamGraph用来保存所有output操作生成DStream DAG。
比如下面是DStream#foreachRDD的代码：
private def foreachRDD(
foreachFunc: (RDD[T], Time) => Unit,
displayInnerRDDOps: Boolean): Unit = {
// 调用了DStream#register()方法
new ForEachDStream(this,
context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()
}
//register方法向DStreamGraph注册当前DStream
// 由于DStream保存了所有的父依赖，因此注册当前DStream
// 就能追溯出整个DStream DAG，相当于注册了DStream DAG
private[streaming] def register(): DStream[T] = {
ssc.graph.addOutputStream(this)
this
}
上面说DStream的output操作相当于触发一个DStream DAG模板的创建，而一个模板对应一种job。 第一节wordcount代码中分别有print和foreachRDD两个output操作，因此DStreamGraph可以理解持有两个DStream DAG，如下图：
尽管创建出来的DStream DAG是一样的，但是依然会创建出两份RDD DAG，生成两类job，
DStreamGraph还肩负着从根据注册的DStreamDAG创建job的任务，后续JobGenerator就是调用DStreamGraph完成创建job。下面是DstreamGraph创建job的方法：
def generateJobs(time: Time): Seq[Job] = {
logDebug("Generating jobs for time " + time)
val jobs = this.synchronized {
// 对每一个因output操作而注册的DStream DAG生成job
outputStreams.flatMap { outputStream =>
val jobOption = outputStream.generateJob(time)
jobOption.foreach(_.setCallSite(outputStream.creationSite))
jobOption
}
}
logDebug("Generated " + jobs.length + " jobs for time " + time)
jobs
}
这是DStream#generateJob方法，time表示每一次生成job的时间
private[streaming] def generateJob(time: Time): Option[Job] = {
// getOrCompute将DStream转换成RDD，转换操作是从当前
// DStream往上游追溯，追溯到源头后在一次往下生成RDD的过程
// 是一次DFS的过程。
getOrCompute(time) match {
case Some(rdd) =>
val jobFunc = () => {
val emptyFunc = { (iterator: Iterator[T]) => {} }
// 创建到RDD后提交job
context.sparkContext.runJob(rdd, emptyFunc)
}
Some(new Job(time, jobFunc))
case None => None
}
}
