Spark Streaming为啥要设置两条线程？.doc

《Spark Streaming为啥要设置两条线程？.doc》由会员上传分享，免费在线阅读，更多相关内容在教育资源-天天文库。

1、JobScheduler详解一：JobSheduler的源码解析1.JobScheduler是SparkStreaming整个调度的核心，相当于SparkCore上的DAGScheduler.2.SparkStreaming为啥要设置两条线程？setMaster指定的两条线程是指程序运行的时候至少需要两条线程。一条线程用于接收数据，需要不断的循环。而我们指定的线程数是用于作业处理的。3.JobSheduler的启动是在StreamContext的start方法被调用的时候启动的。defstart():Unit=synchronized{state

2、match{caseINITIALIZED=>startSite.set(DStream.getCreationSite())StreamingContext.ACTIVATION_LOCK.synchronized{StreamingContext.assertNoOtherContextIsActive()try{validate()//而这里面启动的新线程是调度方面的，因此和我们设置的线程数没有关系。//Startthestreamingschedulerinanewthread,sothatthreadlocalproperties//l

3、ikecallsitesandjobgroupscanberesetwithoutaffectingthoseofthe//currentthread.ThreadUtils.runInNewThread("streaming-start"){sparkContext.setCallSite(startSite.get)sparkContext.clearJobGroup()sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL,"false")schedu

4、ler.start()}4.jobScheduler会负责逻辑层面的Job，并将其物理级别的运行在Spark之上./***ThisclassschedulesjobstoberunonSpark.ItusestheJobGeneratortogenerate*thejobsandrunsthemusingathreadpool.*/private[streaming]classJobScheduler(valssc:StreamingContext)extendsLogging{5.jobScheduler的start方法源码如下：defstar

5、t():Unit=synchronized{if(eventLoop!=null)return//schedulerhasalreadybeenstartedlogDebug("StartingJobScheduler")eventLoop=newEventLoop[JobSchedulerEvent]("JobScheduler"){overrideprotecteddefonReceive(event:JobSchedulerEvent):Unit=processEvent(event)overrideprotecteddefonError(

6、e:Throwable):Unit=reportError("Errorinjobscheduler",e)}eventLoop.start()//attachratecontrollersofinputstreamstoreceivebatchcompletionupdatesfor{inputDStream<-ssc.graph.getInputStreamsrateController<-inputDStream.rateController}ssc.addStreamingListener(rateController)listenerB

7、us.start(ssc.sparkContext)receiverTracker=newReceiverTracker(ssc)inputInfoTracker=newInputInfoTracker(ssc)receiverTracker.start()jobGenerator.start()logInfo("StartedJobScheduler")}6.其中processEvent的源码如下：privatedefprocessEvent(event:JobSchedulerEvent){try{eventmatch{caseJobStar

8、ted(job,startTime)=>handleJobStart(job,startTime)caseJobCompleted(jo