[SPARK-18670][SS]Limit the number of StreamingQueryListener.StreamProgressEvent when there is no data #16108

zsxwing · 2016-12-01T21:31:36Z

What changes were proposed in this pull request?

This PR adds a sql conf spark.sql.streaming.noDataReportInterval to control how long to wait before outputing the next StreamProgressEvent when there is no data.

How was this patch tested?

The added unit test.

…here is no data

zsxwing · 2016-12-01T21:31:51Z

cc @tdas

tdas · 2016-12-01T22:02:08Z

sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -603,6 +603,13 @@ object SQLConf {
      .timeConf(TimeUnit.MILLISECONDS)
      .createWithDefault(10L)

+  val STREAMING_NO_DATA_REPORT_INTERVAL =


Report is not a term used anywhere. How about streamingNoDataEventInterval

tdas · 2016-12-01T22:02:21Z

sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

+  val STREAMING_NO_DATA_REPORT_INTERVAL =
+    SQLConfigBuilder("spark.sql.streaming.noDataReportInterval")
+      .internal()
+      .doc("How long to wait between two progress events when there is no data")


... (in ms) ...

TimeConf supports various units like 10s, 100ms. The unit here is telling which unit we want to use when getting this conf.

tdas · 2016-12-01T22:06:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

+            if (dataAvailable) {
+              postEvent(new QueryProgressEvent(lastProgress))
+            } else {
+              val now = triggerClock.getTimeMillis()


Shouldnt there be an event every time the system switches from dataAvailable = true, to dataAvailable = false. This logic wont do that right?

tdas · 2016-12-01T22:08:59Z

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

+    withSQLConf(SQLConf.STREAMING_NO_DATA_REPORT_INTERVAL.key -> "100ms") {
+      @volatile var progressEventCount = 0
+
+      val listener = new StreamingQueryListener {


can you write a short explanation on what this test does. it hard to understand.

tdas · 2016-12-01T22:09:56Z

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

@@ -191,6 +192,48 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter {
    assert(queryQueryTerminated.exception === newQueryTerminated.exception)
  }

+  test("noDataReportInterval") {


nit: make the name better - only one progressevent per interval when no data

tdas

I think the test can be simplified.

tdas · 2016-12-01T23:44:49Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -196,6 +198,9 @@ class StreamExecution(
      // While active, repeatedly attempt to run batches.
      SparkSession.setActiveSession(sparkSession)

+      // The timestamp we report an event that has no input data
+      var noDataEventTimestamp = Long.MinValue


oh snap.... i didnt realize... this now more confusing with event time and watermark stuff. sorry! may be rename to lastNoDataProgressEventTime?

tdas · 2016-12-01T23:45:08Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -63,6 +63,8 @@ class StreamExecution(

  private val pollingDelayMs = sparkSession.sessionState.conf.streamingPollingDelay

+  private val noDataEventInterval = sparkSession.sessionState.conf.streamingNoDataEventInterval


similar to below, noDataProgressEventInterval?

tdas · 2016-12-01T23:45:30Z

sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -603,6 +603,13 @@ object SQLConf {
      .timeConf(TimeUnit.MILLISECONDS)
      .createWithDefault(10L)

+  val STREAMING_NO_DATA_EVENT_INTERVAL =
+    SQLConfigBuilder("spark.sql.streaming.noDataEventInterval")


noDataProgressEventInterval

tdas · 2016-12-01T23:47:16Z

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

+          }
+        }
+        testStream(MemoryStream[Int].toDS)(actions: _*)
+        // 11 is the max value of the possible numbers of events.


why 11? explanation?

I think this test looks a little complicated. Also you are assuming the fact the when there is no data, it will sleep for 10 ms real time. why not something like this.

val input = new MemoryStream() { @volatile var numTriggers = 0 override def getOffset(): Offset = { numTriggers +=1 ; super.getOffset() } } ... // add listener with progressEventCount and flag isTerminationEventPosted ... testStream(input.toDF) { // on real clock AddData(...) AssertOnQuery { q => eventually (timeout(streamingTimeout)) { assert(input.numTriggers > 100) // at least 100 triggers have occurred } }, StopStream, AssertOnQuery { q => eventually(timeout(streamingTimeout) { assert(isTerminationEventPosted) // event queue has drained assert(numProgressEvent >=1 && numProgressEvent <= 100) // no more than 10 events. } } }

Okay, that wont work as well, because there is not guarantee how long the 100 triggers would take. It may take 100 seconds in a slow machine, in which numProgress events could legitimately have 100 events.

Nonetheless manual clock tests can be improved.

tdas · 2016-12-02T00:00:08Z

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

+          actions += AssertOnQuery { _ =>
+            // Sleep so that if the config `noDataEventInterval` doesn't work, it has enough time
+            // to report too many events.
+            Thread.sleep(10)


this combination of manual clock and real time sleep is quite confusing

SparkQA · 2016-12-02T00:41:19Z

Test build #69505 has finished for PR 16108 at commit 4ace2bd.

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

SparkQA · 2016-12-02T01:46:42Z

Test build #69512 has finished for PR 16108 at commit 3425231.

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

tdas

Test looks good now. But there are other comments that need to be addressed. The name is still confusing, in a different way now. I should have thought of that before suggesting.

tdas · 2016-12-02T03:20:36Z

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

+        testStream(input.toDS)(actions: _*)
+        spark.sparkContext.listenerBus.waitUntilEmpty(10000)
+        // 11 is the max value of the possible numbers of events.
+        assert(numProgressEvent >= 1 && numProgressEvent <= 11)


Shouldnt this be > 1? What if there is a bug that it only outputs once when isData goes from true -> false, and never outputs again, irrespective of the noDataInterval?

SparkQA · 2016-12-02T04:22:08Z

Test build #69526 has finished for PR 16108 at commit f7b8755.

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

SparkQA · 2016-12-02T07:41:08Z

Test build #69545 has finished for PR 16108 at commit be3737f.

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

tdas · 2016-12-02T20:41:30Z

LGTM. Merging to master and 2.1

…ogressEvent when there is no data ## What changes were proposed in this pull request? This PR adds a sql conf `spark.sql.streaming.noDataReportInterval` to control how long to wait before outputing the next StreamProgressEvent when there is no data. ## How was this patch tested? The added unit test. Author: Shixiong Zhu <shixiong@databricks.com> Closes #16108 from zsxwing/SPARK-18670. (cherry picked from commit 56a503d) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

…ogressEvent when there is no data ## What changes were proposed in this pull request? This PR adds a sql conf `spark.sql.streaming.noDataReportInterval` to control how long to wait before outputing the next StreamProgressEvent when there is no data. ## How was this patch tested? The added unit test. Author: Shixiong Zhu <shixiong@databricks.com> Closes apache#16108 from zsxwing/SPARK-18670.

Limit the number of StreamingQueryListener.StreamProgressEvent when t…

4ace2bd

…here is no data

tdas reviewed Dec 1, 2016

View reviewed changes

Address TD's comments

3425231

tdas reviewed Dec 2, 2016

View reviewed changes

Address

f7b8755

tdas reviewed Dec 2, 2016

View reviewed changes

Address more

be3737f

asfgit closed this in 56a503d Dec 2, 2016

zsxwing deleted the SPARK-18670 branch December 2, 2016 21:28

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-18670][SS]Limit the number of StreamingQueryListener.StreamProgressEvent when there is no data #16108

[SPARK-18670][SS]Limit the number of StreamingQueryListener.StreamProgressEvent when there is no data #16108

zsxwing commented Dec 1, 2016 •

edited

Loading

zsxwing commented Dec 1, 2016

tdas Dec 1, 2016

tdas Dec 1, 2016

zsxwing Dec 1, 2016

tdas Dec 1, 2016

tdas Dec 1, 2016

tdas Dec 1, 2016

tdas left a comment

tdas Dec 1, 2016

tdas Dec 1, 2016

tdas Dec 1, 2016

tdas Dec 1, 2016

tdas Dec 2, 2016 •

edited

Loading

tdas Dec 2, 2016 •

edited

Loading

tdas Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

tdas left a comment

tdas Dec 2, 2016

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

tdas commented Dec 2, 2016

		@@ -63,6 +63,8 @@ class StreamExecution(

		private val pollingDelayMs = sparkSession.sessionState.conf.streamingPollingDelay

		private val noDataEventInterval = sparkSession.sessionState.conf.streamingNoDataEventInterval

[SPARK-18670][SS]Limit the number of StreamingQueryListener.StreamProgressEvent when there is no data #16108

[SPARK-18670][SS]Limit the number of StreamingQueryListener.StreamProgressEvent when there is no data #16108

Conversation

zsxwing commented Dec 1, 2016 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

zsxwing commented Dec 1, 2016

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

tdas left a 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

tdas Dec 2, 2016 • edited Loading

Choose a reason for hiding this comment

tdas Dec 2, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

tdas left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 2, 2016

SparkQA commented Dec 2, 2016

tdas commented Dec 2, 2016

zsxwing commented Dec 1, 2016 •

edited

Loading

tdas Dec 2, 2016 •

edited

Loading

tdas Dec 2, 2016 •

edited

Loading