[SPARK-2010] Support for nested data in PySpark SQL

[kanzhang]

JIRA issue https://issues.apache.org/jira/browse/SPARK-2010

This PR adds support for nested collection types in PySpark SQL, including
array, dict, list, set, and tuple. Example,

>>> from array import array
>>> from pyspark.sql import SQLContext
>>> sqlCtx = SQLContext(sc)
>>> rdd = sc.parallelize([
...         {"f1" : array('i', [1, 2]), "f2" : {"row1" : 1.0}},
...         {"f1" : array('i', [2, 3]), "f2" : {"row2" : 2.0}}])
>>> srdd = sqlCtx.inferSchema(rdd)
>>> srdd.collect() == [{"f1" : array('i', [1, 2]), "f2" : {"row1" : 1.0}},
...                    {"f1" : array('i', [2, 3]), "f2" : {"row2" : 2.0}}]
True
>>> rdd = sc.parallelize([
...         {"f1" : [[1, 2], [2, 3]], "f2" : set([1, 2]), "f3" : (1, 2)},
...         {"f1" : [[2, 3], [3, 4]], "f2" : set([2, 3]), "f3" : (2, 3)}])
>>> srdd = sqlCtx.inferSchema(rdd)
>>> srdd.collect() == \
... [{"f1" : [[1, 2], [2, 3]], "f2" : set([1, 2]), "f3" : (1, 2)},
...  {"f1" : [[2, 3], [3, 4]], "f2" : set([2, 3]), "f3" : (2, 3)}]
True

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15651/

[kanzhang]

@pwendell the failure doesn't seem to be related to this patch. Could we re-test?

[aarondav]

Jenkins, retest this please.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15660/

[ahirreddy]

Once it passes Jenkins, LGTM

[ahirreddy]

I think you have to rebase against master, it's including extra commits that you probably don't want in jenkins:

[SPARK-2010] Support for nested data in PySpark SQL (commit: 95689830ed50819bfcbaf6cf9697c2710d7b457a) (detail)
[SPARK-1998] SparkFlumeEvent with body bigger than 1020 bytes are not re... (commit: 2966044) (detail)

https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15660/

[marmbrus]

So I think a key question here is if nested dictionaries are Maps or Structs. Right now the outermost dict is treated like a struct with named attributes, so I'm kind of inclined to continue that trend. Although, I think then we would want some way to create a map. What do you think?

[kanzhang]

I haven't looked too deeply, my initial understanding is that Structs are for user-defined Objects (if we are going to support them in Python), while MapType should suffice if we access dictionaries only as maps. Specifically,

1. Struct stores its fields in order, which makes it usable as schema for the outermost dict, as we need to match schema with data stored in an array for each row. Speaking of ordering, how can we be sure when we map dictionary values to an array, they are all in the same order?

    val rowRdd = rdd.mapPartitions { iter =>
      iter.map { map =>
        new GenericRow(map.values.toArray.asInstanceOf[Array[Any]]): Row
      }
    }

2. Structs can only be used for dicts whose the keys are Strings, right?

3. I don't know enough about access pattern to tell if it is safe to label Set as ArrayType since Set is unordered?

[kanzhang]

@ahirreddy thanks, rebased just now.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15787/

[rxin]

can we move this line before the previous unit test block?

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[kanzhang]

@rxin updated patch based on your review comments. Thx!

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15803/

[rxin]

Thanks. I've merged this in master & branch-1.0.