[SPARK-8992] [SQL] Add pivot to dataframe api

[aray]

This adds a pivot method to the dataframe api.

Following the lead of cube and rollup this adds a Pivot operator that is translated into an Aggregate by the analyzer.

Currently the syntax is like:
courseSales.pivot(Seq($"year"), $"course", Seq("dotNET", "Java"), sum($"earnings"))

Would we be interested in the following syntax also/alternatively? and

courseSales.groupBy($"year").pivot($"course", "dotNET", "Java").agg(sum($"earnings"))
//or
courseSales.groupBy($"year").pivot($"course").agg(sum($"earnings"))

Later we can add it to SQLParser, but as Hive doesn't support it we cant add it there, right?

Also what would be the suggested Java friendly method signature for this?

[rxin]

Jenkins, ok to test.

[aray]

@rxin it looks like Jenkins forgot about building this. Can you help trigger the build again?

[SparkQA]

Test build #1319 has finished for PR 7841 at commit 599e9e0.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class Pivot(

[rxin]

@aray FYI this didn't make it into the 1.5 release (was submitted too close to the feature freeze deadline), but we will try to include it in Spark 1.6.

[JoshRosen]

@rxin, do you want to revisit this now for 1.6?

[aray]

@rxin and @JoshRosen, this is ready for review now.

[rxin]

@aray Thanks a lot for updating this. To help api design, can you take a look at other frameworks and see what their signatures look like?

[aray]

@rxin here is my summary of other frameworks API's

I'm going to use an example dataset form the pandas doc for all the examples (as df)

A	B	C	D
foo	one	small	1
foo	one	large	2
foo	one	large	2
foo	two	small	3
foo	two	small	3
bar	one	large	4
bar	one	small	5
bar	two	small	6
bar	two	large	7

This API

scala> df.groupBy("A", "B").pivot("C", "small", "large").sum("D").show
+---+---+-----+-----+
|  A|  B|small|large|
+---+---+-----+-----+
|foo|two|    6| null|
|bar|two|    6|    7|
|foo|one|    1|    4|
|bar|one|    5|    4|
+---+---+-----+-----+

scala> df.groupBy("A", "B").pivot("C", "small", "large").agg(sum("D"), avg("D")).show
+---+---+------------+------------+------------+------------+
|  A|  B|small sum(D)|small avg(D)|large sum(D)|large avg(D)|
+---+---+------------+------------+------------+------------+
|foo|two|           6|         3.0|        null|        null|
|bar|two|           6|         6.0|           7|         7.0|
|foo|one|           1|         1.0|           4|         2.0|
|bar|one|           5|         5.0|           4|         4.0|
+---+---+------------+------------+------------+------------+

scala> df.pivot(Seq($"A", $"B"), $"C", Seq("small", "large"), sum($"D")).show
+---+---+-----+-----+
|  A|  B|small|large|
+---+---+-----+-----+
|foo|two|    6| null|
|bar|two|    6|    7|
|foo|one|    1|    4|
|bar|one|    5|    4|
+---+---+-----+-----+

We require a list of values for the pivot column as we are required to know the output columns of the operator ahead of time. Pandas and reshape2 do not require this but the comparable SQL operators do. We also allow multiple aggregations which not all implementations allow.

pandas

The comparable metod for pandas is pivot_table(data, values=None, index=None, columns=None, aggfunc='mean', fill_value=None, margins=False, dropna=True)

Example

>>> pivot_table(df, values='D', index=['A', 'B'], columns=['C'], aggfunc=np.sum)
          small  large
foo  one  1      4
     two  6      NaN
bar  one  5      4
     two  6      7

Pandas also allows multiple aggregations:

>>> pivot_table(df, values='D', index=['A', 'B'], columns=['C'], aggfunc=[np.sum, np.average])
          sum       average      
C       large small   large small
A   B                            
bar one     4     5       4     5
    two     7     6       7     6
foo one     4     1       2     1
    two   NaN     6     NaN     3

References

• http://pandas.pydata.org/pandas-docs/stable/reshaping.html
• http://pandas.pydata.org/pandas-docs/stable/generated/pandas.pivot_table.html

See also: pivot, stack, unstack.

reshape2 (R)

The comparable method for reshape2 is dcast(data, formula, fun.aggregate = NULL, ..., margins = NULL, subset = NULL, fill = NULL, drop = TRUE, value.var = guess_value(data))

> dcast(df, A + B ~ C, sum)
Using D as value column: use value.var to override.
    A   B large small
1 bar one     4     5
2 bar two     7     6
3 foo one     4     1
4 foo two     0     6

Note that by default cast fills with the value from applying fun.aggregate to 0 length vector

References

• https://cran.r-project.org/web/packages/reshape2/reshape2.pdf
• http://seananderson.ca/2013/10/19/reshape.html
• http://www.inside-r.org/packages/cran/reshape2/docs/cast

See also: melt.

MS SQL Server

SELECT *
FROM df
pivot (sum(D) for C in ([small], [large])) p

http://sqlfiddle.com/#!3/cf887/3/0

References

• http://sqlhints.com/2014/03/10/pivot-and-unpivot-in-sql-server/

