|
在接受到GenerateJob事件的时候，会执行generateJobs代码，就是在该代码内部产生和调度job的。
- /** Generate jobs and perform checkpointing for the given `time`. */
- private def generateJobs(time: Time) {
- // Checkpoint all RDDs marked for checkpointing to ensure their lineages are
- // truncated periodically. Otherwise, we may run into stack overflows (SPARK-6847).
- ssc.sparkContext.setLocalProperty(RDD.CHECKPOINT_ALL_MARKED_ANCESTORS, "true")
- Try {
- jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch
- graph.generateJobs(time) // generate jobs using allocated block
- } match {
- case Success(jobs) =>
- val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
- jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
- case Failure(e) =>
- jobScheduler.reportError("Error generating jobs for time " + time, e)
- PythonDStream.stopStreamingContextIfPythonProcessIsDead(e)
- }
- eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
- }
（编辑：西安站长网）
【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!
|
