[SPARK-19168][Structured Streaming] StateStore should be aborted upon error #16547

lw-lin · 2017-01-11T07:50:04Z

What changes were proposed in this pull request?

We should call StateStore.abort() when there should be any error before the store is committed.

How was this patch tested?

Manually.

SparkQA · 2017-01-11T07:55:44Z

Test build #71193 has finished for PR 16547 at commit c9f62c1.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

lw-lin · 2017-01-11T08:07:03Z

...lyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala

-            s"$outputMode output mode not supported when there are streaming aggregations on " +
-                s"streaming DataFrames/DataSets")(plan)
+            s"$outputMode output mode requires watermark to be speficied when there are " +
+              s"streaming aggregations on streaming DataFrames/DataSets")(plan)


this is an easy fix, so I've also included it in this patch

SparkQA · 2017-01-11T09:44:03Z

Test build #71195 has finished for PR 16547 at commit f24aeb1.

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

lw-lin · 2017-01-11T12:44:27Z

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

      AddData(inputData, 10),   // Should not emit anything as data less than watermark
      CheckLastBatch(),
-      assertNumStateRows(2)
+      assertNumStateRows(2, 1)  // We also processed the data 10, which is less than watermark


note this is assertNumStateRows(2, 1) without this patch, and is assertNumStateRows(2, 0) with this patch

lw-lin · 2017-01-11T12:45:04Z

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

      AddData(inputData, 10),   // Should not emit anything as data less than watermark
      CheckLastBatch(),
-      assertNumStateRows(2)
+      assertNumStateRows(2, 0)  // We processed nothing in this batch


note this is assertNumStateRows(2, 1) without this patch, and is assertNumStateRows(2, 0) with this patch.

SparkQA · 2017-01-11T14:54:49Z

Test build #71203 has finished for PR 16547 at commit caf7b34.

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

lw-lin · 2017-01-11T15:02:13Z

@tdas @zsxwing would you take a look, thanks!

zsxwing · 2017-01-11T19:59:23Z

@lw-lin I noticed a potential issue in update mode. It may not consume the whole Iterator when the downstream operator is take, which will make the state store not commit. Could you also fix it? The code should be similar to append mode. NVM. Since Structured Streaming doesn't support limit, it won't happen.

There is another issue in this class: it doesn't call StateStore.abort when an error happens. Could you also fix it?

lw-lin · 2017-01-12T02:27:37Z

@zsxwing thanks for the comments. I'll update this.

SparkQA · 2017-01-12T07:47:39Z

Test build #71248 has started for PR 16547 at commit 07a020e.

lw-lin · 2017-01-12T14:03:14Z

Jenkins retest this please

SparkQA · 2017-01-12T16:36:33Z

Test build #71266 has finished for PR 16547 at commit 07a020e.

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

lw-lin · 2017-01-13T00:04:34Z

@zsxwing updated as per your comments; would you take another look?

zsxwing · 2017-01-13T00:32:54Z

Discussed with @tdas offline. Regarding the changes to the append mode, it's unknown that if adding the filter is better because it will apply the filter on all rows but there are usually only few rows out of watermark. It's better to just leave it as it it.

For aborting StateStore, could you use TaskContext.addTaskCompletionListener instead? So that if the error doesn't happen in hasNext or next method, we can still abort it. You can add a flag to indicate if the StateStore is committed or not.

zsxwing · 2017-01-13T00:33:21Z

Could you also update the JIRA and the PR description after your changes?

lw-lin · 2017-01-13T02:54:33Z

@zsxwing thanks for the feedback! Ah, sure, let me update accordingly.

lw-lin · 2017-01-17T16:07:32Z

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

@@ -335,4 +344,67 @@ class StreamingAggregationSuite extends StreamTest with BeforeAndAfterAll {
      CheckLastBatch((90L, 1), (100L, 1), (105L, 1))
    )
  }
+


The idea of the following test is to inject a mock state store, which:

throws an error in its commit() method

marks the aborted flag in its abort() method

Then at the end of the test, we would check if aborted is true

I prefer to not add this complicated test because:

TaskContext.get().addTaskCompletionListener has already been tested in other tests.

Scala Reflection is not thread-safe in 2.10 and this test may be flaky since SQL also uses Scala Reflection.

thanks! I hesitated a little when I added it -- yea it was complicated -- so let me remove it

SparkQA · 2017-01-17T18:25:25Z

Test build #71512 has finished for PR 16547 at commit cf24fad.

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

zsxwing

I suggest to not add the test. Otherwise, LGTM.

zsxwing · 2017-01-18T00:59:10Z

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

@@ -335,4 +344,67 @@ class StreamingAggregationSuite extends StreamTest with BeforeAndAfterAll {
      CheckLastBatch((90L, 1), (100L, 1), (105L, 1))
    )
  }
+


I prefer to not add this complicated test because:

TaskContext.get().addTaskCompletionListener has already been tested in other tests.

Scala Reflection is not thread-safe in 2.10 and this test may be flaky since SQL also uses Scala Reflection.

