sparkstreaming为啥要设置两条线程.docVIP

JobScheduler详解 一：JobSheduler的源码解析 1. JobScheduler是Spark Streaming整个调度的核心，相当于Spark Core上的DAGScheduler. 2. Spark Streaming为啥要设置两条线程？ setMaster指定的两条线程是指程序运行的时候至少需要两条线程。一条线程用于接收数据，需要不断的循环。而我们指定的线程数是用于作业处理的。 3. JobSheduler的启动是在StreamContext的start方法被调用的时候启动的。 def start(): Unit = synchronized { state match { case INITIALIZED = startSite.set(DStream.getCreationSite()) StreamingContext.ACTIVATION_LOCK.synchronized { StreamingContext.assertNoOtherContextIsActive() try { validate() //而这里面启动的新线程是调度方面的，因此和我们设置的线程数没有关系。 // Start the streaming scheduler in a new thread, so that thread local properties // like call sites and job groups can be reset without affecting those of the // current thread. ThreadUtils.runInNewThread(streaming-start) { sparkContext.setCallSite(startSite.get) sparkContext.clearJobGroup() sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, false) scheduler.start() } 4. jobScheduler会负责逻辑层面的Job，并将其物理级别的运行在Spark之上. /** * This class schedules jobs to be run on Spark. It uses the JobGenerator to generate * the jobs and runs them using a thread pool. */ private[streaming] class JobScheduler(val ssc: StreamingContext) extends Logging { 5. jobScheduler的start方法源码如下： def start(): Unit = synchronized { if (eventLoop != null) return // scheduler has already been started logDebug(Starting JobScheduler) eventLoop = new EventLoop[JobSchedulerEvent](JobScheduler) { override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event) override protected def onError(e: Throwable): Unit = reportError(Error in job scheduler, e) } eventLoop.start() // attach rate controllers of input streams to receive batch completion updates for { inputDStream - ssc.graph.getInputStreams rateController - inputDStream.rateController } ssc.addStreamingListener(rateContr