[SPARK-24479][SS] Added config for registering streamingQueryListeners #21504

arunmahadevan · 2018-06-07T01:10:41Z

What changes were proposed in this pull request?

Currently a "StreamingQueryListener" can only be registered programatically. We could have a new config "spark.sql.streamingQueryListeners" similar to "spark.sql.queryExecutionListeners" and "spark.extraListeners" for users to register custom streaming listeners.

How was this patch tested?

New unit test and running example programs.

Please review http://spark.apache.org/contributing.html before opening a pull request.

arunmahadevan · 2018-06-07T01:11:25Z

ping @tdas @jose-torres @HeartSaVioR

HyukjinKwon · 2018-06-07T01:21:17Z

Mind fixing the PR title to [SPARK-24479][SS] Added config for registering streamingQueryListeners?

HyukjinKwon · 2018-06-07T01:21:50Z

sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala

  private var lastTerminatedQuery: StreamingQuery = null

+  sparkSession.sparkContext.conf.get(STREAMING_QUERY_LISTENERS)
+  .foreach { classNames =>


nit:

sparkSession.sparkContext.conf.get(STREAMING_QUERY_LISTENERS).foreach { classNames => ... }

merlintang

LGTM +1

HyukjinKwon · 2018-06-07T01:26:51Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenersConfSuite.scala

+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.execution.streaming._
+import org.apache.spark.sql.streaming.StreamingQueryListener.{QueryStartedEvent, QueryTerminatedEvent, _}


nit: import looks a bit odd {QueryStartedEvent, QueryTerminatedEvent, _}.

SparkQA · 2018-06-07T05:05:35Z

Test build #91510 has finished for PR 21504 at commit d3a3baa.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
.doc(\"List of class names implementing StreamingQueryListener that will be automatically \" +

arunmahadevan · 2018-06-07T17:45:18Z

@HyukjinKwon , thanks for reviewing. Addressed comments.

SparkQA · 2018-06-07T21:44:06Z

Test build #91532 has finished for PR 21504 at commit f721ebe.

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

merlintang · 2018-06-07T22:51:02Z

sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala

+    Utils.loadExtensions(classOf[StreamingQueryListener], classNames,
+      sparkSession.sparkContext.conf).foreach(addListener)
+  }
+


two comments here:

we need to log the registration information here

we need to use try catch for this, it is possible that register fail. this would break the job.

Good point. Addressed, please check.

HyukjinKwon

Seems fine.

HyukjinKwon · 2018-06-08T02:11:34Z

sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala

+      Utils.loadExtensions(classOf[StreamingQueryListener], classNames,
+        sparkSession.sparkContext.conf).foreach(listener => {
+        addListener(listener)
+        logInfo(s"Registered listener ${listener.getClass.getName}")


I would do this at debug level ..

Either debug or info is fine for me, since it would add just couple of log lines only once.

Since its only once and provides information to user I guess info is fine. Similar pattern here https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/SparkContext.scala#L2359

HyukjinKwon · 2018-06-08T02:12:36Z

sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala

+    }
+  } catch {
+    case e: Exception =>
+      throw new SparkException(s"Exception when registering StreamingQueryListener", e)


nit: s seems not needed.

HeartSaVioR

LGTM except comments from @HyukjinKwon.

SparkQA · 2018-06-08T03:55:40Z

Test build #91540 has finished for PR 21504 at commit 02b2973.

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

HyukjinKwon

LGTM

SparkQA · 2018-06-08T19:02:16Z

Test build #91570 has finished for PR 21504 at commit 421e16b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-06-08T19:07:13Z

retest this please

SparkQA · 2018-06-08T21:05:00Z

Test build #91577 has finished for PR 21504 at commit 421e16b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2018-06-08T22:46:26Z

retest this, please

SparkQA · 2018-06-09T02:12:26Z

Test build #91585 has finished for PR 21504 at commit 421e16b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2018-06-09T11:39:13Z

Test failures were from kafka.

retest this, please

HyukjinKwon · 2018-06-11T02:25:12Z

retest this please

SparkQA · 2018-06-11T06:20:17Z

Test build #91645 has finished for PR 21504 at commit 421e16b.

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

HyukjinKwon · 2018-06-12T10:29:01Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

    .toSequence
    .createOptional

+  val STREAMING_QUERY_LISTENERS = buildStaticConf("spark.sql.streamingQueryListeners")


maybe -> spark.sql.streaming.streamingQueryListeners for consistency.

ok makes sense. renamed.

arunmahadevan · 2018-06-12T17:06:35Z

@HyukjinKwon , addressed comments. Can you take it forward?

HyukjinKwon · 2018-06-12T17:16:12Z

Looks fine but let me leave this open for more days before merging it in case someone has some comments on it.

SparkQA · 2018-06-12T19:47:04Z

Test build #91715 has finished for PR 21504 at commit b87c90b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-06-13T01:12:46Z

retest this please

SparkQA · 2018-06-13T03:47:42Z

Test build #91737 has finished for PR 21504 at commit b87c90b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-06-13T03:56:17Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenersConfSuite.scala

+
+
+  override protected def sparkConf: SparkConf =
+    super.sparkConf.set("spark.sql.streamingQueryListeners",


seems a mistake here.

Change-Id: I030ef8a27ec2962170981f2d1b4ed16165e2aef9

jose-torres · 2018-06-13T04:27:38Z

(Don't block on me - I won't have time to review in detail unless needed, but broadly the PR looks fine)

SparkQA · 2018-06-13T07:05:02Z

Test build #91741 has finished for PR 21504 at commit bbec6ce.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-06-13T08:54:12Z

retest this please

SparkQA · 2018-06-13T12:40:26Z

Test build #91761 has finished for PR 21504 at commit bbec6ce.

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

HyukjinKwon · 2018-06-13T12:42:47Z

Merged to master.

SPARK-24480: Added config for registering streamingQueryListeners

d3a3baa

arunmahadevan changed the title ~~SPARK-24480: Added config for registering streamingQueryListeners~~ SPARK-24479: Added config for registering streamingQueryListeners Jun 7, 2018

HyukjinKwon reviewed Jun 7, 2018

View reviewed changes

merlintang reviewed Jun 7, 2018

View reviewed changes

HyukjinKwon reviewed Jun 7, 2018

View reviewed changes

SPARK-24480: Addressed review comments

f721ebe

arunmahadevan changed the title ~~SPARK-24479: Added config for registering streamingQueryListeners~~ [SPARK-24479][SS] Added config for registering streamingQueryListeners Jun 7, 2018

merlintang reviewed Jun 7, 2018

View reviewed changes

SPARK-24480: added logging and exception handling

02b2973

HyukjinKwon reviewed Jun 8, 2018

View reviewed changes

HeartSaVioR approved these changes Jun 8, 2018

View reviewed changes

HyukjinKwon approved these changes Jun 8, 2018

View reviewed changes

reformat the exception string

421e16b

HyukjinKwon reviewed Jun 12, 2018

View reviewed changes

rename config parameter

b87c90b

HyukjinKwon reviewed Jun 13, 2018

View reviewed changes

fix unit tests

bbec6ce

Change-Id: I030ef8a27ec2962170981f2d1b4ed16165e2aef9

asfgit closed this in 7703b46 Jun 13, 2018



		override protected def sparkConf: SparkConf =
		super.sparkConf.set("spark.sql.streamingQueryListeners",

[SPARK-24479][SS] Added config for registering streamingQueryListeners #21504

[SPARK-24479][SS] Added config for registering streamingQueryListeners #21504

Uh oh!

Conversation

arunmahadevan commented Jun 7, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

arunmahadevan commented Jun 7, 2018

Uh oh!

HyukjinKwon commented Jun 7, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

merlintang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 7, 2018

Uh oh!

arunmahadevan commented Jun 7, 2018

Uh oh!

SparkQA commented Jun 7, 2018

Uh oh!

merlintang Jun 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 8, 2018

Uh oh!

HyukjinKwon left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 8, 2018

Uh oh!

HyukjinKwon commented Jun 8, 2018

Uh oh!

SparkQA commented Jun 8, 2018

Uh oh!

HeartSaVioR commented Jun 8, 2018

Uh oh!

SparkQA commented Jun 9, 2018

Uh oh!

HeartSaVioR commented Jun 9, 2018

Uh oh!

HyukjinKwon commented Jun 11, 2018

Uh oh!

SparkQA commented Jun 11, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arunmahadevan commented Jun 12, 2018

Uh oh!

HyukjinKwon commented Jun 12, 2018

Uh oh!

SparkQA commented Jun 12, 2018

Uh oh!

HyukjinKwon commented Jun 13, 2018

Uh oh!

merlintang Jun 7, 2018 •

edited

Loading