[SPARK-20030][SS] Event-time-based timeout for MapGroupsWithState #17361

tdas · 2017-03-20T11:05:04Z

What changes were proposed in this pull request?

Adding event time based timeout. The user sets the timeout timestamp directly using KeyedState.setTimeoutTimestamp. The keys times out when the watermark crosses the timeout timestamp.

How was this patch tested?

Unit tests

SparkQA · 2017-03-20T12:35:13Z

Test build #74881 has finished for PR 17361 at commit 3f77c01.

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

SparkQA · 2017-03-20T20:40:05Z

Test build #74894 has finished for PR 17361 at commit 6e9f408.

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

SparkQA · 2017-03-21T00:46:21Z

Test build #74910 has finished for PR 17361 at commit ac17886.

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

SparkQA · 2017-03-21T03:46:05Z

Test build #74919 has finished for PR 17361 at commit f6d2143.

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

tdas · 2017-03-21T09:23:19Z

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

-              if (outputMode != InternalOutputModes.Update) {
-                throwError("flatMapGroupsWithState in update mode is not supported with " +
+        // mapGroupsWithState and flatMapGroupsWithState
+        case m: FlatMapGroupsWithState if m.isStreaming =>


Refactored this to contains all tests related to map/flatMapGroupsWithState under a single case statement. this way its easier to reason whether all the possible combinations of operator+output-mode+aggregation has been covered.

Also it consolidates all the "valid combinations" of mode + aggs on which additional checks can be made (presence of watermark when timeoutConf = EventTimeTimeout)

This refactoring passes all existing tests in UnsupportedOperationsSuite.

Wow, this is getting complicated...

SparkQA · 2017-03-21T10:46:39Z

Test build #74969 has finished for PR 17361 at commit 2c5592c.

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

SparkQA · 2017-03-21T11:44:56Z

Test build #74971 has finished for PR 17361 at commit 0523aaf.

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

SparkQA · 2017-03-21T22:50:40Z

Test build #75003 has finished for PR 17361 at commit d0758eb.

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

SparkQA · 2017-03-21T23:24:48Z

Test build #75001 has finished for PR 17361 at commit 6759165.

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

marmbrus · 2017-03-21T23:49:23Z

sql/catalyst/src/main/java/org/apache/spark/sql/streaming/KeyedStateTimeout.java

+  /**
+   * Timeout based on processing time. The duration of timeout can be set for each group in
+   * `map/flatMapGroupsWithState` by calling `KeyedState.setTimeoutDuration()`.
+   */
  public static KeyedStateTimeout ProcessingTimeTimeout() { return ProcessingTimeTimeout$.MODULE$; }


Nit: I'd consider removing the Timeout here and as its kind of redundant.

Its just that if someone this does import KeyedStateTimeout._ the code boils down to
flatMapGroupsWithState(Update, ProcessingTime) { ... } with no reference to timeout.

Fine either way.

I'd probably still remove it.

marmbrus · 2017-03-21T23:52:12Z

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

+            }
+            if (watermarkAttributes.isEmpty) {
+              throwError(
+                "Event time timeout is not supported in a [map|flatMap]GroupsWithState " +


I think we are hyphenating event-time?

are we? I didnt know there was a policy. I am fine hyphenating.

i want it to be consistent and the docs hyphenate.

Aah, I get it. The apache docs does have hyphenation.

marmbrus · 2017-03-21T23:52:57Z

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

+            if (watermarkAttributes.isEmpty) {
+              throwError(
+                "Event time timeout is not supported in a [map|flatMap]GroupsWithState " +
+                  "without watermark. Use '[Dataset/DataFrame].withWatermark()' to " +


I would consider making this an affirmative statement. "You must define a watermark on a dataframe in order to use event-time based timeouts".

marmbrus · 2017-03-21T23:55:21Z

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

+        .withWatermark("eventTime", "10 seconds")
+        .as[(String, Long)]
+        .groupByKey[String]((x: (String, Long)) => x._1)
+        .flatMapGroupsWithState[Long, (String, Int)](Update, EventTimeTimeout)(stateFunc)


These types are just here for testing? (i.e. we didn't break inference right?)

I was debugging and I left them there thinking it help readability of tests. I can remove them.

As long as they aren't required its okay.

marmbrus · 2017-03-21T23:56:55Z

sql/catalyst/src/main/java/org/apache/spark/sql/streaming/KeyedStateTimeout.java

+   * Timeout based on event time. The event time timestamp for timeout can be set for each
+   * group in `map/flatMapGroupsWithState` by calling `KeyedState.setTimeoutTimestamp()`.
+   * In addition, you have to define the watermark in the query using `Dataset.withWatermark`.
+   * When the watermark advances beyond the set timestamp of a group, then the group times out.


And no data has arrived for that group.

marmbrus · 2017-03-22T00:16:11Z

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

- * @param isTimeoutEnabled Whether timeout is enabled. This will be used to check whether the user
- *                         is allowed to configure timeouts.
+ * @param timeoutConf   Type of timeout configured. Based on this, different operations will
+ *                        be supported.


nit: indent is inconsistent

marmbrus · 2017-03-22T00:22:55Z

LGTM

SparkQA · 2017-03-22T00:40:16Z

Test build #75010 has finished for PR 17361 at commit 64b6abf.

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

SparkQA · 2017-03-22T03:26:57Z

Test build #75017 has finished for PR 17361 at commit 9c9668b.

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

SparkQA · 2017-03-22T04:08:24Z

Test build #3604 has finished for PR 17361 at commit 9c9668b.

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

hhbyyh · 2017-03-22T21:52:39Z

@tdas Just FYI, I'm getting lint-java error:

yuhao@yuhao-devbox:~/workspace/github/hhbyyh/spark$ ./dev/lint-java
~Using mvn from path: /usr/bin/mvn
Checkstyle checks failed at following occurrences:
[ERROR] src/main/java/org/apache/spark/sql/streaming/KeyedStateTimeout.java:[41,35] (naming) MethodName: Method name 'ProcessingTimeTimeout' must match pattern '^[a-z][a-z0-9][a-zA-Z0-9_]$'.
[ERROR] src/main/java/org/apache/spark/sql/streaming/KeyedStateTimeout.java:[51,35] (naming) MethodName: Method name 'EventTimeTimeout' must match pattern '^[a-z][a-z0-9][a-zA-Z0-9_]$'.
[ERROR] src/main/java/org/apache/spark/sql/streaming/KeyedStateTimeout.java:[54,35] (naming) MethodName: Method name 'NoTimeout' must match pattern '^[a-z][a-z0-9][a-zA-Z0-9_]*$'.

Is it just me? Maybe we should suppress the style error.

First draft of event time timeout

3f77c01

Fixed test

6e9f408

Disallowed timeout timestamp for processing time timeout

ac17886

Added watermark based filtering on event time

f6d2143

Added check for timestamp < watermark

2c5592c

tdas commented Mar 21, 2017

View reviewed changes

tdas changed the title ~~[SPARK-20030][SS][WIP]Event-time-based timeout for MapGroupsWithState~~ [SPARK-20030][SS] Event-time-based timeout for MapGroupsWithState Mar 21, 2017

Minor changes

0523aaf

tdas added 2 commits March 21, 2017 14:18

Added additional setTimeoutTimestamp

6759165

Use KeyedStateTime.* in test suite instead of the inner classes directly

d0758eb

Added docs to KeyedState

64b6abf

marmbrus reviewed Mar 21, 2017

View reviewed changes

marmbrus reviewed Mar 22, 2017

View reviewed changes

Addressed comments

9c9668b

asfgit closed this in c1e87e3 Mar 22, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-20030][SS] Event-time-based timeout for MapGroupsWithState #17361

[SPARK-20030][SS] Event-time-based timeout for MapGroupsWithState #17361

tdas commented Mar 20, 2017

SparkQA commented Mar 20, 2017

SparkQA commented Mar 20, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

tdas Mar 21, 2017 •

edited

tdas Mar 21, 2017

marmbrus Mar 21, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

marmbrus Mar 21, 2017

tdas Mar 22, 2017

marmbrus Mar 22, 2017

marmbrus Mar 21, 2017

tdas Mar 22, 2017

marmbrus Mar 22, 2017

tdas Mar 22, 2017

marmbrus Mar 21, 2017 •

edited

tdas Mar 22, 2017

marmbrus Mar 21, 2017

tdas Mar 22, 2017

marmbrus Mar 22, 2017

marmbrus Mar 21, 2017

marmbrus Mar 22, 2017

marmbrus commented Mar 22, 2017

SparkQA commented Mar 22, 2017

SparkQA commented Mar 22, 2017

SparkQA commented Mar 22, 2017

hhbyyh commented Mar 22, 2017 •

edited

[SPARK-20030][SS] Event-time-based timeout for MapGroupsWithState #17361

[SPARK-20030][SS] Event-time-based timeout for MapGroupsWithState #17361

Conversation

tdas commented Mar 20, 2017

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Mar 20, 2017

SparkQA commented Mar 20, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

tdas Mar 21, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

SparkQA commented Mar 21, 2017

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

Choose a reason for hiding this comment

marmbrus Mar 21, 2017 • edited

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

Choose a reason for hiding this comment

marmbrus commented Mar 22, 2017

SparkQA commented Mar 22, 2017

SparkQA commented Mar 22, 2017

SparkQA commented Mar 22, 2017

hhbyyh commented Mar 22, 2017 • edited

tdas Mar 21, 2017 •

edited

marmbrus Mar 21, 2017 •

edited

hhbyyh commented Mar 22, 2017 •

edited