[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

      .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

    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

      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


  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

      .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 <[email protected]> Closes #16108 from zsxwing/SPARK-18670. (cherry picked from commit 56a503d) Signed-off-by: Tathagata Das <[email protected]>

…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 <[email protected]> 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


		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

Uh oh!

Conversation

zsxwing commented Dec 1, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

zsxwing commented Dec 1, 2016

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!

Choose a reason for hiding this comment

Uh oh!

tdas 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!

tdas Dec 2, 2016 • 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!

tdas Dec 2, 2016 • 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!

SparkQA commented Dec 2, 2016

Uh oh!

SparkQA commented Dec 2, 2016

Uh oh!

tdas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 2, 2016

Uh oh!

SparkQA commented Dec 2, 2016

Uh oh!

tdas commented Dec 2, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

zsxwing commented Dec 1, 2016 •

edited

Loading

tdas Dec 2, 2016 •

edited

Loading

tdas Dec 2, 2016 •

edited

Loading