You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by davies <gi...@git.apache.org> on 2014/07/26 03:16:11 UTC
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] support nested st...
GitHub user davies opened a pull request:
https://github.com/apache/spark/pull/1598
[WIP] [SPARK-2010] [PySpark] support nested structure in SchemaRDD
Convert Row in JavaSchemaRDD into Array[Any] and unpickle them as tuple in Python, then convert them into namedtuple when needed.
This will let nested structure can be accessed as object, also it will reduce the size of serialized data and better performance.
PS: The code will be refactored later and more tests will be added.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/davies/spark nested
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1598.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1598
----
commit 644665a2ebae4bc4a49f28152e3af00681affcc8
Author: Davies Liu <da...@gmail.com>
Date: 2014-07-25T23:11:57Z
use tuple and namedtuple for schemardd
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50821078
This patch looks pretty good and contains some great refactorings. I like how you changed the Python data types inherit from a common `DataType` base class; the cleanup in `_parse_datatype_string` is good, too.
I left a couple of comments on parts of the Scala code. My only concern has to do with the detection of batching in pythonToJava. Aside from that, the rest of this PR looks fine to me.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50219202
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17211/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15620793
--- Diff: python/pyspark/sql.py ---
@@ -951,16 +938,39 @@ def count(self):
"""
return self._jschema_rdd.count()
- def _toPython(self):
- # We have to import the Row class explicitly, so that the reference Pickler has is
- # pyspark.sql.Row instead of __main__.Row
- from pyspark.sql import Row
- jrdd = self._jschema_rdd.javaToPython()
- # TODO: This is inefficient, we should construct the Python Row object
- # in Java land in the javaToPython function. May require a custom
- # pickle serializer in Pyrolite
- return RDD(jrdd, self._sc, BatchedSerializer(
- PickleSerializer())).map(lambda d: Row(d))
+ def collect(self):
+ """
+ Return a list that contains all of the rows in this RDD.
+
+ Each object in the list is on Row, the fields can be accessed as
+ attributes.
+ """
+ rows = RDD.collect(self)
+ cls = _create_cls(self.schema())
+ return map(cls, rows)
+
+ # convert Row in JavaSchemaRDD into namedtuple, let access fields easier
--- End diff --
thx
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50843120
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17621/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50681477
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17488/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15669831
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Array[_]] = {
--- End diff --
I'd give this a more descriptive name, like you did for `pythonToJavaMap`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50821976
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17609/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15714554
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---
@@ -413,7 +413,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
/**
* Peek at the first row of the RDD and infer its schema.
- * It is only used by PySpark.
+ * This function is outdated, PySpark does not use it anymore
--- End diff --
Should we just delete this?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15678985
--- Diff: python/pyspark/sql.py ---
@@ -405,70 +359,432 @@ def _parse_datatype_string(datatype_string):
>>> check_datatype(complex_arraytype)
True
>>> # Complex MapType.
- >>> complex_maptype = MapType(complex_structtype, complex_arraytype, False)
+ >>> complex_maptype = MapType(complex_structtype,
+ ... complex_arraytype, False)
>>> check_datatype(complex_maptype)
True
"""
- left_bracket_index = datatype_string.find("(")
- if left_bracket_index == -1:
+ index = datatype_string.find("(")
+ if index == -1:
# It is a primitive type.
- left_bracket_index = len(datatype_string)
- type_or_field = datatype_string[:left_bracket_index]
- rest_part = datatype_string[left_bracket_index+1:len(datatype_string)-1].strip()
- if type_or_field == "StringType":
- return StringType()
- elif type_or_field == "BinaryType":
- return BinaryType()
- elif type_or_field == "BooleanType":
- return BooleanType()
- elif type_or_field == "TimestampType":
- return TimestampType()
- elif type_or_field == "DecimalType":
- return DecimalType()
- elif type_or_field == "DoubleType":
- return DoubleType()
- elif type_or_field == "FloatType":
- return FloatType()
- elif type_or_field == "ByteType":
- return ByteType()
- elif type_or_field == "IntegerType":
- return IntegerType()
- elif type_or_field == "LongType":
- return LongType()
- elif type_or_field == "ShortType":
- return ShortType()
+ index = len(datatype_string)
+ type_or_field = datatype_string[:index]
+ rest_part = datatype_string[index + 1:len(datatype_string) - 1].strip()
+
+ if type_or_field in _all_primitive_types:
+ return _all_primitive_types[type_or_field]()
+
elif type_or_field == "ArrayType":
last_comma_index = rest_part.rfind(",")
containsNull = True
- if rest_part[last_comma_index+1:].strip().lower() == "false":
+ if rest_part[last_comma_index + 1:].strip().lower() == "false":
containsNull = False
- elementType = _parse_datatype_string(rest_part[:last_comma_index].strip())
+ elementType = _parse_datatype_string(
+ rest_part[:last_comma_index].strip())
return ArrayType(elementType, containsNull)
+
elif type_or_field == "MapType":
last_comma_index = rest_part.rfind(",")
valueContainsNull = True
- if rest_part[last_comma_index+1:].strip().lower() == "false":
+ if rest_part[last_comma_index + 1:].strip().lower() == "false":
valueContainsNull = False
- keyType, valueType = _parse_datatype_list(rest_part[:last_comma_index].strip())
+ keyType, valueType = _parse_datatype_list(
+ rest_part[:last_comma_index].strip())
return MapType(keyType, valueType, valueContainsNull)
+
elif type_or_field == "StructField":
first_comma_index = rest_part.find(",")
name = rest_part[:first_comma_index].strip()
last_comma_index = rest_part.rfind(",")
nullable = True
- if rest_part[last_comma_index+1:].strip().lower() == "false":
+ if rest_part[last_comma_index + 1:].strip().lower() == "false":
nullable = False
dataType = _parse_datatype_string(
- rest_part[first_comma_index+1:last_comma_index].strip())
+ rest_part[first_comma_index + 1:last_comma_index].strip())
return StructField(name, dataType, nullable)
+
elif type_or_field == "StructType":
# rest_part should be in the format like
# List(StructField(field1,IntegerType,false)).
- field_list_string = rest_part[rest_part.find("(")+1:-1]
+ field_list_string = rest_part[rest_part.find("(") + 1:-1]
fields = _parse_datatype_list(field_list_string)
return StructType(fields)
+# Mapping Python types to Spark SQL DateType
+_type_mappings = {
+ bool: BooleanType,
+ int: IntegerType,
+ long: LongType,
+ float: DoubleType,
+ str: StringType,
+ unicode: StringType,
+ decimal.Decimal: DecimalType,
+ datetime.datetime: TimestampType,
+ datetime.date: TimestampType,
+ datetime.time: TimestampType,
+}
+
+
+def _infer_type(obj):
+ """Infer the DataType from obj"""
+ if obj is None:
+ raise ValueError("Can not infer type for None")
+
+ dataType = _type_mappings.get(type(obj))
+ if dataType is not None:
+ return dataType()
+
+ if isinstance(obj, dict):
+ if not obj:
+ raise ValueError("Can not infer type for empty dict")
+ key, value = obj.iteritems().next()
+ return MapType(_infer_type(key), _infer_type(value), True)
+ elif isinstance(obj, (list, array.array)):
+ if not obj:
+ raise ValueError("Can not infer type for empty list/array")
+ return ArrayType(_infer_type(obj[0]), True)
+ else:
+ try:
+ return _infer_schema(obj)
+ except ValueError:
+ raise ValueError("not supported type: %s" % type(obj))
+
+
+def _infer_schema(row):
+ """Infer the schema from dict/namedtuple/object"""
+ if isinstance(row, dict):
+ items = sorted(row.items())
+ elif isinstance(row, tuple):
+ if hasattr(row, "_fields"): # namedtuple
+ items = zip(row._fields, tuple(row))
+ elif all(isinstance(x, tuple) and len(x) == 2
+ for x in row):
+ items = row
+ elif hasattr(row, "__dict__"): # object
+ items = sorted(row.__dict__.items())
+ else:
+ raise ValueError("Can not infer schema for type: %s" % type(row))
+
+ fields = [StructField(k, _infer_type(v), True) for k, v in items]
+ return StructType(fields)
+
+
+def _create_converter(obj, dataType):
+ """Create an converter to drop the names of fields in obj """
+ if not _has_struct(dataType):
+ return lambda x: x
+
+ elif isinstance(dataType, ArrayType):
+ conv = _create_converter(obj[0], dataType.elementType)
+ return lambda row: map(conv, row)
+
+ elif isinstance(dataType, MapType):
+ value = obj.values()[0]
+ conv = _create_converter(value, dataType.valueType)
+ return lambda row: dict((k, conv(v)) for k, v in row.iteritems())
+
+ # dataType must be StructType
+ names = [f.name for f in dataType.fields]
+
+ if isinstance(obj, dict):
+ conv = lambda o: tuple(o.get(n) for n in names)
+
+ elif isinstance(obj, tuple):
+ if hasattr(obj, "_fields"): # namedtuple
+ conv = tuple
+ elif all(isinstance(x, tuple) and len(x) == 2
+ for x in obj):
+ conv = lambda o: tuple(v for k, v in o)
+
+ elif hasattr(obj, "__dict__"): # object
+ conv = lambda o: [o.__dict__.get(n, None) for n in names]
+
+ nested = any(_has_struct(f.dataType) for f in dataType.fields)
+ if not nested:
+ return conv
+
+ row = conv(obj)
+ convs = [_create_converter(v, f.dataType)
+ for v, f in zip(row, dataType.fields)]
+
+ def nested_conv(row):
+ return tuple(f(v) for f, v in zip(convs, conv(row)))
+
+ return nested_conv
+
+
+def _dropSchema(rows, schema):
+ """Drop all the names of fields, becoming tuples"""
+ iterator = iter(rows)
+ row = iterator.next()
+ converter = _create_converter(row, schema)
+ yield converter(row)
+ for i in iterator:
+ yield converter(i)
+
+
+_BRAKETS = {'(': ')', '[': ']', '{': '}'}
--- End diff --
Small typo, should be BRACKETS with a C
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15669952
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Array[_]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[_] => Try(objs.map(obj => obj match {
+ case list: JArrayList[_] => list.toArray // list
+ case obj if obj.getClass.isArray => // tuple
+ obj.asInstanceOf[Array[_]].toArray
+ })) match {
--- End diff --
cool!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15714150
--- Diff: python/pyspark/sql.py ---
@@ -522,71 +866,112 @@ def _ssql_ctx(self):
return self._scala_SQLContext
def inferSchema(self, rdd):
- """Infer and apply a schema to an RDD of L{dict}s.
+ """Infer and apply a schema to an RDD of L{Row}s.
+
+ We peek at the first row of the RDD to determine the fields' names
+ and types. Nested collections are supported, which include array,
+ dict, list, Row, tuple, namedtuple, or object.
--- End diff --
Can we add a note about the complications with us only looking at the first row. In particular, the type can't change in future rows and this breaks if there are `None` in the first row. Also do we throw an exceptions for `None`s?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50717628
This feature will slow down normal access (by attribute or position), so I did not put it in. User still can use position to access the field with special names (such as keywords).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15620443
--- Diff: python/pyspark/sql.py ---
@@ -634,20 +645,18 @@ def jsonFile(self, path, schema=None):
>>> sqlCtx.registerRDDAsTable(srdd1, "table1")
>>> srdd2 = sqlCtx.sql(
... "SELECT field1 AS f1, field2 as f2, field3 as f3, field6 as f4 from table1")
- >>> srdd2.collect() == [
- ... {"f1":1, "f2":"row1", "f3":{"field4":11, "field5": None}, "f4":None},
- ... {"f1":2, "f2":None, "f3":{"field4":22, "field5": [10, 11]}, "f4":[{"field7": "row2"}]},
- ... {"f1":None, "f2":"row3", "f3":{"field4":33, "field5": []}, "f4":None}]
- True
+ >>> srdd2.collect()
+ [Row(f1=1, f2=u'row1', f3=Row(field4=11, field5=None), f4=None), \
+Row(f1=2, f2=None, f3=Row(field4=22, field5=[10, 11]), f4=[Row(field7=u'row2')]), \
+Row(f1=None, f2=u'row3', f3=Row(field4=33, field5=[]), f4=None)]
--- End diff --
good idea!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50944097
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17700/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] [SQL] support nes...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50561427
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17399/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50840172
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17616/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50818803
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17600/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50828772
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17609/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50703903
QA results for PR 1598:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17525/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50716626
Sure, sounds good.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15620816
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
+ if _has_struct(_f.dataType):
+ # delay creating object until accessing it
+ _getter = _create_getter(_f.dataType, _i)
+ else:
+ _getter = itemgetter(_i)
+ locals()[_f.name] = property(_getter)
+ del _i, _f, _getter
+
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return ("Row(%s)" % ", ".join("%s=%r" % (n, getattr(self, n))
+ for n in self._fields))
+ def __reduce__(self):
+ # pickle as namedtuple
+ return (_restore_object, ("Row", self._fields, tuple(self)))
--- End diff --
This is not entirely true. You can access columns that are named with reserve words you just have to use backticks. The SQL 92 standards does disallow somethings, for example starting an identifier with a number in most character sets.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15670058
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Array[_]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[_] => Try(objs.map(obj => obj match {
--- End diff --
Does this behave properly in the unbatched case? In general, I don't think that it's safe to detect batching by checking whether the returned item is a list, since that might inadvertently flatten a list-of-lists that was serialized without batching.
(see https://github.com/apache/spark/pull/1338#discussion-diff-15508664)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] [SQL] support nes...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50559104
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17397/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616522
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
--- End diff --
Also since this is a class property and a constant, I'd call it `_FIELDS` instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50689024
Made some comments on it from the Python side. @JoshRosen you may also want to take a look at the named tuple / class generation stuff here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50221527
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17211/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616738
--- Diff: python/pyspark/sql.py ---
@@ -951,16 +938,39 @@ def count(self):
"""
return self._jschema_rdd.count()
- def _toPython(self):
- # We have to import the Row class explicitly, so that the reference Pickler has is
- # pyspark.sql.Row instead of __main__.Row
- from pyspark.sql import Row
- jrdd = self._jschema_rdd.javaToPython()
- # TODO: This is inefficient, we should construct the Python Row object
- # in Java land in the javaToPython function. May require a custom
- # pickle serializer in Pyrolite
- return RDD(jrdd, self._sc, BatchedSerializer(
- PickleSerializer())).map(lambda d: Row(d))
+ def collect(self):
+ """
+ Return a list that contains all of the rows in this RDD.
+
+ Each object in the list is on Row, the fields can be accessed as
+ attributes.
+ """
+ rows = RDD.collect(self)
+ cls = _create_cls(self.schema())
+ return map(cls, rows)
+
+ # convert Row in JavaSchemaRDD into namedtuple, let access fields easier
--- End diff --
You should expand this comment a bit, e.g. "Convert each object in the RDD to a Row with the right class for this SchemaRDD, so that fields can be accessed as attributes." Also this needs to appear in some kind of class comment at the top, e.g. say "This class receives raw tuples from Java but assigns a class to it in all its data-collection methods (mapPartitionsWithIndex, collect, take, etc) so that PySpark sees them as Row objects with named fields".
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15713885
--- Diff: python/pyspark/sql.py ---
@@ -494,18 +838,18 @@ def __init__(self, sparkContext, sqlContext=None):
ValueError:...
>>> from datetime import datetime
- >>> allTypes = sc.parallelize([{"int": 1, "string": "string", "double": 1.0, "long": 1L,
- ... "boolean": True, "time": datetime(2010, 1, 1, 1, 1, 1), "dict": {"a": 1},
- ... "list": [1, 2, 3]}])
- >>> srdd = sqlCtx.inferSchema(allTypes).map(lambda x: (x.int, x.string, x.double, x.long,
- ... x.boolean, x.time, x.dict["a"], x.list))
+ >>> allTypes = sc.parallelize([Row(int=1, string="string",
+ ... double=1.0, long=1L, boolean=True, list=[1, 2, 3],
+ ... time=datetime(2010, 1, 1, 1, 1, 1), dict={"a": 1})])
+ >>> srdd = sqlCtx.inferSchema(allTypes).map(lambda x: (x.int, x.string,
+ ... x.double, x.long, x.boolean, x.time, x.dict["a"], x.list))
--- End diff --
It would be great to also add a SQL test here to make sure that types are matching up with those expected in the execution engine. (though we might change the names to avoid conflict with reserved words, as we have not implemented identifier escaping). In particular the complex nested ones like dict and list. Also it would be good to add a nested Row to the input types.
Something like:
```python
srdd.registerAsTable("pythonData")
sqlCtx.sql("SELECT dict['a'], list[0], nested.nestedField").collect() ...
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15714321
--- Diff: python/pyspark/sql.py ---
@@ -522,71 +866,112 @@ def _ssql_ctx(self):
return self._scala_SQLContext
def inferSchema(self, rdd):
- """Infer and apply a schema to an RDD of L{dict}s.
+ """Infer and apply a schema to an RDD of L{Row}s.
+
+ We peek at the first row of the RDD to determine the fields' names
+ and types. Nested collections are supported, which include array,
+ dict, list, Row, tuple, namedtuple, or object.
- We peek at the first row of the RDD to determine the fields names
- and types, and then use that to extract all the dictionaries. Nested
- collections are supported, which include array, dict, list, set, and
- tuple.
+ Each row in `rdd` should be Row object or namedtuple or objects,
+ using dict is deprecated.
+ >>> rdd = sc.parallelize(
+ ... [Row(field1=1, field2="row1"),
+ ... Row(field1=2, field2="row2"),
+ ... Row(field1=3, field2="row3")])
>>> srdd = sqlCtx.inferSchema(rdd)
- >>> srdd.collect() == [{"field1" : 1, "field2" : "row1"}, {"field1" : 2, "field2": "row2"},
- ... {"field1" : 3, "field2": "row3"}]
- True
+ >>> srdd.collect()[0]
+ Row(field1=1, field2=u'row1')
- >>> from array import array
+ >>> NestedRow = Row("f1", "f2")
+ >>> nestedRdd1 = sc.parallelize([
+ ... NestedRow(array('i', [1, 2]), {"row1": 1.0}),
+ ... NestedRow(array('i', [2, 3]), {"row2": 2.0})])
>>> srdd = sqlCtx.inferSchema(nestedRdd1)
- >>> srdd.collect() == [{"f1" : [1, 2], "f2" : {"row1" : 1.0}},
- ... {"f1" : [2, 3], "f2" : {"row2" : 2.0}}]
- True
+ >>> srdd.collect()
+ [Row(f1=[1, 2], f2={u'row1': 1.0}), ..., f2={u'row2': 2.0})]
+ >>> nestedRdd2 = sc.parallelize([
+ ... NestedRow([[1, 2], [2, 3]], [1, 2]),
+ ... NestedRow([[2, 3], [3, 4]], [2, 3])])
>>> srdd = sqlCtx.inferSchema(nestedRdd2)
- >>> srdd.collect() == [{"f1" : [[1, 2], [2, 3]], "f2" : [1, 2]},
- ... {"f1" : [[2, 3], [3, 4]], "f2" : [2, 3]}]
- True
+ >>> srdd.collect()
+ [Row(f1=[[1, 2], [2, 3]], f2=[1, 2]), ..., f2=[2, 3])]
"""
- if (rdd.__class__ is SchemaRDD):
- raise ValueError("Cannot apply schema to %s" % SchemaRDD.__name__)
- elif not isinstance(rdd.first(), dict):
- raise ValueError("Only RDDs with dictionaries can be converted to %s: %s" %
- (SchemaRDD.__name__, rdd.first()))
- jrdd = self._pythonToJavaMap(rdd._jrdd)
- srdd = self._ssql_ctx.inferSchema(jrdd.rdd())
- return SchemaRDD(srdd, self)
+ if isinstance(rdd, SchemaRDD):
+ raise TypeError("Cannot apply schema to SchemaRDD")
+
+ first = rdd.first()
+ if not first:
+ raise ValueError("The first row in RDD is empty, "
+ "can not infer schema")
+ if type(first) is dict:
+ warnings.warn("Using RDD of dict to inferSchema is deprecated")
+
+ schema = _infer_schema(first)
+ rdd = rdd.mapPartitions(lambda rows: _drop_schema(rows, schema))
+ return self.applySchema(rdd, schema)
def applySchema(self, rdd, schema):
- """Applies the given schema to the given RDD of L{dict}s.
+ """
+ Applies the given schema to the given RDD of L{tuple} or L{list}s.
--- End diff --
... These tuples or lists can contain complex nested structures like lists maps for nested rows. It is important that the schema matches the types of the objects in each row or exceptions could be thrown at runtime.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50909213
@marmbrus I had add the interface just as you commented, also update the description in this PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50854552
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17648/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50925845
This is looking really good to me! I'm very excited to have much more complete support for SQL in pyspark. A few minor comments on docs and testing, but I think we can merge this soon.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15620424
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
+ if _has_struct(_f.dataType):
+ # delay creating object until accessing it
+ _getter = _create_getter(_f.dataType, _i)
+ else:
+ _getter = itemgetter(_i)
+ locals()[_f.name] = property(_getter)
+ del _i, _f, _getter
+
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return ("Row(%s)" % ", ".join("%s=%r" % (n, getattr(self, n))
+ for n in self._fields))
+ def __reduce__(self):
+ # pickle as namedtuple
+ return (_restore_object, ("Row", self._fields, tuple(self)))
--- End diff --
I think we should have some limitation about name of fields, they could not be anything. We already have this limit in SQL parser, the name of field can not be key words, such `group`.
How to address these, check them with an black list and raise exception?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15713580
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJavaArray(pyRDD: JavaRDD[Array[Byte]], batched: Boolean): JavaRDD[Array[_]] = {
--- End diff --
The whole PythonRDD is private, so does it still need this?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15717487
--- Diff: python/pyspark/sql.py ---
@@ -522,71 +866,112 @@ def _ssql_ctx(self):
return self._scala_SQLContext
def inferSchema(self, rdd):
- """Infer and apply a schema to an RDD of L{dict}s.
+ """Infer and apply a schema to an RDD of L{Row}s.
+
+ We peek at the first row of the RDD to determine the fields' names
+ and types. Nested collections are supported, which include array,
+ dict, list, Row, tuple, namedtuple, or object.
--- End diff --
It will throw exceptions if there is any empty value (None, [] or {}) in the first row.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50705120
(Actually it might also work and just not let the user reference that field. I guess that would be okay from our point of view -- they'd quickly notice the problem.)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50714300
@mateiz We could also provide an dict interface as an fallback, so user can access these fields just like dict, such as row["pass"] ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15620926
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
+ if _has_struct(_f.dataType):
+ # delay creating object until accessing it
+ _getter = _create_getter(_f.dataType, _i)
+ else:
+ _getter = itemgetter(_i)
+ locals()[_f.name] = property(_getter)
+ del _i, _f, _getter
+
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return ("Row(%s)" % ", ".join("%s=%r" % (n, getattr(self, n))
+ for n in self._fields))
+ def __reduce__(self):
+ # pickle as namedtuple
+ return (_restore_object, ("Row", self._fields, tuple(self)))
--- End diff --
`Row` can not be accessed directly here, only by self.__class__, maybe self._FIELDS will be better?
btw, namedtuple uses `_fields` for names, namedtuple does not support name startswith "_", so we can not use namedtuple?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15714585
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
--- End diff --
Sorry, this was never public API maybe we should just delete it instead of deprecating it?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616339
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
--- End diff --
Comma after ","
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50681922
@yhuai @marmbrus @mateiz plz take a look at this, thx!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15712557
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJavaArray(pyRDD: JavaRDD[Array[Byte]], batched: Boolean): JavaRDD[Array[_]] = {
--- End diff --
`private[spark]`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by yhuai <gi...@git.apache.org>.
Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15714998
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---
@@ -503,12 +504,9 @@ class SQLContext(@transient val sparkContext: SparkContext)
}
row
- case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) =>
- val converted = c.map {
- case (key, value) =>
- (convert(key, keyType), convert(value, valueType))
- }
- JMapWrapper(converted)
+ case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) => c.map {
+ case (key, value) => (convert(key, keyType), convert(value, valueType))
+ }.toMap
--- End diff --
Will `case (c: java.util.Map[_, _], struct: StructType)` happen with your change? How do we handle inner structs?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] [SQL] support nes...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50435088
A StructType is presented as an namedtuple in Python, which is called Row.
The Row is generated according schema, there is no predefined Row class, so it's better to keep it internal.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616509
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
+ if _has_struct(_f.dataType):
+ # delay creating object until accessing it
+ _getter = _create_getter(_f.dataType, _i)
+ else:
+ _getter = itemgetter(_i)
+ locals()[_f.name] = property(_getter)
+ del _i, _f, _getter
+
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return ("Row(%s)" % ", ".join("%s=%r" % (n, getattr(self, n))
+ for n in self._fields))
+ def __reduce__(self):
+ # pickle as namedtuple
+ return (_restore_object, ("Row", self._fields, tuple(self)))
--- End diff --
It might be clearer to write `Row._fields` instead of `self._fields` here, otherwise it seems like a getter for a field called "_fields". Actually what happens if the table has a column named "_field"? It might be broken then. Add a test for that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616211
--- Diff: python/pyspark/sql.py ---
@@ -634,20 +645,18 @@ def jsonFile(self, path, schema=None):
>>> sqlCtx.registerRDDAsTable(srdd1, "table1")
>>> srdd2 = sqlCtx.sql(
... "SELECT field1 AS f1, field2 as f2, field3 as f3, field6 as f4 from table1")
- >>> srdd2.collect() == [
- ... {"f1":1, "f2":"row1", "f3":{"field4":11, "field5": None}, "f4":None},
- ... {"f1":2, "f2":None, "f3":{"field4":22, "field5": [10, 11]}, "f4":[{"field7": "row2"}]},
- ... {"f1":None, "f2":"row3", "f3":{"field4":33, "field5": []}, "f4":None}]
- True
+ >>> srdd2.collect()
+ [Row(f1=1, f2=u'row1', f3=Row(field4=11, field5=None), f4=None), \
+Row(f1=2, f2=None, f3=Row(field4=22, field5=[10, 11]), f4=[Row(field7=u'row2')]), \
+Row(f1=None, f2=u'row3', f3=Row(field4=33, field5=[]), f4=None)]
--- End diff --
Breaking the doc comment like this is kind of weird; could you instead do a `for r in srdd2.collect(): print r` and get one per line?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15670161
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
--- End diff --
In this case, it's actually fine to check for a list in order to detect batching because we're not expecting to find an RDD of lists.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50221066
Can you add [SQL] to these PRs as well?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15713728
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJavaArray(pyRDD: JavaRDD[Array[Byte]], batched: Boolean): JavaRDD[Array[_]] = {
--- End diff --
Ah, I did not realize that. It could still perhaps be marked protected (to prevent other spark users from depending on it directly), but thats not as big of a deal.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] [SQL] support nes...
Posted by yhuai <gi...@git.apache.org>.
Github user yhuai commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50433935
With this PR, what does a `StructType` represent? namedtuple or array? Do we still keep the Row class in PySpark?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15669918
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Array[_]] = {
--- End diff --
In future, we will remove pythonToJavaMap, so I think pythonToJava() will make sense.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15669434
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Array[_]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[_] => Try(objs.map(obj => obj match {
--- End diff --
I think the nesting / indentation here makes this code hard to understand. I'd move the Try down to the next line and ensure that everything that falls under this `case` is indented at least two spaces underneath it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15712514
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
--- End diff --
If this function is outdated should we mark it `@deprecated`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50717656
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17563/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50950775
Thanks for working on this! I've merge it to master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50699862
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17525/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15712536
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
--- End diff --
Perhaps it is too late if this was already public API... but should this be marked `private[spark]`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50809839
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17600/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50821872
@JoshRosen thanks for the review, I had addressed all your comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50945446
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17703/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50822091
The failed test is not related to his PR. it will test again.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15620941
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
+ if _has_struct(_f.dataType):
+ # delay creating object until accessing it
+ _getter = _create_getter(_f.dataType, _i)
+ else:
+ _getter = itemgetter(_i)
+ locals()[_f.name] = property(_getter)
+ del _i, _f, _getter
+
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return ("Row(%s)" % ", ".join("%s=%r" % (n, getattr(self, n))
+ for n in self._fields))
+ def __reduce__(self):
+ # pickle as namedtuple
+ return (_restore_object, ("Row", self._fields, tuple(self)))
--- End diff --
On the Python side, I'd skip this field if the name starts with `__` and ends with `__`. This should be enough to cover the Python reserved methods without us having to create a list. Separately, in this code, we can rename the local variables to `__i__` and `__f__`, or maybe you can move this code into a def() and call that so they're local variables.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] [SQL] support nes...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50564440
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17397/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50850668
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17648/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50688906
QA results for PR 1598:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17488/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50940004
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17703/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50859069
@marmbrus @mateiz @JoshRosen Could you take another look as this? I had removed schema as string, and added Row as the suggested type to infer schema.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50835529
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17616/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50698835
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17519/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616605
--- Diff: python/pyspark/sql.py ---
@@ -467,6 +380,108 @@ def _parse_datatype_string(datatype_string):
return StructType(fields)
+
+_cached_namedtuples = {}
+
+def _restore_object(name, fields, obj):
+ """ Restore namedtuple object during unpickling. """
+ cls = _cached_namedtuples.get(fields)
+ if cls is None:
+ cls = namedtuple(name, fields)
+ def __reduce__(self):
+ return (_restore_object, (name, fields, tuple(self)))
+ cls.__reduce__ = __reduce__
+ _cached_namedtuples[fields] = cls
+ return cls(*obj)
+
+def _create_object(cls, v):
+ """ Create an customized object with class `cls`. """
+ return cls(v) if v is not None else v
+
+def _create_getter(dt, i):
+ """ Create a getter for item `i` with schema """
+ cls = _create_cls(dt)
+ def getter(self):
+ return _create_object(cls, self[i])
+ return getter
+
+def _has_struct(dt):
+ """Return whether `dt` is or has StructType in it"""
+ if isinstance(dt, StructType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _has_struct(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _has_struct(dt.valueType)
+ return False
+
+def _create_cls(dataType):
+ """
+ Create an class by dataType
+
+ The created class is similar to namedtuple, but can have nested schema.
+ """
+ from operator import itemgetter
+
+ if isinstance(dataType, ArrayType):
+ cls = _create_cls(dataType.elementType)
+ class List(list):
+ def __getitem__(self, i):
+ # create object with datetype
+ return _create_object(cls, list.__getitem__(self, i))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "[%s]" % (", ".join(repr(self[i])
+ for i in range(len(self))))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (list, (list(self),))
+ return List
+
+ elif isinstance(dataType, MapType):
+ vcls = _create_cls(dataType.valueType)
+ class Dict(dict):
+ def __getitem__(self, k):
+ # create object with datetype
+ return _create_object(vcls, dict.__getitem__(self, k))
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return "{%s}" % (", ".join("%r: %r" % (k, self[k])
+ for k in self))
+ def __reduce__(self):
+ # pickle as dict, the nested struct can be reduced by itself
+ return (dict, (dict(self),))
+ return Dict
+
+ elif not isinstance(dataType, StructType):
+ raise Exception("unexpected data type: %s" % dataType)
+
+ class Row(tuple):
+ """ Row in SchemaRDD """
+ _fields = tuple(f.name for f in dataType.fields)
+
+ # create property for fast access
+ # use local vars begins with "_"
+ for _i,_f in enumerate(dataType.fields):
+ if _has_struct(_f.dataType):
+ # delay creating object until accessing it
+ _getter = _create_getter(_f.dataType, _i)
+ else:
+ _getter = itemgetter(_i)
+ locals()[_f.name] = property(_getter)
+ del _i, _f, _getter
+
+ def __repr__(self):
+ # call collect __repr__ for nested objects
+ return ("Row(%s)" % ", ".join("%s=%r" % (n, getattr(self, n))
+ for n in self._fields))
+ def __reduce__(self):
+ # pickle as namedtuple
+ return (_restore_object, ("Row", self._fields, tuple(self)))
--- End diff --
Same thing if a table has fields called `__repr__`, `__str__`, etc. Is there any way we can check for these? I think we should just not expose those to Python (just skip them in the call to `locals()` above).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15669681
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala ---
@@ -701,6 +682,53 @@ private[spark] object PythonRDD extends Logging {
}
}
+
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
+ * This function is outdated, PySpark does not use it anymore
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // not in batch mode
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple to Array (no recursive conversions).
+ * It is only used by pyspark.sql.
+ */
+ def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Array[_]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // in case of objects are pickled in batch mode
+ case objs: JArrayList[_] => Try(objs.map(obj => obj match {
+ case list: JArrayList[_] => list.toArray // list
+ case obj if obj.getClass.isArray => // tuple
+ obj.asInstanceOf[Array[_]].toArray
+ })) match {
--- End diff --
Instead of using this match, you could also do `.getOrElse(Seq(objs.toArray))`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50703165
QA results for PR 1598:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17519/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50839012
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17620/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50723544
Ah okay, then I'm fine leaving that out.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50841921
If we are going to create our own function that builds rows instead of using named tuples (which is increasingly growing on me...) Perhaps we could use `**kwargs`?
I'm thinking something like `inferSchema(rdd.map(lambda x: row(name = x[1], age = x[2]))`
Just a thought... that seems lightweight but doesn't suffer the semantic ambiguities inherit with dictionaries.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50839289
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17621/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1598
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by yhuai <gi...@git.apache.org>.
Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15618206
--- Diff: python/pyspark/sql.py ---
@@ -273,21 +234,11 @@ def __init__(self, name, dataType, nullable):
self.nullable = nullable
def __repr__(self):
- return "StructField(" + self.name + "," + \
- self.dataType.__repr__() + "," + \
- str(self.nullable).lower() + ")"
+ return "StructField(%s,%r,%s)" % (self.name, self.dataType,
+ str(self.nullable).lower())
- def __eq__(self, other):
- return (isinstance(other, self.__class__) and
- self.name == other.name and
- self.dataType == other.dataType and
- self.nullable == other.nullable)
-
- def __ne__(self, other):
- return not self.__eq__(other)
-
-class StructType(object):
+class StructType(DataType):
"""Spark SQL StructType
The data type representing namedtuple values.
--- End diff --
Should we change it to `The data type representing rows.`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15616248
--- Diff: python/pyspark/sql.py ---
@@ -795,6 +795,7 @@ class LocalHiveContext(HiveContext):
An in-process metadata data is created with data stored in ./metadata.
Warehouse data is stored in in ./warehouse.
+ disable these tests tempory
--- End diff --
What does this mean?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50842346
QA results for PR 1598:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17620/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15717851
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---
@@ -503,12 +504,9 @@ class SQLContext(@transient val sparkContext: SparkContext)
}
row
- case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) =>
- val converted = c.map {
- case (key, value) =>
- (convert(key, keyType), convert(value, valueType))
- }
- JMapWrapper(converted)
+ case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) => c.map {
+ case (key, value) => (convert(key, keyType), convert(value, valueType))
+ }.toMap
--- End diff --
The Row() in python will be convert into tuple(), so It's fine to remove this case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50911295
I had created an issue to track schema as String: https://issues.apache.org/jira/browse/SPARK-2789
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50726056
QA results for PR 1598:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17563/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [WIP] [SPARK-2010] [PySpark] [SQL] support nes...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50566103
QA results for PR 1598:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17399/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50813384
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17603/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50936511
QA tests have started for PR 1598. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17700/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by yhuai <gi...@git.apache.org>.
Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/1598#discussion_r15714858
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---
@@ -503,12 +504,9 @@ class SQLContext(@transient val sparkContext: SparkContext)
}
row
- case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) =>
- val converted = c.map {
- case (key, value) =>
- (convert(key, keyType), convert(value, valueType))
- }
- JMapWrapper(converted)
+ case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) => c.map {
+ case (key, value) => (convert(key, keyType), convert(value, valueType))
+ }.toMap
--- End diff --
Should we update the part of `case (c: java.util.Map[_, _], struct: StructType) ` as well?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50705095
@davies actually regarding the field names, apparently you also can't use Python keywords. I'm not sure how to get those programmatically but you can find a list at https://docs.python.org/2.5/ref/keywords.html.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2010] [PySpark] [SQL] support nested st...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1598#issuecomment-50821816
QA results for PR 1598:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>class DataType(object):<br>class PrimitiveType(DataType):<br>class StringType(PrimitiveType):<br>class BinaryType(PrimitiveType):<br>class BooleanType(PrimitiveType):<br>class TimestampType(PrimitiveType):<br>class DecimalType(PrimitiveType):<br>class DoubleType(PrimitiveType):<br>class FloatType(PrimitiveType):<br>class ByteType(PrimitiveType):<br>class IntegerType(PrimitiveType):<br>class LongType(PrimitiveType):<br>class ShortType(PrimitiveType):<br>class ArrayType(DataType):<br>class MapType(DataType):<br>class StructField(DataType):<br>class StructType(DataType):<br>class List(list):<br>class Dict(dict):<br>class Row(tuple):<br>class Row(tuple):<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17603/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---