[SPARK-14160] Time Windowing functions for Datasets #12008

brkyvz · 2016-03-28T19:06:55Z

What changes were proposed in this pull request?

This PR adds the function window as a column expression.

window can be used to bucket rows into time windows given a time column. With this expression, performing time series analysis on batch data, as well as streaming data should become much more simpler.

Usage

Assume the following schema:

sensor_id, measurement, timestamp

To average 5 minute data every 1 minute (window length of 5 minutes, slide duration of 1 minute), we will use:

df.groupBy(window("timestamp", “5 minutes”, “1 minute”), "sensor_id")
  .agg(mean("measurement").as("avg_meas"))

This will generate windows such as:

09:00:00-09:05:00
09:01:00-09:06:00
09:02:00-09:07:00 ...

Intervals will start at every slideDuration starting at the unix epoch (1970-01-01 00:00:00 UTC).
To start intervals at a different point of time, e.g. 30 seconds after a minute, the startTime parameter can be used.

df.groupBy(window("timestamp", “5 minutes”, “1 minute”, "30 second"), "sensor_id")
  .agg(mean("measurement").as("avg_meas"))

This will generate windows such as:

09:00:30-09:05:30
09:01:30-09:06:30
09:02:30-09:07:30 ...

Support for Python will be made in a follow up PR after this.

How was this patch tested?

This patch has some basic unit tests for the TimeWindow expression testing that the parameters pass validation, and it also has some unit/integration tests testing the correctness of the windowing and usability in complex operations (multi-column grouping, multi-column projections, joins).

SparkQA · 2016-03-28T19:08:15Z

Test build #54343 has finished for PR 12008 at commit 9e7febb.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-03-28T19:13:52Z

Test build #54344 has finished for PR 12008 at commit 5787e35.

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

marmbrus · 2016-03-28T19:14:41Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TimeWindow.scala

+	override def dataType: DataType = outputType
+
+	private def outputType: StructType = StructType(Seq(
+		StructField("start", TimestampType), StructField("end", TimestampType)))


Nit: indents are off in this file and the next.

marmbrus · 2016-03-28T19:19:09Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+              Add(division, Literal(i - windowExpr.maxNumOverlapping)),
+              Literal(windowExpr.slideDuration)),
+            Literal(windowExpr.startTime)),
+          Literal(1000000))


This might be easier to read if it was written with the dsl.

marmbrus · 2016-03-28T19:21:14Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

+      windowDuration: String,
+      slideDuration: String,
+      startTime: String): Column = withExpr {
+    TimeWindow(timeColumn.expr, windowDuration, slideDuration, startTime)


Maybe we should just parse here?

Or maybe in the companion object? Or with another constructor? The _param / lazy val is a little odd.

+1 on parsing it here

Added the companion object apply method

marmbrus · 2016-03-28T19:21:42Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TimeWindow.scala

+		(cal.months * 4 * CalendarInterval.MICROS_PER_WEEK + cal.microseconds) / 1000000
+	}
+
+	// The window duration in seconds


Does this mean that the smallest window is 1 second?

unfortunately yes. The conversion from LongType to TimestampType has second precision.

rxin · 2016-03-28T20:04:24Z

lower case Dataset

SparkQA · 2016-03-28T20:55:15Z

Test build #54345 has finished for PR 12008 at commit ad98902.

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

rxin · 2016-03-28T21:59:33Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

+   * @group datetime_funcs
+   * @since 2.0.0
+   */
+  def window(timeColumn: Column, windowDuration: String): Column = {


experimental tag

SparkQA · 2016-03-28T22:04:44Z

Test build #54370 has finished for PR 12008 at commit 6a784b7.

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

SparkQA · 2016-03-28T22:19:10Z

Test build #54371 has finished for PR 12008 at commit b7154b2.

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

SparkQA · 2016-03-29T00:04:19Z

Test build #54372 has finished for PR 12008 at commit 7eea448.

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

brkyvz · 2016-03-29T00:13:24Z

retest this please

SparkQA · 2016-03-29T02:00:24Z

Test build #54387 has finished for PR 12008 at commit 7eea448.

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

some cleanup

SparkQA · 2016-03-30T01:48:46Z

Test build #54484 has finished for PR 12008 at commit b4e2fc2.

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

zsxwing · 2016-03-30T06:10:34Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TimeWindow.scala

+      throw new IllegalArgumentException(
+        s"The provided interval ($interval) did not correspond to a valid interval string.")
+    }
+    (cal.months * 4 * CalendarInterval.MICROS_PER_WEEK + cal.microseconds) / 1000000


4 weeks == 1 month looks weird. Maybe define window("timestamp", "1 month") as groupBy(getMonthInYear("timestamp")) is more intuitive?

4 weeks == 1 month looks weird. Maybe define window("timestamp", "1 month") as groupBy(getMonthInYear("timestamp")) is more intuitive?

This looks hard to implement. Maybe we just don't need to support month or year.

By the way, does this line mean that the user cannot use window("timestamp", "500 milliseconds")?

We can't operate on TimestampType as if they are Longs. They get cast to
LongType AFAIK, which has second precision. I'm not sure if supporting ms
is possible.
On Mar 29, 2016 11:14 PM, "Shixiong Zhu" notifications@github.com wrote:

In
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TimeWindow.scala
#12008 (comment):

