SparkStreaming作业提交源码分析数据处理篇在调用StreamingContext的start函数的时候，会调用JobScheduler的start函数。而JobScheduler的start函数会启动ReceiverTracker和jobGenerator。在启动jobGenerator的时候，系统会根据这次是从Checkpoint恢复与否分别调用restart和startFirstTime函数。/**Startgenerationofjobs*/defstart():Unit=synchronized{if(eventActor!=null)return//generatorhasalreadybeenstartedeventActor=ssc.env.actorSystem.actorOf(Props(newActor{defreceive={caseevent:JobGeneratorEvent=>processEvent(event)}}),"JobGenerator")if(ssc.isCheckpointPresent){restart()}else{startFirstTime()}}}startFirstTime函数会分别启动DStreamGraph和JobGenerator线程privatedefstartFirstTime(){valstartTime=newTime(timer.getStartTime())graph.start(startTime-graph.batchDuration)timer.start(startTime.milliseconds)logInfo("StartedJobGeneratorat"+startTime)}privatevaltimer=newRecurringTimer(clock,ssc.graph.batchDuration.milliseconds,longTime=>eventActor!GenerateJobs(newTime(longTime)),"JobGenerator")JobGenerator线程会每隔ssc.graph.batchDuration.milliseconds的时间生成Jobs，这个时间就是我们初始化StreamingContext的时候传进来的，生成Jobs是通过Akka调用generateJobs方法：*Generatejobsandperformcheckpointforthegiven`time`.*/privatedefgenerateJobs(time:Time){//SetthesparkEnvinthisthread,sothatjobgenerationcodecanaccessthe//environmentExample:BlockRDDsarecreatedinthisthread,anditneeds//toaccessBlockManager//Update:ThisisprobablyredundantafterthreadlocalstuffinsparkEnvhas//beenremoved.SparkEnv.set(ssc.env)Try{//allocatereceivedblockstobatchjobScheduler.receiverTracker.allocateBlocksToBatch(time)graph.generateJobs(time)//generatejobsusingallocatedblock}match{caseSuccess(jobs)=>valreceivedBlockInfos=jobScheduler.receiverTracker.getBlocksOfBatch(time).mapValues{_.toArray}jobScheduler.submitJobSet(JobSet(time,jobs,receivedBlockInfos))caseFailure(e)=>jobScheduler.reportError("Errorgeneratingjobsfortime"+time,e)}eventActor!DoCheckpoint(time)}在generateJobs方法中的jobScheduler.receiverTracker.allocateBlocksToBatch(time)很重要，其最终调用的是allocateBlocksToBatch函数，其定义如下：defallocateBlocksTo