[SPARK-18838][core] Add separate listener queues to LiveListenerBus. #19211

vanzin · 2017-09-12T21:31:50Z

This change modifies the live listener bus so that all listeners are
added to queues; each queue has its own thread to dispatch events,
making it possible to separate slow listeners from other more
performance-sensitive ones.

The public API has not changed - all listeners added with the existing
"addListener" method, which after this change mostly means all
user-defined listeners, end up in a default queue. Internally, there's
an API allowing listeners to be added to specific queues, and that API
is used to separate the internal Spark listeners into 3 categories:
application status listeners (e.g. UI), executor management (e.g. dynamic
allocation), and the event log.

The queueing logic, while abstracted away in a separate class, is kept
as much as possible hidden away from consumers. Aside from choosing their
queue, there's no code change needed to take advantage of queues.

Test coverage relies on existing tests; a few tests had to be tweaked
because they relied on LiveListenerBus.postToAll being synchronous,
and the change makes that method asynchronous. Other tests were simplified
not to use the asynchronous LiveListenerBus.

This change modifies the live listener bus so that all listeners are added to queues; each queue has its own thread to dispatch events, making it possible to separate slow listeners from other more performance-sensitive ones. The public API has not changed - all listeners added with the existing "addListener" method, which after this change mostly means all user-defined listeners, end up in a default queue. Internally, there's an API allowing listeners to be added to specific queues, and that API is used to separate the internal Spark listeners into 3 categories: application status listeners (e.g. UI), executor management (e.g. dynamic allocation), and the event log. The queueing logic, while abstracted away in a separate class, is kept as much as possible hidden away from consumers. Aside from choosing their queue, there's no code change needed to take advantage of queues. Metrics were also simplified a little bit; the live bus now keeps track of metrics per queue instead of individual listeners. This is mostly to make integration with the existing metrics code in `ListenerBus` easier, without having to refactor the code; that can be done later if queue-level metrics are not enough. Test coverage relies on existing tests; a few tests had to be tweaked because they relied on `LiveListenerBus.postToAll` being synchronous, and the change makes that method asynchronous. Other tests were simplified not to use the asynchronous LiveListenerBus.

SparkQA · 2017-09-13T00:40:44Z

Test build #81693 has finished for PR 19211 at commit 4a6858f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Also fix a problem where the queue size metric might be out of date.

This change makes the event queue implement SparkListenerBus and inherit all the metrics and dispatching behavior, making the change easier on the scheduler and also restoring per-listener metrics.

SparkQA · 2017-09-13T16:17:31Z

Test build #81724 has started for PR 19211 at commit 2915a5e.

SparkQA · 2017-09-13T16:32:38Z

Test build #81725 has started for PR 19211 at commit cf5c6ce.

SparkQA · 2017-09-13T16:47:30Z

Test build #81726 has started for PR 19211 at commit 24f5c8d.

vanzin · 2017-09-13T17:15:01Z

(Nevermind the test failures, I killed the obsolete builds.)

vanzin · 2017-09-13T17:16:15Z

retest this please

SparkQA · 2017-09-13T20:28:06Z

Test build #81728 has finished for PR 19211 at commit ad6ff49.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-09-14T04:17:57Z

Test build #81744 has finished for PR 19211 at commit 20b8382.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2017-09-14T18:17:53Z

I think this is ready now, pinging some people: @squito @JoshRosen @cloud-fan

SparkQA · 2017-09-14T21:27:49Z

Test build #81795 has finished for PR 19211 at commit 1d43d3a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

ajbozarth · 2017-09-14T23:42:14Z

I did a read-through of your code and didn't find any glaring issues, but I'd like to go through it a bit more before passing judgment. Overall a nice new feature though.

cloud-fan · 2017-09-15T15:31:49Z

core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala

@@ -217,7 +217,7 @@ private[spark] class ExecutorAllocationManager(
   * the scheduling task.
   */
  def start(): Unit = {
-    listenerBus.addListener(listener)
+    listenerBus.addToQueue(listener, LiveListenerBus.EXECUTOR_MGMT_QUEUE)


will we have more queues in the future? If it's not too many, maybe we can use different methods, like addToMGMTQueue

cloud-fan · 2017-09-15T15:58:16Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

+   * Start an asynchronous thread to dispatch events to the underlying listeners.
+   *
+   * @param sc Used to stop the SparkContext in case the a listener fails.
+   * @param metrics Used to report listener performance metrics.


there is no metrics param.

cloud-fan · 2017-09-15T16:01:34Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

-              processingEvent = false
-            }
-          }
+  private val queues = new CopyOnWriteArrayList[AsyncEventQueue]()