Oracle 11g

SELECT *
FROM df
pivot (sum(D) for C in ('small', 'large')) p

http://sqlfiddle.com/#!4/29bc5/3/0

Oracle also allows multiple aggregations and with similar output to this api

SELECT *
FROM df
pivot (sum(D) as sum, avg(D) as avg for C in ('small', 'large')) p

http://sqlfiddle.com/#!4/29bc5/5/0

References

• http://www.oracle.com/technetwork/articles/sql/11g-pivot-097235.html
• http://docs.oracle.com/cd/B28359_01/server.111/b28286/statements_10002.htm#CHDCEJJE
• http://www.techonthenet.com/oracle/pivot.php

---

Let me know if I can do anything else to help this along. Also would you mind adding me to the jenkins whitelist so I can test it?

[marmbrus]

ok to test

[SparkQA]

Test build #44249 has finished for PR 7841 at commit f2827ea.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Pivot(\n

[rxin]

I like your 2nd interface more (group by and then pivot), since it is easier to get that working for both Java and Scala. We can implement a simpler interface for Python/R that's closer to existing frameworks.

How hard would it be to not require the values?

[aray]

@rxin, Not requiring the values would necessitate doing a separate query for the distinct values of the column before the pivot query. It looks like at least some DF operations (eg, drop) would need the result so even if we made Pivot.output lazy we would be running an unexpected job.

If a user really didn't want to specify the values, they can explicitly do the query:

    df.groupBy("A", "B").pivot("C", df.select("C").distinct.collect.map(_.getString(0)): _*).sum("D")

Needing to know the output columns of an operator for analysis/planning is probably why the other SQL implementations require the values also (technically Oracle supports omitting it but only in XML mode where you essentially just get one column).

[SparkQA]

Test build #44643 has finished for PR 7841 at commit 403f966.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Pivot(\n

[rxin]

@aray sorry was away for spark summit - back now and will get to this today.

[SparkQA]

Test build #45316 has finished for PR 7841 at commit d8e473c.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Pivot(\n

[SparkQA]

Test build #45366 has finished for PR 7841 at commit 2417548.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Pivot(\n

[rxin]

@aray I talked to a few more people about this. Most like the 2nd API more (groupBy.pivot.agg).

I think it'd also be better to remove the requirement to specify values, e.g. just take in a column without the values. So it looks like

courseSales.groupBy($"year").pivot($"course").agg(sum($"earnings"))

Can you update the pull request? Thanks.

[rxin]

BTW we can also later add a variant that allows users to specify values directly, in order to avoid materializing the intermediate data.

[aray]

@rxin Updated, the values are now optional.

[rxin]

can we remove this?

[aray]

sure

[yhuai]

Why?

[yhuai]

@aray This is very cool! Here are a few things I'd like to discuss.

1. Should we always ask users to provide pivot values?
2. For the type of pivot values, should we just use String? Or, we should use Literal Column?
3. For the column name of a pivot aggregation column, should we always show the aggregate function in a column name? (right now, we only show function name if we have more than 1 aggregations)

[SparkQA]

Test build #45645 has finished for PR 7841 at commit 88dd513.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class SlidingRDDPartition[T](val idx: Int, val prev: Partition, val tail: Seq[T], val offset: Int)\n * class SlidingRDD[T: ClassTag](@transient val parent: RDD[T], val windowSize: Int, val step: Int)\n * case class Pivot(\n

[aray]

@yhuai RE your questions (3 was already addressed above):

1. Should we always ask users to provide pivot values?

The argument for not requiring values I think is convenience and also similarity to other non-sql tools mentioned above. The negative is performance, but since we give them the option to specify I don't think that is a problem.

1. For the type of pivot values, should we just use String? Or, we should use Literal Column?

I initially used strings as the type since that is the common usage scenario. But I agree that using Literal's is the better solution and will avoid casts which could hurt performance. For convenience I kept the second method (changed to pivot(pivotColumn: String, values: Any*)) which I think will be the prefered way to use pivot since it is more succinct.

I really appreciate the review. Let me know if I can do anything else to help!

[yhuai]

Seems we still need to check the number of children and make sure we have a single child?

[aray]

It should now work fine with aggregate functions that have multiple children as long as they ignore updates when all values are null. For example Corr should work since it only updates its aggregation buffer if both its arguments are non null.

[yhuai]

oh, yes. You are right.

[SparkQA]

Test build #45659 has finished for PR 7841 at commit 12a8270.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Pivot(\n

[aray]

@yhuai I think this addresses everything we discussed, let me know if I missed anything or if there is anything else I can do. Again, thanks for the code review.

[yhuai]

LGTM pending jenkins.

[SparkQA]

Test build #45673 has finished for PR 7841 at commit 676f1ac.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class Pivot(\n

[yhuai]

Thanks! Merging to master and branch 1.6.

[rxin]

@aray do you want to submit a pull request for python api too?

[aray]

@rxin sure I'll put together a PR for the python API tonight

[rxin]

@aray this pull request was highlighted in http://www.slideshare.net/databricks/deep-dive-into-catalyst-apache-spark-20s-optimizer

[pushpam002]

thank you