[SPARK-22017] Take minimum of all watermark execs in StreamExecution. #19239

jose-torres · 2017-09-14T21:53:50Z

What changes were proposed in this pull request?

Take the minimum of all watermark exec nodes as the "real" watermark in StreamExecution, rather than picking one arbitrarily.

How was this patch tested?

new unit test

tdas · 2017-09-14T22:21:43Z

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

    val runId: UUID,
    val currentBatchId: Long,
-    offsetSeqMetadata: OffsetSeqMetadata)
+    private[sql] val offsetSeqMetadata: OffsetSeqMetadata)


just make it val. Anything inside sql.execution does not show up in docs, and therefore we just keep to val so that we can debug when we need to dig deep.

tdas · 2017-09-14T22:23:12Z

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

+   * timestamp present here will be used as the overall query watermark in offsetSeqMetadata;
+   * the query watermark is what's logged and used to age out old state.
+   */
+  protected var attributeWatermarkMsMap: AttributeMap[Long] = AttributeMap(Seq())


this is needed only inside a single function right? so make it a local variable.
even better, if you dont have to make it a var, make it a val in the usual functional way.

This map has to persist and get updated across batches, and I'm not sure how to do that with a local variable or a val.

I see. I was mistaken. However in that case, use a mutable.HashMap instead of var. That the code style we use with scala is not to use vars unless absolutely needed.

tdas · 2017-09-14T22:24:09Z

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

    if (hasNewData) {
      var batchWatermarkMs = offsetSeqMetadata.batchWatermarkMs
-      // Update the eventTime watermark if we find one in the plan.
+      // Update the eventTime watermarks if we find any in the plan.


its still a single watermark that is being updated. it just happens to be updated using multiple watermarks.

Well, we're updating multiple watermarks in the map. We later update offsetSeqMetadata with the new minimum one, but that's not in this block.

Yeah. I had made this comment when i was thinking that we dont need the mutable map. Ignore this.

tdas · 2017-09-14T22:24:54Z

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

+        == watermark)
+    }
+
+    populateNewWatermarkFromData(first, 11)


if these are always used together, then these functions can be merged .. right?

Also, I think you can use the testStream..AddData... AssertOnQuery pattern. its cleaner.
https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala#L180

The problem is that watermark recalculation happens at the beginning of each batch, and to sequence executions I have to call CheckData or CheckLastBatch. So that method ends up producing a test multiple times longer, since a single entry is:

AddData(realData)
CheckLastBatch
AddData(0)
CheckLastBatch
AssertOnQuery

even better example - https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/streaming/EventTimeWatermarkSuite.scala#L107

You are right. what you have is better.

tdas · 2017-09-14T23:15:44Z

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

+
+    val second = MemoryStream[Int]
+
+    val secondAggregation = second.toDF()


there is no aggregation here.

tdas · 2017-09-14T23:37:05Z

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

