Spark2.1.0——深入理解事件总线
Spark2.1.0——深入理解事件总线
概览
Spark程序在运行的过程中，Driver端的很多功能都依赖于事件的传递和处理，而事件总线在这中间发挥着至关重要的纽带作用。事件总线通过异步线程，提高了Driver执行的效率。
Spark定义了一个特质[1]ListenerBus，可以接收事件并且将事件提交到对应事件的监听器。为了对ListenerBus有个直观的理解，我们先来看看它的代码实现，见代码清单1。
代码清单1 ListenerBus的定义
private[spark] trait ListenerBus[L <: AnyRef, E] extends Logging {
private[spark] val listeners = new CopyOnWriteArrayList[L]
final def addListener(listener: L): Unit = {
listeners.add(listener)
}
final def removeListener(listener: L): Unit = {
listeners.remove(listener)
}
final def postToAll(event: E): Unit = {
val iter = listeners.iterator
while (iter.hasNext) {
val listener = iter.next()
try {
doPostEvent(listener, event)
} catch {
case NonFatal(e) =>
logError(s"Listener ${Utils.getFormattedClassName(listener)} threw an exception", e)
}
}
}
protected def doPostEvent(listener: L, event: E): Unit
private[spark] def findListenersByClass[T <: L : ClassTag](): Seq[T] = {
val c = implicitly[ClassTag[T]].runtimeClass
listeners.asScala.filter(_.getClass == c).map(_.asInstanceOf[T]).toSeq
}
}
代码清单1中展示了ListenerBus是个泛型特质，其泛型参数为 [L <: AnyRef, E]，其中L是代表监听器的泛型参数，可以看到ListenerBus支持任何类型的监听器，E是代表事件的泛型参数。ListenerBus中各个成员的作用如下：
- listeners：用于维护所有注册的监听器，其数据结构为CopyOnWriteArrayList[L]；
- addListener：向listeners中添加监听器的方法，由于listeners采用CopyOnWriteArrayList来实现，所以addListener方法是线程安全的；
- removeListener：从listeners中移除监听器的方法，由于listeners采用CopyOnWriteArrayList来实现，所以removeListener方法是线程安全的；
- postToAll：此方法的作用是将事件投递给所有的监听器。虽然CopyOnWriteArrayList本身是线程的安全的，但是由于postToAll方法内部引入了“先检查后执行”的逻辑，因而postToAll方法不是线程安全的，所以所有对postToAll方法的调用应当保证在同一个线程中；
- doPostEvent：用于将事件投递给指定的监听器，此方法只提供了接口定义，具体实现需要子类提供；
- findListenersByClass：查找与指定类型相同的监听器列表。
下面将分别对以下内容进行介绍：
- ListenerBus的继承体系
- SparkListenerBus详解
- LiveListenerBus详解
[1] 特质是Scala语言中提供真正的多重继承的语法特性，类似于Java的Interface，但是又可以实现方法。有关Scala特质的更多介绍请访问Scala官网http://www.scala-lang.org。
ListenerBus的继承体系
理解了ListenerBus的定义后，本小节一起来看看有哪些类继承了它。ListenerBus的类继承体系如图1所示。
图1 ListenerBus的类继承体系
从图1中可以看到有三种ListenerBus的具体实现，分别为：
- SparkListenerBus：用于将SparkListenerEvent类型的事件投递到SparkListenerInterface类型的监听器；
- StreamingQueryListenerBus：用于将StreamingQueryListener.Event类型的事件投递到StreamingQueryListener类型的监听器，此外还会将StreamingQueryListener.Event类型的事件交给SparkListenerBus；
- StreamingListenerBus：用于将StreamingListenerEvent类型的事件投递到StreamingListener类型的监听器，此外还会将StreamingListenerEvent类型的事件交给SparkListenerBus。
SparkListenerBus也有两种实现：
- LiveListenerBus：采用异步线程将SparkListenerEvent类型的事件投递到SparkListener类型的监听器；
- ReplayListenerBus：用于从序列化的事件数据中重播事件。
有了对事件总线的这些介绍，读者已经在宏观上对其有所认识。但是如果没有具体的实现，ListenerBus本身也无法发挥作用。下一小节我们将选择对SparkListenerBus从更加微观的角度说明如何使用事件总线。
SparkListenerBus详解
有了上一节对ListenerBus类继承体系的介绍，本小节将详细介绍SparkListenerBus的实现，见代码清单2。
代码清单2 SparkListenerBus的实现
private[spark] trait SparkListenerBus
extends ListenerBus[SparkListenerInterface, SparkListenerEvent] { protected override def doPostEvent(
listener: SparkListenerInterface,
event: SparkListenerEvent): Unit = {
event match {
case stageSubmitted: SparkListenerStageSubmitted =>
listener.onStageSubmitted(stageSubmitted)
case stageCompleted: SparkListenerStageCompleted =>
listener.onStageCompleted(stageCompleted)
case jobStart: SparkListenerJobStart =>
listener.onJobStart(jobStart)
case jobEnd: SparkListenerJobEnd =>
listener.onJobEnd(jobEnd)
case taskStart: SparkListenerTaskStart =>
listener.onTaskStart(taskStart)
case taskGettingResult: SparkListenerTaskGettingResult =>
listener.onTaskGettingResult(taskGettingResult)
case taskEnd: SparkListenerTaskEnd =>
listener.onTaskEnd(taskEnd)
case environmentUpdate: SparkListenerEnvironmentUpdate =>
listener.onEnvironmentUpdate(environmentUpdate)
case blockManagerAdded: SparkListenerBlockManagerAdded =>
listener.onBlockManagerAdded(blockManagerAdded)
case blockManagerRemoved: SparkListenerBlockManagerRemoved =>
listener.onBlockManagerRemoved(blockManagerRemoved)
case unpersistRDD: SparkListenerUnpersistRDD =>
listener.onUnpersistRDD(unpersistRDD)
case applicationStart: SparkListenerApplicationStart =>
listener.onApplicationStart(applicationStart)
case applicationEnd: SparkListenerApplicationEnd =>
listener.onApplicationEnd(applicationEnd)
case metricsUpdate: SparkListenerExecutorMetricsUpdate =>
listener.onExecutorMetricsUpdate(metricsUpdate)
case executorAdded: SparkListenerExecutorAdded =>
listener.onExecutorAdded(executorAdded)
case executorRemoved: SparkListenerExecutorRemoved =>
listener.onExecutorRemoved(executorRemoved)
case blockUpdated: SparkListenerBlockUpdated =>
listener.onBlockUpdated(blockUpdated)
case logStart: SparkListenerLogStart => // ignore event log metadata
case _ => listener.onOtherEvent(event)
}
} }
我们看到SparkListenerBus已经实现了ListenerBus的doPostEvent方法，通过对SparkListenerEvent事件的匹配，执行SparkListenerInterface监听器的相应方法。
这里的SparkListenerEvent其实是个特质，代码清单2中列出的SparkListenerStageSubmitted、SparkListenerStageCompleted等都是继承了SparkListenerEvent特质的样例类[2]。为说明问题，这里仅仅摘选SparkListenerEvent及部分SparkListenerEvent子类的实现如下：
@DeveloperApi
@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "Event")
trait SparkListenerEvent {
protected[spark] def logEvent: Boolean = true
} @DeveloperApi
case class SparkListenerStageSubmitted(stageInfo: StageInfo, properties: Properties = null)
extends SparkListenerEvent @DeveloperApi
case class SparkListenerStageCompleted(stageInfo: StageInfo) extends SparkListenerEvent @DeveloperApi
case class SparkListenerTaskStart(stageId: Int, stageAttemptId: Int, taskInfo: TaskInfo)
extends SparkListenerEvent
// 省略其他SparkListenerEvent的实现
private[spark] case class SparkListenerLogStart(sparkVersion: String) extends SparkListenerEvent
SparkListenerInterface也是一个特质，其中定义了所有SparkListener应当遵守的接口规范。由于SparkListenerInterface中定义了很多接口，为说明问题只摘抄SparkListenerInterface中的部分接口定义，代码如下：
private[spark] trait SparkListenerInterface {
def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit
def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted): Unit
// 省略其他接口方法
def onOtherEvent(event: SparkListenerEvent): Unit
}
结合代码清单2，我们知道以上代码片段中的onStageCompleted和onStageSubmitted将在SparkListenerBus的doPostEvent方法中分别匹配到SparkListenerStageCompleted和SparkListenerStageSubmitted事件时执行，而对于doPostEvent中无法匹配的事件，都将执行onOtherEvent方法。
在详细介绍了ListenerBus及SparkListenerBus后，我们知道当有事件需要通知监听器的时候，可以调用ListenerBus的postToAll方法，postToAll方法遍历所有监听器并调用SparkListenerBus实现的doPostEvent方法，doPostEvent方法对事件类型进行匹配后调用监听器的不同方法。整个投递事件的过程是通过方法调用实现的，所以这是一个同步调用。在监听器比较多的时候这个过程会相对比较耗时（比如用于写日志的EventLoggingListener在调度频繁的时候，有可能导致写入延迟，这将导致部分事件的丢失。此问题已在spark2.3.0版本中得到改进。），在Spark UI（在《Spark内核设计的艺术 架构设计与实现》一书的第4章中详细介绍）中为了达到页面的即时刷新 ，实现了SparkListenerBus的子类LiveListenerBus。下一小节将围绕LiveListenerBus来详细说明异步投递消息的实现细节。
[2] 样例类是Scala语言的语法特性。样例类是一种特殊的类型，常用作事件、参数、模式匹配等。有关样例类的更多介绍，请读者阅读Scala语言的相关资料。
LiveListenerBus详解
LiveListenerBus继承了SparkListenerBus，并实现了将事件异步投递给监听器，达到实时刷新UI界面数据的效果。LiveListenerBus主要由以下部分组成：
- eventQueue：是SparkListenerEvent事件的阻塞队列，队列大小可以通过Spark属性spark.scheduler.listenerbus.eventqueue.size进行配置，默认为10000（Spark早期版本中属于静态属性，固定为10000，这导致队列堆满时，只得移除一些最老的事件，最终导致各种问题与bug）；
- started：标记LiveListenerBus的启动状态的AtomicBoolean类型的变量；
- stopped：标记LiveListenerBus的停止状态的AtomicBoolean类型的变量；
- droppedEventsCounter：使用AtomicLong类型对删除的事件进行计数，每当日志打印了droppedEventsCounter后，会将droppedEventsCounter重置为0；
- lastReportTimestamp：用于记录最后一次日志打印droppedEventsCounter的时间戳；
- processingEvent：用来标记当前正有事件被listenerThread线程处理；
- logDroppedEvent：AtomicBoolean类型的变量，用于标记是否由于eventQueue已满，导致新的事件被删除；
- eventLock：用于当有新的事件到来时释放信号量，当对事件进行处理时获取信号量；
- listeners：继承自LiveListenerBus的监听器数组；
- listenerThread：处理事件的线程。
异步事件处理线程
listenerThread用于异步处理eventQueue中的事件，为了便于说明，这里将展示listenerThread及LiveListenerBus中的主要代码片段，见代码清单3。
代码清单3 LiveListenerBus主要逻辑的代码片段
private lazy val EVENT_QUEUE_CAPACITY = validateAndGetQueueSize()
private lazy val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY) private def validateAndGetQueueSize(): Int = {
val queueSize = sparkContext.conf.get(LISTENER_BUS_EVENT_QUEUE_SIZE)
if (queueSize <= 0) {
throw new SparkException("spark.scheduler.listenerbus.eventqueue.size must be > 0!")
}
queueSize
} private val started = new AtomicBoolean(false)
private val stopped = new AtomicBoolean(false)
private val droppedEventsCounter = new AtomicLong(0L)
@volatile private var lastReportTimestamp = 0L
private var processingEvent = false
private val logDroppedEvent = new AtomicBoolean(false)
private val eventLock = new Semaphore(0) private val listenerThread = new Thread(name) {
setDaemon(true)
override def run(): Unit = Utils.tryOrStopSparkContext(sparkContext) {
LiveListenerBus.withinListenerThread.withValue(true) {
while (true) {
eventLock.acquire() // 获取信号量
self.synchronized {
processingEvent = true
}
try {
val event = eventQueue.poll //从eventQueue中获取事件
if (event == null) {
// Get out of the while loop and shutdown the daemon thread
if (!stopped.get) {
throw new IllegalStateException("Polling `null` from eventQueue means" +
" the listener bus has been stopped. So `stopped` must be true")
}
return
}
postToAll(event) // 事件处理
} finally {
self.synchronized {
processingEvent = false
}
}
}
}
}
}
通过分析代码清单3，listenerThread的工作步骤为：
- 不断获取信号量（当可以获取信号量时，说明还有事件未处理）；
- 通过同步控制，将processingEvent设置为true；
- 从eventQueue中获取事件；
- 调用超类ListenerBus的postToAll方法（postToAll方法对监听器进行遍历，并调用SparkListenerBus的doPostEvent方法对事件进行匹配后执行监听器的相应方法）；
- 每次循环结束依然需要通过同步控制，将processingEvent设置为false；
值得一提的是，listenerThread的run方法中调用了Utils的tryOrStopSparkContext，tryOrStopSparkContext方法可以保证当listenerThread的内部循环抛出异常后启动一个新的线程停止SparkContext（SparkContext的内容将在第4章详细介绍，tryOrStopSparkContext方法的具体实现请阅读Utils工具类的实现）。
LiveListenerBus的消息投递
在解释了异步线程listenerThread的工作内容后，还有一个要点没有解释：eventQueue中的事件是如何放进去的呢？由于eventQueue定义在LiveListenerBus中，因此ListenerBus和SparkListenerBus中并没有操纵eventQueue的方法，要将事件放入eventQueue只能依靠LiveListenerBus自己了，其post方法就是为此目的而生的，见代码清单4。
代码清单4 向LiveListenerBus投递SparkListenerEvent事件
def post(event: SparkListenerEvent): Unit = {
if (stopped.get) {
logError(s"$name has already stopped! Dropping event $event")
return
}
val eventAdded = eventQueue.offer(event) // 向eventQueue中添加事件
if (eventAdded) {
eventLock.release()
} else {
onDropEvent(event)
droppedEventsCounter.incrementAndGet()
}
// 打印删除事件数的日志
val droppedEvents = droppedEventsCounter.get
if (droppedEvents > 0) {
if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) {
if (droppedEventsCounter.compareAndSet(droppedEvents, 0)) {
val prevLastReportTimestamp = lastReportTimestamp
lastReportTimestamp = System.currentTimeMillis()
logWarning(s"Dropped $droppedEvents SparkListenerEvents since " +
new java.util.Date(prevLastReportTimestamp))
}
}
}
}
从代码清单4看到post方法的处理步骤如下：
- 判断LiveListenerBus是否已经处于停止状态；
- 向eventQueue中添加事件。如果添加成功，则释放信号量进而催化listenerThread能够有效工作。如果eventQueue已满造成添加失败，则移除事件，并对删除事件计数器droppedEventsCounter进行自增；
- 如果有事件被删除，并且当前系统时间距离上一次打印droppedEventsCounter超过了60秒则将droppedEventsCounter打印到日志。
LiveListenerBus与监听器
与LiveListenerBus配合使用的监听器，并非是父类SparkListenerBus的类型参数SparkListenerInterface，而是继承自SparkListenerInterface的SparkListener及其子类。图2列出了Spark中监听器SparkListener以及它的6种最常用的实现[3]。
图2 SparkListener的类继承体系
SparkListener虽然实现了SparkListenerInterface中的每个方法，但是其实都是空实现，具体的实现需要交给子类去完成。
本文首先对事件总线的接口定义进行了一些介绍，之后选择ListenerBus的子类SparkListenerBus与LiveListenerBus作为具体的实现例子进行分析，最后本文选择LiveListenerBus作为具体的实现例子进行分析，这里将通过图3更加直观的展示ListenerBus、SparkListenerBus及LiveListenerBus的工作原理。
图3 LiveListenerBus的工作流程图
最后对于图3作一些补充说明：图中的DAGScheduler、SparkContext、BlockManagerMasterEndpoint、DriverEndpoint及LocalSchedulerBackend都是LiveListenerBus的事件来源，它们都是通过调用LiveListenerBus的post方法将消息交给异步线程listenerThread处理的。
[3] 除了本节列出的的六种SparkListener的子类外，还有很多其他的子类，这里就不一一列出了，感兴趣的读者可以查阅Spark相关文档或阅读源码知晓。
关于《Spark内核设计的艺术 架构设计与实现》
经过近一年的准备，基于Spark2.1.0版本的《Spark内核设计的艺术 架构设计与实现》一书现已出版发行，图书如图：
纸质版售卖链接如下：
京东：https://item.jd.com/12302500.html
最新文章
- iOS9支付宝无法调起客户端
- [C#] 走进异步编程的世界 - 在 GUI 中执行异步操作
- Android开发环境搭建
- 【requireJS源码学习01】了解整个requireJS的结构
- phonegap android3.5.1 Crosswalk
- Lucene基础（四）-- 结合数据库使用
- [置顶] 使用U盘安装ubuntu系统
- hdu1754 基础线段树
- http://python3-cookbook.readthedocs.io/zh_CN/latest/c14/p01_testing_output_sent_to_stdout.html
- 极光推送CTO黄鑫：技术人员要建立自己的知识图谱
- CF Educational Codeforces Round 10 D. Nested Segments 离散化+树状数组
- System.arraycopy(src, srcPos, dest, destPos, length) 与 Arrays.copyOf(original, newLength)区别
- struts2.1.6教程七、国际化
- Spring Boot – 自定义PropertyEditor
- 如何使用C语言的面向对象
- Android开发之使用Handler刷新UI控件
- LeetCode: Best Time to Buy and Sell Stock II 解题报告
- hihocoder博弈游戏·Nim游戏·三
- ElasticSearch 索引模块——全文检索
- 提取html里面url()导入的图片正则表达式