spark事件通知的三種方式

1、Spark事件通知的機制Spark 為協(xié)調(diào)各個組件完成任務(wù)及任務(wù)處理采用了多種方式進行了各個組件之間的通訊。這也是 spark 集群間及節(jié)點間協(xié)調(diào)的關(guān)鍵。下面逐一介紹。一，單個自己消息處理方式DAGSchedulerEventProsLoop 該類繼承自 EventLoop。是一個典型的生產(chǎn)消費模型。A),生產(chǎn)者通過調(diào)用DAGSchedulerEventProsLoop.t(event: E)來將消息進行發(fā)布。B),消費者了一個線程， 循環(huán)的消費消息 eventQueue.take() ， 調(diào)用EventlooponReceive(event)進行處理。DAGSchedulerEventPro

2、實現(xiàn)了 doOnReceive，sLoop對事件進行模式匹配然后交給具體的消息處理函數(shù)。private def doOnReceive(event: DAGSchedulerEvent): Unit = event match case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)=dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite,listener, properties)case MapStageS

3、ubmitted(jobId, dependency, callSite, listener, properties) =dagScheduler.handleMapStageSubmitted(jobId, dependency, callSite, listener,properties)case StageCancelled(stageId) =dagScheduler.handleStageCancellation(stageId)case JobCancelled(jobId) =dagScheduler.handleJobCancellation(jobId)case JobGro

4、upCancelled(grou) = dagScheduler.handleJobGroupCancelled(grou)case AllJobsCancelled =dagScheduler.doCancelAllJobs()case ExecutorAdded(execId, host) =C),消息緩存消息最終是于 EventLoop 的 new LinkedBlockingDequeE()里。二，SparkListeners 和 ListenerBusSparkUI 的各個指標(biāo)都是，由 ListenerBus 最為生產(chǎn)者將消息，推送到消息緩存出默認支持 1 萬，然后推送給各個 Lis

5、tener 進行處理，然后個 Listener 的數(shù)據(jù)，進行展示。的 Spark 的 webUIPage 去獲取各A)，生產(chǎn)者LiveListenerBus/StreamingListenerBus 調(diào)用其父類AsynchronousListenerBus 的t方法將消息加入 new LinkedBlockingQueueE(EVENT_QUEUE_CAPACITY)，容量 1 萬。B),消費者了一個消費者線程，線程有 while(true)進行AsynchronousListenerBus消息處理。val event = eventQueue.polltToAll(event)val eve

6、ntAdded = eventQueue.offer(event)dagScheduler.handleExecutorAdded(execId, host)case ExecutorLost(execId) =dagScheduler.handleExecutorLost(execId, fetchFailed = false)case BeginEvent(task, taskInfo) = dagScheduler.handeginEvent(task, taskInfo)case GettingResultEvent(taskInfo) =dagScheduler.handleGetT

7、askResult(taskInfo)case completion CompletionEvent(task, reason, _, _, taskInfo, taskMetrics) =dagSchedpletion(completion)case TaskSetFailed(taskSet, reason, exception) =dagScheduler.handleTaskSetFailed(taskSet, reason, exception)case ResubmitFailedStages =dagScheduler.resubmitFailedStages()C),消息的具體

8、處理ListenerBus 的tToAll 方遍歷所有了的 Listener。最終在 ontEvent 方法中將消息進行了處理。ontEvent 在源碼中的兩個重要現(xiàn):SparkListenerBus 和 StreamingListenerBus的 ontEvent。privatespark trait SparkListenerBus extends ListenerBusSparkListener,SparkListenerEvent override def ontEvent(listener: SparkListener, event: SparkListenerEvent): Uni

9、t = event match case stageSubmitted: SparkListenerStageSubmitted =listener.onStageSubmitted(stageSubmitted)casepleted: Spapleted =lpleted(pleted)case jobStart: SparkListenerJobStart =listener.onJobStart(jobStart)case jobEnd: SparkListenerJobEnd =listener.onJobEnd(jobEnd)case taskStart: SparkListener

10、TaskStart =listener.onTaskStart(taskStart)case taskGettingResult: SparkListenerTaskGettingResult =listener.onTaskGettingResult(taskGettingResult)case taskEnd: SparkListenerTaskEnd =listener.onTaskEnd(taskEnd)case environmentUpdate: SparkListenerEnvironmentUpdate =final deftToAll(event: E): Unit = va

11、l iter = listeners.iteratorwhile (iter.hasNext) val listener = iter.next()try ontEvent(listener, event) catch case NonFatal(e) =logError(sListener $Utils.getFormattedClame(listener) threw an exception, e)privatespark class StreamingListenerBusextends AsynchronousListenerBusStreamingListener,Streamin

12、gListenerEvent(StreamingListenerBus)with Logging private val logDroppedEvent = new Atomic(false) override def ontEvent(listener: StreamingListener, event:StreamingListenerEvent): Unit = event match case receiverStarted: StreamingListenerReceiverStarted =listener.onReceiverStarted(receiverStarted)cas

13、e receiverError: StreamingListenerReceiverError =listener.onReceiverError(receiverError)case receiverStopped: StreamingListenerReceiverStopped =listener.onEnvironmentUpdate(environmentUpdate)case blockManagerAdded: SparkListenerBlockManagerAdded =listener.onBlockManagerAdded(blockManagerAdded)case b

14、lockManagerRemoved: SparkListenerBlockManagerRemoved =listener.onBlockManagerRemoved(blockManagerRemoved)case unpersistRDD: SparkListenerUnpersistRDD =listener.onUnpersistRDD(unpersistRDD)case applicationStart: SparkListenerApplicationStart =listener.onApplicationStart(applicationStart)case applicat

15、ionEnd: SparkListenerApplicationEnd =listener.onApplicationEnd(applicationEnd)case metricsUpdate: SparkListenerExecutorMetricsUpdate =listener.onExecutorMetricsUpdate(metricsUpdate)case executorAdded: SparkListenerExecutorAdded =listener.onExecutorAdded(executorAdded)case executorRemoved: SparkListe

16、nerExecutorRemoved =listener.onExecutorRemoved(executorRemoved)case blockUpdated: SparkListenerBlockUpdated =listener.onBlockUpdated(blockUpdated)case logStart: SparkListenerLogStart = / ignore event log metadataD),消息的緩存消息是緩存在 AsynchronousListenerBusEVENT_QUEUE_CAPACITY=10000三，Spark 多進程之間的通訊 RPCSpar

17、k 的已經(jīng)是 netty。rpc 老版本是用 akka 實現(xiàn)的，spark1.6 以后雖然保留，akka 但是默認實現(xiàn)其實，rpc 采用 netty 之前，rpc 是通過 akka，而文件傳輸是通過 netty。現(xiàn)在相當(dāng)于全部采用了 netty 的機制。val eventQueue = new LinkedBlockingQueueE(EVENT_QUEUE_CAPACITY)listener.onReceiverStopped(receiverStopped)case batchSubmitted: StreamingListenerBatchSubmitted =listener.onBa

18、tchSubmitted(batchSubmitted)case batchStarted: StreamingListenerBatchStarted =listener.onBatchStarted(batchStarted)casepleted: Streamipleted =lpleted(pleted)case outputOperationStarted: StreamingListenerOutputOperationStarted =listener.onOutputOperationStarted(outputOperationStarted)casepleted: StreamingListenleted=listener.onpleted(pleted)case _ =override def onDropEvent(event: StreamingListenerEvent): Unit = if (pareAndSet(false, true) / Only log the following message once to avoid duplicated annoying logs. logError(DropStreamingListenerEvent because no remaining