[SPARK-7322] [SQL] [WIP] Support Window Function in DataFrame

[chenghao-intel]

This is a WIP PR for early feedback.

The usage is kind of like:

df.select(
        $"key",
        avg("value").over(
          partitionBy($"key")
          .orderBy($"value")
          .rows   // RowFrame
          .between
          .preceding(1)
          .and
          .following(1))

df.select(
        $"key",
        avg("value").over(
          partitionBy($"key")
          .orderBy($"value")
          .range// RangeFrame
          .preceding(1))

// Define a new window
val w = partitionBy("key").orderBy("value")
df.select(
   lead("key").over(w), // binding the predefined window
   lead("value").over(w)) // binding the predefined window

• API design
• Scala Doc
• More specific Unit Tests
• Python API Support
• More function added e.g. (NTILE, ROW_NUMBER, DENSE_RANK, RANK, CUME_DIST and PERCENT_RANK, which supported by Hive)

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[SparkQA]

Test build #32572 has started for PR 6104 at commit 4368a81.

[SparkQA]

Test build #32572 has finished for PR 6104 at commit 4368a81.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class WindowFunctionDefinition protected[sql](

[AmplabJenkins]

Merged build finished. Test PASSed.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/32572/
Test PASSed.

[hvanhovell]

Hi,

In the JIRA the following examples is given:

df.select(
  df.store,
  df.date,
  df.sales,
  avg(df.sales).over.partitionBy(df.store)
                    .orderBy(df.store) 
                    .rowsFollowing(0)    // this means from unbounded preceding to current row
)

Is there a reason for why the aggregation operation has moved from the beginning (the style in the JIRA), to the end (style above)? Are both still possible? I'd prefer the former, since it seems a bit shorter, and more recognizable comming from SQL.

On a related note. Is it also an idea to be able to create a seperate window (groupBy/orderBy) definition and use this definition in one or more windowed aggregates. For example:

val window = partitionBy($"store").orderBy($"date)
df.select (
  $"store"
  ,$"date"
  ,sum($"sales").over(window).rowsFollowing(0).as("TotalSales")
  ,sum($"sales").over(window).rowsFollowing(0).rowsPreceding(2).as("SalesLast3M")
)

[rxin]

It seems to me it'd be easier to have the aggregate function in the front also. @chenghao-intel any reason you designed it this way? Is it to accommodate multiple aggregates for the same window?

[chenghao-intel]

Oh, actually it was my bad, the over should be a independent function call, not a method of Column.

The window definition should be quite independent also, and it can be given a name, as @rxin said, it would be simpler if accommodating multiple functions. And, it would be easier to create functions(aggregate/window function) via the object WindowFunctionDefinition instead of the DataFrame, right?

[chenghao-intel]

rowsFollowing(0) is a good idea for representing the current row. :)

For .over(window).rowsFollowing(0) V.S. over.rowsFollowing(0).withPartition, the former might make it a little more complicated in code-wise. I can update the code once we get more feedback?

[rxin]

@chenghao-intel - I think it's better to follow SQL more closely here. I don't think it is much harder to write multiple aggregates, since the user can easily just add a function to create the window statement. On the contrary, it is more clear how many aggregates we are applying if we put this in the front.

[chenghao-intel]

Ok, I will update the code soon.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[SparkQA]

Test build #32702 has started for PR 6104 at commit d37fbe4.

[chenghao-intel]

I've updated the code and description, however, I have no idea if we can remove the toColumn method from the WindownFunctionDefinition.
Any thoughts? @rxin

[SparkQA]

Test build #32702 has finished for PR 6104 at commit d37fbe4.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class WindowFunctionDefinition protected[sql](

[AmplabJenkins]

Merged build finished. Test PASSed.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/32702/
Test PASSed.

[ash211]

There was talk earlier of referencing the window function API that jooq uses when implementing this in SparkSQL. Is it a goal to make this similar to jooq's syntax?

http://blog.jooq.org/2013/11/03/probably-the-coolest-sql-feature-window-functions/

[rxin]

Yup it is fairly similar.

[chenghao-intel]

Thank you @ash211 That's really a cool idea to support a named window in the page that you send. I will update the code.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[SparkQA]

Test build #32827 has started for PR 6104 at commit a4da7fe.

[chenghao-intel]

@rxin @yhuai , any comments on the API design? Then I can start the rest of the work.

[SparkQA]

Test build #32827 has finished for PR 6104 at commit a4da7fe.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class WindowFunctionDefinition protected[sql](

[AmplabJenkins]

Merged build finished. Test FAILed.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/32827/
Test FAILed.

[chenghao-intel]

@rxin Updated!
But still need to add scaladoc, before I doing this, can you review the interface again?

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[chenghao-intel]

change the function name.

[SparkQA]

Test build #33254 has started for PR 6104 at commit d625a64.

[rxin]

i think about it more -- this is actually problematic for java because none partitionBy, orderBy won't become static methods due to conflicts with the Window class. Here's the fix.

Move "class Window" into "object Window", and rename it to WindowSpec, and then just define the two partitionBy / orderBy top level methods in object Window. If we need another method for a window spec that doesn't have partitionBy/orderBy, we can add another one - I don't have a good name for it yet.

[rxin]

@yhuai suggested "allRows"

[SparkQA]

Test build #33254 has finished for PR 6104 at commit d625a64.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class Window

[AmplabJenkins]

Merged build finished. Test FAILed.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/33254/
Test FAILed.

[rxin]

Actually I will submit a PR against your branch.