+      union.processAllAvailable()
+    }
+
+    def assertQueryWatermark(watermark: Int): Unit = {


nit: assertWatermark.

tdas · 2017-09-14T23:38:22Z

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

+    }
+
+    def assertQueryWatermark(watermark: Int): Unit = {
+      assert(union.asInstanceOf[StreamingQueryWrapper].streamingQuery


this line break is hard to read. how about you break it with an intermediate variable (e.g. val lastExecution = ... ; assert(...)

tdas · 2017-09-14T23:40:49Z

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

+
+    val secondAggregation = second.toDF()
+      .withColumn("eventTime", $"value".cast("timestamp"))
+      .withWatermark("eventTime", "10 seconds")


Can you update the test to have different watermark delays, so that we test that we are choosing min delay, but the min watermark?

tdas · 2017-09-14T23:42:34Z

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

+    populateNewWatermarkFromData(first, 11)
+    assertQueryWatermark(1000)
+
+    // Watermark stays at 1 from the left when right watermark moves to 2


nit: Watermark to "global watermark" .. to differentiate from "right watermark" later in the sentence.

SparkQA · 2017-09-15T00:34:53Z

Test build #81797 has finished for PR 19239 at commit 6a4c80b.

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

jose-torres · 2017-09-15T02:05:42Z

addressed comments

SparkQA · 2017-09-15T04:44:57Z

Test build #81803 has finished for PR 19239 at commit 8b60538.

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

tdas · 2017-09-15T06:01:58Z

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

+  /**
+   * A map of current watermarks, keyed by the position of the watermark operator in the
+   * physical plan. The minimum watermark timestamp present here will be used and persisted as the
+   * query's watermark when preparing each batch, so it's ok that this val isn't fault-tolerant.


Make it clear that "this state is 'soft state' that does not affect the correctness and semantic guarantees of watermarks."

tdas · 2017-09-15T06:07:32Z

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

            logDebug(s"Observed event time stats: ${e.eventTimeStats.value}")
-            e.eventTimeStats.value.max - e.delayMs
-        }.headOption.foreach { newWatermarkMs =>
+            val newAttributeWatermarkMs = e.eventTimeStats.value.max - e.delayMs


Add the e or index in the log debug to differentiate between different operators.

also... newAttributeWatermarkMs -> newWatermarkMs
why "attribute"?

tdas · 2017-09-15T06:08:01Z

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

-            e.eventTimeStats.value.max - e.delayMs
-        }.headOption.foreach { newWatermarkMs =>
+            val newAttributeWatermarkMs = e.eventTimeStats.value.max - e.delayMs
+            val mappedWatermarkMs: Option[Long] = watermarkMsMap.get(index)


nit: Option[Long] is not needed.

mappedWatermarkMs -> previousWatermarkMs or prevWatermarkMs
more semantically meaningful.

tdas · 2017-09-15T06:59:45Z

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

+          case _ =>
+        }
+
+        // Update the query watermark to the minimum of all attribute watermarks.


Add a little bit more docs saying that this is the safe thing to do as watermark guarantees .....

tdas · 2017-09-15T07:07:57Z

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

+      assert(lastExecution.offsetSeqMetadata.batchWatermarkMs == watermark)
+    }
+
+    generateAndAssertNewWatermark(first, Seq(11), 1000)


instead of having two variations for handing single input and double inputs, you can do something like this.

def getWatermarkAfterData(firstData: Seq[Int] = Seq.empty, secondData: Seq[Int] = Seq.empty): Long = { if (firstData.nonEmpty) first.add(firstData) if (secondData.nonEmpty) second.add(secondData) union.processAllAvailable() // add a dummy batch so lastExecution has the new watermark first.addData(0) union.processAllAvailable() // get updated watermark val lastExecution = union.asInstanceOf[StreamingQueryWrapper].streamingQuery.lastExecution lastExecution.offsetSeqMetadata.batchWatermarkMs } assert(getWatermarkAfterData(firstData = Seq(...)) === 10000) assert(getWatermarkAfterData(secondData = Seq(...)) === 10000) assert(getWatermarkAfterData(firstData = Seq(...), secondData = Seq(...)) === 10000)

BTW, why is the first watermark at 1000?

SparkQA · 2017-09-15T10:55:34Z

Test build #81814 has finished for PR 19239 at commit 8b60538.

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

tdas · 2017-09-15T22:09:17Z

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

+    // Global watermark doesn't decrement with simultaneous data
+    assert(getWatermarkAfterData(firstData = Seq(100), secondData = Seq(100)) == 120000)
+    assert(getWatermarkAfterData(firstData = Seq(140), secondData = Seq(100)) == 120000)
+    assert(getWatermarkAfterData(firstData = Seq(100), secondData = Seq(135)) == 130000)


test recovery of the minimum after a restart.

SparkQA · 2017-09-15T23:44:49Z

Test build #81832 has finished for PR 19239 at commit 13affc7.

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

SparkQA · 2017-09-16T03:39:33Z

Test build #81837 has finished for PR 19239 at commit cdf4361.

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

tdas · 2017-09-16T04:17:35Z

thank you @Joseph-Torres merged to master.

jose-torres added 3 commits September 13, 2017 14:49

Implement multiple watermark StreamExecution support.

4a7f53f

partially fix test

9b9cd19

Finish rewriting test

6a4c80b

tdas reviewed Sep 14, 2017

View reviewed changes

make IncrementalExecution.offsetSeqMetadata non-private

484940e

tdas reviewed Sep 14, 2017

View reviewed changes

properly name test dataframes

032f555

tdas reviewed Sep 14, 2017

View reviewed changes

Combine test helper functions.

d7f5f60

tdas reviewed Sep 14, 2017

View reviewed changes

jose-torres added 2 commits September 14, 2017 18:18

Key watermarks by relative position rather than attribute.

2f07f90

Address test comments.

8b60538

tdas reviewed Sep 15, 2017

View reviewed changes

jose-torres closed this Sep 15, 2017

jose-torres deleted the SPARK-22017 branch September 15, 2017 08:11

jose-torres restored the SPARK-22017 branch September 15, 2017 08:13

jose-torres reopened this Sep 15, 2017

Address review comments

13affc7

tdas reviewed Sep 15, 2017

View reviewed changes

Add query restart test.

cdf4361

asfgit closed this in 0bad10d Sep 16, 2017


		val second = MemoryStream[Int]

		val secondAggregation = second.toDF()

[SPARK-22017] Take minimum of all watermark execs in StreamExecution. #19239

[SPARK-22017] Take minimum of all watermark execs in StreamExecution. #19239

Uh oh!

Conversation

jose-torres commented Sep 14, 2017

What changes were proposed in this pull request?

How was this patch tested?

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!

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!

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!

SparkQA commented Sep 15, 2017

Uh oh!

jose-torres commented Sep 15, 2017

Uh oh!

SparkQA commented Sep 15, 2017

Uh oh!

tdas Sep 15, 2017 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tdas Sep 15, 2017 • 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!

tdas Sep 15, 2017 • 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 Sep 15, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 15, 2017

Uh oh!

SparkQA commented Sep 16, 2017

Uh oh!

tdas commented Sep 16, 2017

Uh oh!

tdas Sep 15, 2017 •

edited

Loading

tdas Sep 15, 2017 •

edited

Loading

tdas Sep 15, 2017 •

edited

Loading