SparkQA · 2017-01-18T05:53:28Z

Test build #71565 has finished for PR 16547 at commit 0f9e54d.

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

zsxwing · 2017-01-18T18:52:26Z

LGTM. Merging to master and 2.1. Thanks!

… error ## What changes were proposed in this pull request? We should call `StateStore.abort()` when there should be any error before the store is committed. ## How was this patch tested? Manually. Author: Liwei Lin <lwlin7@gmail.com> Closes #16547 from lw-lin/append-filter. (cherry picked from commit 569e506) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

… error ## What changes were proposed in this pull request? We should call `StateStore.abort()` when there should be any error before the store is committed. ## How was this patch tested? Manually. Author: Liwei Lin <lwlin7@gmail.com> Closes apache#16547 from lw-lin/append-filter.

lw-lin force-pushed the append-filter branch from c9f62c1 to f24aeb1 Compare January 11, 2017 08:05

lw-lin commented Jan 11, 2017

View reviewed changes

lw-lin force-pushed the append-filter branch from f24aeb1 to caf7b34 Compare January 11, 2017 12:40

lw-lin commented Jan 11, 2017

View reviewed changes

Abort StateStore on error

cf24fad

lw-lin force-pushed the append-filter branch from 07a020e to cf24fad Compare January 17, 2017 16:00

lw-lin changed the title ~~[SPARK-19168][Structured Streaming] Improvement: filter late data using watermark for Append mode~~ [SPARK-19168][Structured Streaming] StateStore should be aborted upon error Jan 17, 2017

lw-lin commented Jan 17, 2017

View reviewed changes

zsxwing requested changes Jan 18, 2017

View reviewed changes

Remove test

0f9e54d

asfgit closed this in 569e506 Jan 18, 2017

lw-lin deleted the append-filter branch February 8, 2017 02:59

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-19168][Structured Streaming] StateStore should be aborted upon error #16547

[SPARK-19168][Structured Streaming] StateStore should be aborted upon error #16547

lw-lin commented Jan 11, 2017 •

edited

Loading

SparkQA commented Jan 11, 2017

lw-lin Jan 11, 2017

SparkQA commented Jan 11, 2017

lw-lin Jan 11, 2017

lw-lin Jan 11, 2017 •

edited

Loading

SparkQA commented Jan 11, 2017

lw-lin commented Jan 11, 2017

zsxwing commented Jan 11, 2017 •

edited

Loading

lw-lin commented Jan 12, 2017

SparkQA commented Jan 12, 2017

lw-lin commented Jan 12, 2017

SparkQA commented Jan 12, 2017

lw-lin commented Jan 13, 2017

zsxwing commented Jan 13, 2017

zsxwing commented Jan 13, 2017

lw-lin commented Jan 13, 2017 •

edited

Loading

lw-lin Jan 17, 2017 •

edited

Loading

zsxwing Jan 18, 2017

lw-lin Jan 18, 2017

SparkQA commented Jan 17, 2017

zsxwing left a comment

zsxwing Jan 18, 2017

SparkQA commented Jan 18, 2017

zsxwing commented Jan 18, 2017

[SPARK-19168][Structured Streaming] StateStore should be aborted upon error #16547

[SPARK-19168][Structured Streaming] StateStore should be aborted upon error #16547

Conversation

lw-lin commented Jan 11, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Jan 11, 2017

lw-lin Jan 11, 2017

Choose a reason for hiding this comment

SparkQA commented Jan 11, 2017

lw-lin Jan 11, 2017

Choose a reason for hiding this comment

lw-lin Jan 11, 2017 • edited Loading

Choose a reason for hiding this comment

SparkQA commented Jan 11, 2017

lw-lin commented Jan 11, 2017

zsxwing commented Jan 11, 2017 • edited Loading

lw-lin commented Jan 12, 2017

SparkQA commented Jan 12, 2017

lw-lin commented Jan 12, 2017

SparkQA commented Jan 12, 2017

lw-lin commented Jan 13, 2017

zsxwing commented Jan 13, 2017

zsxwing commented Jan 13, 2017

lw-lin commented Jan 13, 2017 • edited Loading

lw-lin Jan 17, 2017 • edited Loading

Choose a reason for hiding this comment

zsxwing Jan 18, 2017

Choose a reason for hiding this comment

lw-lin Jan 18, 2017

Choose a reason for hiding this comment

SparkQA commented Jan 17, 2017

zsxwing left a comment

Choose a reason for hiding this comment

zsxwing Jan 18, 2017

Choose a reason for hiding this comment

SparkQA commented Jan 18, 2017

zsxwing commented Jan 18, 2017

lw-lin commented Jan 11, 2017 •

edited

Loading

lw-lin Jan 11, 2017 •

edited

Loading

zsxwing commented Jan 11, 2017 •

edited

Loading

lw-lin commented Jan 13, 2017 •

edited

Loading

lw-lin Jan 17, 2017 •

edited

Loading