假设你已经了解job是如何被划分及提交的，若不了解请前往spark streaming 流程详解

当前位置是JobGenerator类的generateJobs的方法，我们重点看上面的generateJobs方法中的这一段：

jobScheduler.receiverTracker.allocateBlocksToBatch(time) 
graph.generateJobs(time)

allocateBlocksToBatch这里做了什么呢？

//我们可以看到receiverTracker中这个方法的实现：
def allocateBlocksToBatch(batchTime: Time): Unit = {
    if (receiverInputStreams.nonEmpty) {
      receivedBlockTracker.allocateBlocksToBatch(batchTime)
    }
  }

通过receivedBlockTracker将真正的数据：block，和batch对应起来。其中receivedBlockTracker有两个map用来存放block信息：streamIdToUnallocatedBlockQueues、timeToAllocatedBlocks。详细不讲啦，有兴趣的可以自己扒代码

⚠️接下来看下：graph.generateJobs(time)，这里DStreamGraph做了什么事情呢？

def generateJobs(time: Time): Seq[Job] = {
    
    val jobs = this.synchronized {
      outputStreams.flatMap { outputStream =>
        val jobOption = outputStream.generateJob(time)
      }
    }
   
    jobs
  }

可以看到原来是调用了outputstream的generateJob的方法，我们看一个outputstream的实现，如：ForEachDStream

private[streaming]
class ForEachDStream[T: ClassTag] (
   parent: DStream[T],
   foreachFunc: (RDD[T], Time) => Unit,
   displayInnerRDDOps: Boolean
 ) extends DStream[Unit](parent.ssc) {

 ....

 override def generateJob(time: Time): Option[Job] = {
   parent.getOrCompute(time) match {
     case Some(rdd) =>
       val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
         foreachFunc(rdd, time)//这里很有意思，以装饰器的模式执行了我们代码里写的那一堆对rdd的处理
       }
       Some(new Job(time, jobFunc))
     case None => None
   }
 }
}

在这里我们可以看到，outputstream其实是调用了parent的Compute方法，一层一层递归，最会会调用到inputstream的Compute方法，那么我们一起看下inputstream的Compute方：

override def compute(validTime: Time): Option[RDD[T]] = {
   val blockRDD = {

    
   val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)

   createBlockRDD(validTime, blockInfos)
   }
   Some(blockRDD)
 }

可以看到首先是通过receiverTracker获取到了batch的所有block info，然后new了rdd，这就跟前面的receiverTracker.allocateBlocksToBatch(time)对上了