can we have something like

private val defaultQueue = ... private val executorManagementQueue = ... ...

I think the number of queues will be small.

I can add specific methods for each internal queue; but I'd like to keep the internal management of queues more generic. One of the ideas in #18253 was to allow filtering of which events are enqueued at all (e.g. don't enqueue SparkListenerBlockUpdated because it's not written to the event logs, reducing the load on the event log queue). Leaving the internal management more generic would allow that to be more easily / cleanly implemented later (instead of "addToEventLogQueue", you'd have a "addCustomQueue" method with a subclass of AsyncEventQueue).

SparkQA · 2017-09-15T20:31:45Z

Test build #81826 has finished for PR 19211 at commit ed71477.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-09-18T06:14:19Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

+      case None =>
+        val newQueue = new AsyncEventQueue(queue, conf, metrics)
+        newQueue.addListener(listener)
+        if (started.get() && !stopped.get()) {


stopped.get is always false here.

cloud-fan · 2017-09-18T06:14:52Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

+   */
+  private def addToQueue(listener: SparkListenerInterface, queue: String): Unit = synchronized {
+    if (stopped.get()) {
+      throw new IllegalStateException("LiveListenerBus is stopped.")


shall we also check !started.get?

No. It's ok and actually expected to add listeners before the bus is started.

cloud-fan · 2017-09-18T06:20:00Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

+
+  /** Post an event to all queues. */
+  def post(event: SparkListenerEvent): Unit = {
+    if (!stopped.get()) {


shall we add an assert that this bus is started?

No, as the class javadoc says, it's ok to send events before the bus is started.

SparkQA · 2017-09-18T18:53:52Z

Test build #81884 has finished for PR 19211 at commit 77dd8ec.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tgravescs · 2017-09-18T19:59:53Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

-private[spark] class LiveListenerBus(conf: SparkConf) extends SparkListenerBus {
-
-  self =>
+private[spark] class LiveListenerBus(conf: SparkConf) {

  import LiveListenerBus._

  private var sparkContext: SparkContext = _


maybe I missed it but I don't see this being set anywhere now

Ah, this should be set in start(), probably deleted in one of the code moves.

SparkQA · 2017-09-18T23:35:45Z

Test build #81895 has finished for PR 19211 at commit 35d3428.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-09-19T04:04:16Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

+    queues.asScala.flatMap { queue => queue.findListenersByClass[T]() }
+  }
+
+  private[spark] def listeners: JList[SparkListenerInterface] = {


where do we use this method?

Used by tests.

cloud-fan · 2017-09-19T04:04:38Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

  }
+
+  private[spark] def findListenersByClass[T <: SparkListenerInterface : ClassTag](): Seq[T] = {


let's add a comment to say it testing only.

cloud-fan · 2017-09-19T04:05:09Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

+
+  private[scheduler] val APP_STATUS_QUEUE = "appStatus"
+
+  private[scheduler] val EXECUTOR_MGMT_QUEUE = "executorMgmt"


MGMT is hard to understand, how about MANAGEMENT?

That's a pretty common abbreviation for "management", but sure.

cloud-fan · 2017-09-19T04:16:45Z

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala


    // After listener bus has stopped, posting events should not increment counter
    bus.stop()
    (1 to 5).foreach { _ => bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded)) }
    assert(counter.count === 5)
-    assert(bus.metrics.numEventsPosted.getCount === 5)
-
-    // Make sure per-listener-class timers were created:


do we still have this?

oh seems we don't need it anymore.

cloud-fan · 2017-09-19T04:20:44Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

@@ -575,8 +575,6 @@ class StreamingContextSuite extends SparkFunSuite with BeforeAndAfter with TimeL

  test("getActive and getActiveOrCreate") {
    require(StreamingContext.getActive().isEmpty, "context exists from before")
-    sc = new SparkContext(conf)


why this change?

Because the test stops the underlying streaming context and thus the spark context, and you can't add listeners to the bus after that. Thus, it was failing.

cloud-fan · 2017-09-19T04:21:45Z

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

-private[spark] class LiveListenerBus(conf: SparkConf) extends SparkListenerBus {
-
-  self =>
+private[spark] class LiveListenerBus(conf: SparkConf) {


is it still valid to call LiveListenerBus.postToAll? If not, shall we overwrite postToAll and throw exception?

also LiveListenerBus.addListener seems not valid too.

LiveListenerBus.postToAll doesn't exist anymore, and neither does LiveListenerBus.addListener. But I'm not sure what you're asking.

oh sorry I missed that, LiveListenerBus doesn't extend SparkListenerBus anymore.

cloud-fan · 2017-09-19T04:24:19Z

LGTM except some minor comments

jerryshao · 2017-09-19T06:32:38Z

core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala

+      this.sc = sc
+      dispatchThread.start()
+    } else {
+      throw new IllegalStateException(s"$name already started!")


Is it better to be a no-op here, or a warning log instead?

seems fine to me -- its only called in LiveListenerBus, where we guarantee this is true. seems better to fail-fast if its messed up

squito · 2017-09-19T19:12:30Z

lgtm

do you know how much better this makes it? Eg., if we had an existing case where things go haywire in dynamic allocation because of this -- we could see if after this change, the only dropped events are in eventLog, hopefully. Seems like the right change regardless, but it would be good to know the effect.

vanzin · 2017-09-19T19:35:54Z

I don't really have a readily available cluster that can easily put that much pressure on the listener bus; I tried with the one I have and monitored the listener bus metrics, and found pretty much what I expected (slightly more latency in the event log queue, but not enough in my tests to actually cause event drops).

SparkQA · 2017-09-19T19:56:55Z

Test build #81940 has finished for PR 19211 at commit 283b733.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-09-20T05:42:05Z

thanks, merging to master!

dongjoon-hyun · 2017-09-20T21:30:44Z

Hi, @vanzin and @cloud-fan .
This seems to break the master branch. Could you take a look?

https://amplab.cs.berkeley.edu/jenkins/view/Spark%20QA%20Test%20(Dashboard)/job/spark-master-test-sbt-hadoop-2.7/3508/

vanzin · 2017-09-20T21:50:03Z

Following assert is failing in some cases, let me fix it and send a follow up PR:

    val listeners = sc.listenerBus.findListenersByClass[SaveExecutorInfo]
    assert(listeners.size === 1)

This change modifies the live listener bus so that all listeners are added to queues; each queue has its own thread to dispatch events, making it possible to separate slow listeners from other more performance-sensitive ones. The public API has not changed - all listeners added with the existing "addListener" method, which after this change mostly means all user-defined listeners, end up in a default queue. Internally, there's an API allowing listeners to be added to specific queues, and that API is used to separate the internal Spark listeners into 3 categories: application status listeners (e.g. UI), executor management (e.g. dynamic allocation), and the event log. The queueing logic, while abstracted away in a separate class, is kept as much as possible hidden away from consumers. Aside from choosing their queue, there's no code change needed to take advantage of queues. Test coverage relies on existing tests; a few tests had to be tweaked because they relied on `LiveListenerBus.postToAll` being synchronous, and the change makes that method asynchronous. Other tests were simplified not to use the asynchronous LiveListenerBus. Author: Marcelo Vanzin <vanzin@cloudera.com> Closes apache#19211 from vanzin/SPARK-18838. (cherry picked from commit c6ff59a) Cloudera ID: CDH-51951

…veListenerBus. This change modifies the live listener bus so that all listeners are added to queues; each queue has its own thread to dispatch events, making it possible to separate slow listeners from other more performance-sensitive ones. The public API has not changed - all listeners added with the existing "addListener" method, which after this change mostly means all user-defined listeners, end up in a default queue. Internally, there's an API allowing listeners to be added to specific queues, and that API is used to separate the internal Spark listeners into 3 categories: application status listeners (e.g. UI), executor management (e.g. dynamic allocation), and the event log. The queueing logic, while abstracted away in a separate class, is kept as much as possible hidden away from consumers. Aside from choosing their queue, there's no code change needed to take advantage of queues. Test coverage relies on existing tests; a few tests had to be tweaked because they relied on `LiveListenerBus.postToAll` being synchronous, and the change makes that method asynchronous. Other tests were simplified not to use the asynchronous LiveListenerBus. Author: Marcelo Vanzin <vanzin@cloudera.com> Closes apache#19211 from vanzin/SPARK-18838. (cherry picked from commit c6ff59a) Cloudera ID: CDH-51951 (cherry picked from commit 6cb9e4d) (cherry picked from commit 8cbc8656344412f3b2db39367f7def9f6a03512a)

Marcelo Vanzin added 3 commits September 13, 2017 00:35

Better name for ListenerQueue.

9fc30ba

Cleanup metric initialization a bit.

6bee214

Also fix a problem where the queue size metric might be out of date.

Move dispatching to the event thread.

2915a5e

This change makes the event queue implement SparkListenerBus and inherit all the metrics and dispatching behavior, making the change easier on the scheduler and also restoring per-listener metrics.

Minor cleanup.

cf5c6ce

Undo obsolete changes.

24f5c8d

Fix a comment after changes.

ad6ff49

Marcelo Vanzin added 2 commits September 13, 2017 18:01

Undo unnecessary changes.

563be32

Make the event log queue name a constant too.

20b8382

Merge branch 'master' into SPARK-18838

1d43d3a

cloud-fan reviewed Sep 15, 2017

View reviewed changes

Marcelo Vanzin added 3 commits September 15, 2017 10:10

Explicit add methods for each queue.

324041f

Rename addListener() to addToSharedQueue().

3b218c2

Remove stale comment.

ed71477

cloud-fan reviewed Sep 18, 2017

View reviewed changes

Remove redundant check.

77dd8ec

tgravescs reviewed Sep 18, 2017

View reviewed changes

Keep reference to SparkContext.

35d3428

cloud-fan reviewed Sep 19, 2017

View reviewed changes

jerryshao reviewed Sep 19, 2017

View reviewed changes

Nits.

283b733

asfgit closed this in c6ff59a Sep 20, 2017

vanzin deleted the SPARK-18838 branch September 20, 2017 22:13

ashangit mentioned this pull request Jul 20, 2018

[SPARK-18838][CORE] Add separate listener queues to LiveListenerBus. criteo-forks/spark#75

Merged

HeartSaVioR mentioned this pull request May 22, 2019

[SPARK-27804] LiveListenerBus may create multiple AsyncEventQueues with same name in concurrent scene #24676

Closed

juliuszsompolski mentioned this pull request Nov 6, 2019

[SPARK-29724][SPARK-29726][WEBUI][SQL] Support JDBC/ODBC tab for HistoryServer WebUI #26378

Closed

		}

		private[spark] def findListenersByClass[T <: SparkListenerInterface : ClassTag](): Seq[T] = {


		private[scheduler] val APP_STATUS_QUEUE = "appStatus"

		private[scheduler] val EXECUTOR_MGMT_QUEUE = "executorMgmt"

[SPARK-18838][core] Add separate listener queues to LiveListenerBus. #19211

[SPARK-18838][core] Add separate listener queues to LiveListenerBus. #19211

Conversation

vanzin commented Sep 12, 2017 • edited Loading

SparkQA commented Sep 13, 2017

SparkQA commented Sep 13, 2017

SparkQA commented Sep 13, 2017

SparkQA commented Sep 13, 2017

vanzin commented Sep 13, 2017

vanzin commented Sep 13, 2017

SparkQA commented Sep 13, 2017

SparkQA commented Sep 14, 2017

vanzin commented Sep 14, 2017

SparkQA commented Sep 14, 2017

ajbozarth commented Sep 14, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Sep 15, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Sep 18, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Sep 18, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Sep 19, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

squito commented Sep 19, 2017

vanzin commented Sep 19, 2017

SparkQA commented Sep 19, 2017

cloud-fan commented Sep 20, 2017

dongjoon-hyun commented Sep 20, 2017 • edited Loading

vanzin commented Sep 20, 2017

vanzin commented Sep 12, 2017 •

edited

Loading

dongjoon-hyun commented Sep 20, 2017 •

edited

Loading