一、数据处理原理剖析
每隔我们设置的batch interval 的time，就去找ReceiverTracker，将其中的，从上次划分batch的时间，到目前为止的这个batch interval time间隔内的block封装为一个batch; 其次，会将这个batch中的数据，去创建为一个初始的RDD，一个batch内，在这段时间封装了几个block，就代表这个batch对应的RDD内会有几个partition; 这个batch对应的RDD的partition决定了数据处理阶段的并行度，这个跟调优关系很大，如果想增加数据处理阶段的性能，就考虑增加并行度，那么就考虑缩短block interval; 只有output操作中，使用了ForEachStream，其中定义了generatorJob()方法，在数据处理阶段，才触发针对接收到的一个一个batch的数据，触发小的job，去处理该batch的数据; 最后一步，去找JobScheduler去调度job，job的输入RDD，就是batch对应的RDD;
二、源码分析
入口，JobGenerator的generateJobs()方法
###org.apache.spark.streaming.scheduler/JobGenerator.scala /**
* 定时，调度generateJobs()方法，传入一个time，其实就是一个batch interval内的时间段
*/
private def generateJobs(time: Time) {
// Set the SparkEnv in this thread, so that job generation code can access the environment
// Example: BlockRDDs are created in this thread, and it needs to access BlockManager
// Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed.
SparkEnv.set(ssc.env)
Try {
// 找到ReceiverTracker，调用其allocateBlocksToBatch方法，将当前时间段内的block分配给一个batch，并为其
// 创建一个RDD
jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch
// 调用DSteamGraph的generateJobs()来根据程序定义的DSteam之间的依赖关系和算子，生成job
graph.generateJobs(time) // generate jobs using allocated block
} match {
// 如果成功创建了job
case Success(jobs) =>
// 从ReceiverTracker中，获取当前batch interval对应的block数据
val receivedBlockInfos =
jobScheduler.receiverTracker.getBlocksOfBatch(time).mapValues { _.toArray }
// 用jobScheduler提交job，其对应的原始数据，是那批block
jobScheduler.submitJobSet(JobSet(time, jobs, receivedBlockInfos))
case Failure(e) =>
jobScheduler.reportError("Error generating jobs for time " + time, e)
}
eventActor ! DoCheckpoint(time)
}