[SPARK-16114] [SQL] structured streaming event time window example #13957

jjthomas · 2016-06-28T23:51:31Z

What changes were proposed in this pull request?

A structured streaming example with event time windowing.

How was this patch tested?

Run locally

tdas · 2016-06-29T08:47:19Z

use the same JIRA number as the previous one.

tdas · 2016-06-29T08:49:59Z

examples/src/main/scala/org/apache/spark/examples/sql/streaming/EventTimeWindow.scala

+ * dev0,7.0,2015-03-18T12:00:00
+ * dev1,8.0,2015-03-18T12:00:10
+ * dev0,5.0,2015-03-18T12:00:20
+ * dev1,3.0,2015-03-18T12:00:30


i dont like it that the user has to enter data in a particular format. that make this clumsy to use.

how about this. lets add an option in the socket source, that attaches the receiving time as a column in the data. so the returned schema would be (time, line) if that option is specified. then we can very easily write a simple sliding window word count example using that.
what do you think?

tdas · 2016-06-30T00:31:43Z

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

@@ -67,7 +71,9 @@ class TextSocketSource(host: String, port: Int, sqlContext: SQLContext)
              return
            }
            TextSocketSource.this.synchronized {
-              lines += line
+              lines += ((line,
+                new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")


no need to create a format object every time. better to reuse.

tdas · 2016-07-01T05:29:56Z

ok to test

tdas · 2016-07-01T05:30:19Z

examples/src/main/python/sql/streaming/structured_network_wordcount.py

@@ -16,7 +16,7 @@
 #

 """
- Counts words in UTF8 encoded, '\n' delimited text received from the network every second.
+ Counts words in UTF8 encoded, '\n' delimited text received from the network.


good catch!

SparkQA · 2016-07-01T07:05:46Z

Test build #61597 has finished for PR 13957 at commit 3e15cc9.

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

tdas · 2016-07-01T07:55:18Z

.../main/scala/org/apache/spark/examples/sql/streaming/StructuredNetworkWordCountWindowed.scala

+
+    // Split the lines into words, retaining timestamps
+    val words = lines.flatMap(line =>
+      line._1.split(" ")


put the following map in the same line.

tdas · 2016-07-01T08:03:31Z

this looks good for scala, please add the java and python.

SparkQA · 2016-07-01T20:23:57Z

Test build #61635 has finished for PR 13957 at commit 57a8b11.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- public final class JavaStructuredNetworkWordCountWindowed

tdas · 2016-07-01T20:28:04Z

examples/src/main/python/sql/streaming/structured_network_wordcount_windowed.py

+    words = lines.select(
+        # explode turns each item in an array into a separate row
+        explode(
+            split(lines.value, ' ')


looks too nested, put explode and split in the same line.

tdas · 2016-07-01T20:41:03Z

this is looking good. a few cosmetic changes. can you update the windows operation section in the programming guide to use this code snippet? And link to the full source code

tdas · 2016-07-01T22:24:30Z

...ain/java/org/apache/spark/examples/sql/streaming/JavaStructuredNetworkWordCountWindowed.java

+ *    `$ nc -lk 9999`
+ * and then run the example
+ *    `$ bin/run-example sql.streaming.JavaStructuredNetworkWordCountWindowed
+ *    localhost 9999 <window duration> <optional slide duration>`


window duration in seconds...

tdas · 2016-07-01T22:24:55Z

...ain/java/org/apache/spark/examples/sql/streaming/JavaStructuredNetworkWordCountWindowed.java

+  public static void main(String[] args) throws Exception {
+    if (args.length < 3) {
+      System.err.println("Usage: JavaStructuredNetworkWordCountWindowed <hostname> <port>" +
+        " <window duration in seconds> <optional slide duration in seconds>");


optional stuff is usually written as [ ... ]. Add an example here with the 10 seconds and 5 seconds. 10 and 5 so that the user can start seeing the window immediately.

SparkQA · 2016-07-01T22:29:10Z

Test build #61638 has finished for PR 13957 at commit 8bb543d.

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

tdas · 2016-07-01T22:29:30Z

...ain/java/org/apache/spark/examples/sql/streaming/JavaStructuredNetworkWordCountWindowed.java

+
+    // Group the data by window and word and compute the count of each group
+    Dataset<Row> windowedCounts = words.groupBy(
+      functions.window(words.col("timestamp"), windowArg, slideArg),


windowArg --> windowDuration
slideArg --> slideInterval

So that this code snippet in the guide is self explanatory.
And make sure its consistent in other languages.

tdas · 2016-07-01T22:37:17Z

examples/src/main/python/sql/streaming/structured_network_wordcount_windowed.py

+
+    # Group the data by window and word and compute the count of each group
+    windowedCounts = words.groupBy(
+        window(words.timestamp, windowArg, slideArg),


rename here

SparkQA · 2016-07-01T22:39:13Z

Test build #61639 has finished for PR 13957 at commit e7a81e1.

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

tdas · 2016-07-01T22:44:56Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/TextSocketStreamSuite.scala

+      source = null
+    }
+  }
+


add tests below to make sure includeTimestamp errors are checked.

SparkQA · 2016-07-05T21:09:04Z

Test build #61789 has finished for PR 13957 at commit 893f70e.

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

SparkQA · 2016-07-05T22:49:01Z

Test build #61790 has finished for PR 13957 at commit 8f97b66.

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

tdas · 2016-07-12T00:54:08Z

LGTM. Merging this to master and 2.0. Thanks!

## What changes were proposed in this pull request? A structured streaming example with event time windowing. ## How was this patch tested? Run locally Author: James Thomas <jamesjoethomas@gmail.com> Closes #13957 from jjthomas/current. (cherry picked from commit 9e2c763) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

EventTimeWindow example

7572459

tdas reviewed Jun 29, 2016
View reviewed changes

responded to td comments

6732dcd

jjthomas changed the title ~~structured streaming event time window example~~ [SPARK-16114] [SQL] structured streaming event time window example Jun 30, 2016

tdas reviewed Jun 30, 2016
View reviewed changes

jjthomas added 2 commits June 30, 2016 18:21

Merge remote-tracking branch 'upstream/master' into current

a8e60e7

responded to TD comments

3e15cc9

tdas reviewed Jul 1, 2016
View reviewed changes

java and python

57a8b11

tdas reviewed Jul 1, 2016
View reviewed changes

td comments

8bb543d

tdas reviewed Jul 1, 2016
View reviewed changes

orderBy

e7a81e1

tdas reviewed Jul 1, 2016
View reviewed changes

jjthomas added 2 commits July 5, 2016 14:05

responded to comments

893f70e

fix

8f97b66

asfgit closed this in 9e2c763 Jul 12, 2016

[SPARK-16114] [SQL] structured streaming event time window example #13957

[SPARK-16114] [SQL] structured streaming event time window example #13957

Uh oh!

Conversation

jjthomas commented Jun 28, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

tdas commented Jun 29, 2016

Uh oh!

tdas Jun 29, 2016

Choose a reason for hiding this comment

Uh oh!

tdas Jun 30, 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 commented Jul 1, 2016

Uh oh!

tdas Jul 1, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 1, 2016

Uh oh!

tdas Jul 1, 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 commented Jul 1, 2016

Uh oh!

SparkQA commented Jul 1, 2016

Uh oh!

tdas Jul 1, 2016

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Uh oh!

tdas Jul 1, 2016

Choose a reason for hiding this comment

Uh oh!

tdas Jul 1, 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!

SparkQA commented Jul 1, 2016

Uh oh!

tdas Jul 1, 2016

Choose a reason for hiding this comment

Uh oh!

tdas Jul 1, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 1, 2016

Uh oh!

tdas Jul 1, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 5, 2016

Uh oh!

SparkQA commented Jul 5, 2016

Uh oh!

tdas commented Jul 12, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

tdas Jun 30, 2016 •

edited

Loading

tdas Jul 1, 2016 •

edited

Loading

tdas commented Jul 1, 2016 •

edited

Loading

tdas Jul 1, 2016 •

edited

Loading

tdas commented Jul 12, 2016 •

edited

Loading