Spark事件监听总线流程分析-Spark商业环境实战

本套系列博客从真实商业环境抽取案例进行总结和分享，并给出Spark源码解读及商业实战指导，请持续关注本套博客。版权声明：本套Spark源码解读及商业实战归做者（秦凯新）全部，禁止转载，欢迎学习。

Spark商业环境实战及调优进阶系列

• Spark商业环境实战-Spark内置框架rpc通信机制及RpcEnv基础设施
• Spark商业环境实战-Spark事件监听总线流程分析
• Spark商业环境实战-Spark存储体系底层架构剖析
• Spark商业环境实战-Spark底层多个MessageLoop循环线程执行流程分析
• Spark商业环境实战-Spark二级调度系统Stage划分算法和最佳任务调度细节剖析
• Spark商业环境实战-Spark任务延迟调度及调度池Pool架构剖析
• Spark商业环境实战-Task粒度的缓存聚合排序结构AppendOnlyMap详细剖析
• Spark商业环境实战-ExternalSorter 排序器在Spark Shuffle过程当中设计思路剖析
• Spark商业环境实战-StreamingContext启动流程及Dtream 模板源码剖析
• Spark商业环境实战-ReceiverTracker与BlockGenerator数据流接收过程剖析

1. Spark事件监听总线流程分析

1.1 Spark事件监听总线流程分析

以下图所示事件日志监听器EventLoggingListener，实现了SparkListenerInterface接口，重写了全部的事件处理函数，包括Stage提交，stage完成等。

经过在SparkContext初始化时把日志监听器EventLoggingListener注册到LiveListenerBus事件总线上,并启动LiveListenerBus内部的Thread线程，监听提交到总线上的事件，调用SparkListenerBus的eventQueue.poll -->postToAll(event) --> doPostEvent方法，并进行事件匹配后处理,如：EventLoggingListener执行StageSubmited提交。

1.2 Spark UI 事件监听总线流程分析

Spark UI的可视化展现，是有不一样的监听器实现的，他们都分别注册在LiveListenerBus上，以下面SparkContext的初始化片断：

if (conf.getBoolean("spark.ui.enabled", true)) {
    Some(SparkUI.createLiveUI(this, _conf, listenerBus, _jobProgressListener,
      _env.securityManager, appName, startTime = startTime))
  } else {
    // For tests, do not enable the UI
    None
  }
复制代码

下面片断展现的是SparkUI.createLiveUI方法，能够看到监听器的注册，经过事件的投递（如：DAGScheduler ，DriverEndpoint等），从而实现UI的数据展现：

val _jobProgressListener: JobProgressListener = jobProgressListener.getOrElse {
  val listener = new JobProgressListener(conf)
  listenerBus.addListener(listener)
  listener
}

val environmentListener = new EnvironmentListener
val storageStatusListener = new StorageStatusListener(conf)
val executorsListener = new ExecutorsListener(storageStatusListener, conf)
val storageListener = new StorageListener(storageStatusListener)
val operationGraphListener = new RDDOperationGraphListener(conf)

listenerBus.addListener(environmentListener)
listenerBus.addListener(storageStatusListener)
listenerBus.addListener(executorsListener)
listenerBus.addListener(storageListener)
listenerBus.addListener(operationGraphListener)
复制代码

3 结语

秦凯新 于深圳 2018-10-28