private def getIntervalInSeconds(interval: String): Long = {

if (StringUtils.isBlank(interval)) {

throw new IllegalArgumentException(

"The window duration, slide duration and start time cannot be null or blank.")

}

val intervalString = if (interval.startsWith("interval")) {

interval

} else {

"interval " + interval

}

val cal = CalendarInterval.fromString(intervalString)

if (cal == null) {

throw new IllegalArgumentException(

s"The provided interval ($interval) did not correspond to a valid interval string.")

}

(cal.months * 4 * CalendarInterval.MICROS_PER_WEEK + cal.microseconds) / 1000000

By the way, does it mean that the user cannot use window("timestamp", "500
milliseconds")?

—
You are receiving this because you authored the thread.
Reply to this email directly or view it on GitHub
https://github.com/apache/spark/pull/12008/files/b4e2fc23585413b1bf50e2487437dd38b9cd748f#r57840903

timestamp precision i think is 100ns

I agree we probably don't want to support month intervals since they are variable length. If people want to group on calendar boundaries they can use existing data functions.

SparkQA · 2016-03-31T01:23:51Z

Test build #54574 has finished for PR 12008 at commit f756556.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PreciseTimestamp(child: Expression) extends UnaryExpression with ExpectsInputTypes

SparkQA · 2016-03-31T02:41:54Z

Test build #54575 has finished for PR 12008 at commit 8bc9799.

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

SparkQA · 2016-03-31T17:28:27Z

Test build #54640 has finished for PR 12008 at commit ff27b7a.

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

marmbrus · 2016-04-01T20:18:56Z

Thanks! Merging to master.

## What changes were proposed in this pull request? The `window` function was added to Dataset with [this PR](#12008). This PR adds the Python, and SQL, API for this function. With this PR, SQL, Java, and Scala will share the same APIs as in users can use: - `window(timeColumn, windowDuration)` - `window(timeColumn, windowDuration, slideDuration)` - `window(timeColumn, windowDuration, slideDuration, startTime)` In Python, users can access all APIs above, but in addition they can do - In Python: `window(timeColumn, windowDuration, startTime=...)` that is, they can provide the startTime without providing the `slideDuration`. In this case, we will generate tumbling windows. ## How was this patch tested? Unit tests + manual tests Author: Burak Yavuz <brkyvz@gmail.com> Closes #12136 from brkyvz/python-windows.

## What changes were proposed in this pull request? The `window` function was added to Dataset with [this PR](#12008). This PR adds the R API for this function. With this PR, SQL, Java, and Scala will share the same APIs as in users can use: - `window(timeColumn, windowDuration)` - `window(timeColumn, windowDuration, slideDuration)` - `window(timeColumn, windowDuration, slideDuration, startTime)` In Python and R, users can access all APIs above, but in addition they can do - In R: `window(timeColumn, windowDuration, startTime=...)` that is, they can provide the startTime without providing the `slideDuration`. In this case, we will generate tumbling windows. ## How was this patch tested? Unit tests + manual tests Author: Burak Yavuz <brkyvz@gmail.com> Closes #12141 from brkyvz/R-windows.

brkyvz added 6 commits March 26, 2016 14:02

time windowing

168568f

finished adding time windowing

df7bce0

works

03c1456

minor clean up

a41ee4c

tests

db50465

finished

9e7febb

brkyvz added 2 commits March 28, 2016 12:08

one more test

535d592

added apache headers

5787e35

marmbrus reviewed Mar 28, 2016
View reviewed changes

fix indentation

ed9ec34

marmbrus reviewed Mar 28, 2016
View reviewed changes

try this

9301b1c

marmbrus reviewed Mar 28, 2016
View reviewed changes

fix test file as well

ad98902

marmbrus reviewed Mar 28, 2016
View reviewed changes

brkyvz changed the title ~~[SPARK-14160] Time Windowing functions for DataSets~~ [SPARK-14160] Time Windowing functions for Datasets Mar 28, 2016

address most comments

6a784b7

rxin reviewed Mar 28, 2016
View reviewed changes

added experimental tags

b7154b2

fix ss

7eea448

marmbrus mentioned this pull request Mar 29, 2016

[SPARK-14176][SQL]Add DataFrameWriter.trigger to set the stream batch period #11976

Closed

marmbrus and others added 3 commits March 29, 2016 18:11

some cleanup

200e480

more cleanup

83dd46d

Merge pull request #6 from marmbrus/pr/12008

b4e2fc2

some cleanup

make as much progress as possible

f1aa357

zsxwing reviewed Mar 30, 2016
View reviewed changes

brkyvz added 2 commits March 30, 2016 18:20

finished

f756556

minor clean up

4eb57c5

fix ss

8bc9799

fix test

ff27b7a

asfgit closed this in 1b829ce Apr 1, 2016

This was referenced Apr 3, 2016

[SPARK-14353] Dataset Time Window window API for Python, and SQL #12136

Closed

[SPARK-14353] Dataset Time Window window API for R #12141

Closed

brkyvz deleted the df-time-window branch February 3, 2019 20:59

[SPARK-14160] Time Windowing functions for Datasets #12008

[SPARK-14160] Time Windowing functions for Datasets #12008

Uh oh!

Conversation

brkyvz commented Mar 28, 2016

What changes were proposed in this pull request?

Usage

How was this patch tested?

Uh oh!

SparkQA commented Mar 28, 2016

Uh oh!

SparkQA commented Mar 28, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rxin commented Mar 28, 2016

Uh oh!

SparkQA commented Mar 28, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Mar 28, 2016

Uh oh!

SparkQA commented Mar 28, 2016

Uh oh!

SparkQA commented Mar 29, 2016

Uh oh!

brkyvz commented Mar 29, 2016

Uh oh!

SparkQA commented Mar 29, 2016

Uh oh!

SparkQA commented Mar 30, 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!

SparkQA commented Mar 31, 2016

Uh oh!

SparkQA commented Mar 31, 2016

Uh oh!

SparkQA commented Mar 31, 2016

Uh oh!

marmbrus commented Apr 1, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants