每隔我们设置的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;

###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)

  }