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

版权声明：本套技术专栏是作者（秦凯新）平时工作的总结和升华，通过从真实商业环境抽取案例进行总结和分享，并给出商业应用的调优建议和集群环境容量规划等内容，请持续关注本套博客。版权声明：禁止转载，欢迎学习。

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

• Spark商业环境实战-Spark内置框架rpc通讯机制及RpcEnv基础设施
• Spark商业环境实战-Spark事件监听总线流程分析

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