[SPARK-2010] [PySpark] [SQL] support nested structure in SchemaRDD

Convert Row in JavaSchemaRDD into Array[Any] and unpickle them as tuple in Python, then convert them into namedtuple, so use can access fields just like attributes.

This will let nested structure can be accessed as object, also it will reduce the size of serialized data and better performance.

root
 |-- field1: integer (nullable = true)
 |-- field2: string (nullable = true)
 |-- field3: struct (nullable = true)
 |    |-- field4: integer (nullable = true)
 |    |-- field5: array (nullable = true)
 |    |    |-- element: integer (containsNull = false)
 |-- field6: array (nullable = true)
 |    |-- element: struct (containsNull = false)
 |    |    |-- field7: string (nullable = true)

Then we can access them by row.field3.field5[0]  or row.field6[5].field7

It also will infer the schema in Python, convert Row/dict/namedtuple/objects into tuple before serialization, then call applySchema in JVM. During inferSchema(), the top level of dict in row will be StructType, but any nested dictionary will be MapType.

You can use pyspark.sql.Row to convert unnamed structure into Row object, make the RDD can be inferable. Such as:

ctx.inferSchema(rdd.map(lambda x: Row(a=x[0], b=x[1]))

Or you could use Row to create a class just like namedtuple, for example:

Person = Row("name", "age")
ctx.inferSchema(rdd.map(lambda x: Person(*x)))

Also, you can call applySchema to apply an schema to a RDD of tuple/list and turn it into a SchemaRDD. The `schema` should be StructType, see the API docs for details.

schema = StructType([StructField("name, StringType, True),
                                    StructType("age", IntegerType, True)])
ctx.applySchema(rdd, schema)

PS: In order to use namedtuple to inferSchema, you should make namedtuple picklable.

Author: Davies Liu <davies.liu@gmail.com>

Closes #1598 from davies/nested and squashes the following commits:

f1d15b6 [Davies Liu] verify schema with the first few rows
8852aaf [Davies Liu] check type of schema
abe9e6e [Davies Liu] address comments
61b2292 [Davies Liu] add @deprecated to pythonToJavaMap
1e5b801 [Davies Liu] improve cache of classes
51aa135 [Davies Liu] use Row to infer schema
e9c0d5c [Davies Liu] remove string typed schema
353a3f2 [Davies Liu] fix code style
63de8f8 [Davies Liu] fix typo
c79ca67 [Davies Liu] fix serialization of nested data
6b258b5 [Davies Liu] fix pep8
9d8447c [Davies Liu] apply schema provided by string of names
f5df97f [Davies Liu] refactor, address comments
9d9af55 [Davies Liu] use arrry to applySchema and infer schema in Python
84679b3 [Davies Liu] Merge branch 'master' of github.com:apache/spark into nested
0eaaf56 [Davies Liu] fix doc tests
b3559b4 [Davies Liu] use generated Row instead of namedtuple
c4ddc30 [Davies Liu] fix conflict between name of fields and variables
7f6f251 [Davies Liu] address all comments
d69d397 [Davies Liu] refactor
2cc2d45 [Davies Liu] refactor
182fb46 [Davies Liu] refactor
bc6e9e1 [Davies Liu] switch to new Schema API
547bf3e [Davies Liu] Merge branch 'master' into nested
a435b5a [Davies Liu] add docs and code refactor
2c8debc [Davies Liu] Merge branch 'master' into nested
644665a [Davies Liu] use tuple and namedtuple for schemardd

5 files changed, 996 insertions, 444 deletions

diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala b/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
index 94d666aa92..fe9a9e50ef 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
@@ -25,7 +25,7 @@ import java.util.{List => JList, ArrayList => JArrayList, Map => JMap, Collectio
 import scala.collection.JavaConversions._
 import scala.language.existentials
 import scala.reflect.ClassTag
-import scala.util.Try
+import scala.util.{Try, Success, Failure}
 
 import net.razorvine.pickle.{Pickler, Unpickler}
 
@@ -536,25 +536,6 @@ private[spark] object PythonRDD extends Logging {
     file.close()
   }
 
-  /**
-   * Convert an RDD of serialized Python dictionaries to Scala Maps (no recursive conversions).
-   * It is only used by pyspark.sql.
-   * TODO: Support more Python types.
-   */
-  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: java.util.ArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
-          // not in batch mode
-          case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
-        }
-      }
-    }
-  }
-
   private def getMergedConf(confAsMap: java.util.HashMap[String, String],
       baseConf: Configuration): Configuration = {
     val conf = PythonHadoopUtil.mapToConf(confAsMap)
@@ -701,6 +682,54 @@ 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
+   */
+  @deprecated
+  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[_]] = {
+
+    def toArray(obj: Any): Array[_] = {
+      obj match {
+        case objs: JArrayList[_] =>
+          objs.toArray
+        case obj if obj.getClass.isArray =>
+          obj.asInstanceOf[Array[_]].toArray
+      }
+    }
+
+    pyRDD.rdd.mapPartitions { iter =>
+      val unpickle = new Unpickler
+      iter.flatMap { row =>
+        val obj = unpickle.loads(row)
+        if (batched) {
+          obj.asInstanceOf[JArrayList[_]].map(toArray)
+        } else {
+          Seq(toArray(obj))
+        }
+      }
+    }.toJavaRDD()
+  }
+
   /**
    * Convert and RDD of Java objects to and RDD of serialized Python objects, that is usable by
    * PySpark.
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index e8fcc900ef..309f5a9b60 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -318,9 +318,9 @@ class RDD(object):
         >>> sorted(rdd.map(lambda x: (x, 1)).collect())
         [('a', 1), ('b', 1), ('c', 1)]
         """
-        def func(split, iterator):
+        def func(_, iterator):
             return imap(f, iterator)
-        return PipelinedRDD(self, func, preservesPartitioning)
+        return self.mapPartitionsWithIndex(func, preservesPartitioning)
 
     def flatMap(self, f, preservesPartitioning=False):
         """
@@ -1184,7 +1184,7 @@ class RDD(object):
                 if not isinstance(x, basestring):
                     x = unicode(x)
                 yield x.encode("utf-8")
-        keyed = PipelinedRDD(self, func)
+        keyed = self.mapPartitionsWithIndex(func)
         keyed._bypass_serializer = True
         keyed._jrdd.map(self.ctx._jvm.BytesToString()).saveAsTextFile(path)
 
@@ -1382,7 +1382,7 @@ class RDD(object):
                 yield pack_long(split)
                 yield outputSerializer.dumps(items)
 
-        keyed = PipelinedRDD(self, add_shuffle_key)
+        keyed = self.mapPartitionsWithIndex(add_shuffle_key)
         keyed._bypass_serializer = True
         with _JavaStackTrace(self.context) as st:
             pairRDD = self.ctx._jvm.PairwiseRDD(
diff --git a/python/pyspark/sql.py b/python/pyspark/sql.py
index 9388ead5ea..f840475ffa 100644
--- a/python/pyspark/sql.py
+++ b/python/pyspark/sql.py
@@ -15,7 +15,17 @@
 # limitations under the License.
 #
 
+
+import sys
+import types
+import itertools
+import warnings
+import decimal
+import datetime
+import keyword
 import warnings
+from array import array
+from operator import itemgetter
 
 from pyspark.rdd import RDD, PipelinedRDD
 from pyspark.serializers import BatchedSerializer, PickleSerializer
@@ -26,10 +36,30 @@ __all__ = [
     "StringType", "BinaryType", "BooleanType", "TimestampType", "DecimalType",
     "DoubleType", "FloatType", "ByteType", "IntegerType", "LongType",
     "ShortType", "ArrayType", "MapType", "StructField", "StructType",
-    "SQLContext", "HiveContext", "LocalHiveContext", "TestHiveContext", "SchemaRDD", "Row"]
+    "SQLContext", "HiveContext", "LocalHiveContext", "TestHiveContext",
+    "SchemaRDD", "Row"]
+
+
+class DataType(object):
+    """Spark SQL DataType"""
+
+    def __repr__(self):
+        return self.__class__.__name__
+
+    def __hash__(self):
+        return hash(str(self))
+
+    def __eq__(self, other):
+        return (isinstance(other, self.__class__) and
+                self.__dict__ == other.__dict__)
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
 
 
 class PrimitiveTypeSingleton(type):
+    """Metaclass for PrimitiveType"""
+
     _instances = {}
 
     def __call__(cls):
@@ -38,148 +68,105 @@ class PrimitiveTypeSingleton(type):
         return cls._instances[cls]
 
 
-class StringType(object):
+class PrimitiveType(DataType):
+    """Spark SQL PrimitiveType"""
+
+    __metaclass__ = PrimitiveTypeSingleton
+
+    def __eq__(self, other):
+        # because they should be the same object
+        return self is other
+
+
+class StringType(PrimitiveType):
     """Spark SQL StringType
 
     The data type representing string values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "StringType"
 
 
-class BinaryType(object):
+class BinaryType(PrimitiveType):
     """Spark SQL BinaryType
 
     The data type representing bytearray values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "BinaryType"
 
 
-class BooleanType(object):
+class BooleanType(PrimitiveType):
     """Spark SQL BooleanType
 
     The data type representing bool values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "BooleanType"
 
 
-class TimestampType(object):
+class TimestampType(PrimitiveType):
     """Spark SQL TimestampType
 
     The data type representing datetime.datetime values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "TimestampType"
 
 
-class DecimalType(object):
+class DecimalType(PrimitiveType):
     """Spark SQL DecimalType
 
     The data type representing decimal.Decimal values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "DecimalType"
 
 
-class DoubleType(object):
+class DoubleType(PrimitiveType):
     """Spark SQL DoubleType
 
     The data type representing float values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "DoubleType"
 
 
-class FloatType(object):
+class FloatType(PrimitiveType):
     """Spark SQL FloatType
 
     The data type representing single precision floating-point values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
 
-    def __repr__(self):
-        return "FloatType"
 
-
-class ByteType(object):
+class ByteType(PrimitiveType):
     """Spark SQL ByteType
 
     The data type representing int values with 1 singed byte.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "ByteType"
 
 
-class IntegerType(object):
+class IntegerType(PrimitiveType):
     """Spark SQL IntegerType
 
     The data type representing int values.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
 
-    def __repr__(self):
-        return "IntegerType"
 
-
-class LongType(object):
+class LongType(PrimitiveType):
     """Spark SQL LongType
 
-    The data type representing long values. If the any value is beyond the range of
-    [-9223372036854775808, 9223372036854775807], please use DecimalType.
-
+    The data type representing long values. If the any value is
+    beyond the range of [-9223372036854775808, 9223372036854775807],
+    please use DecimalType.
     """
-    __metaclass__ = PrimitiveTypeSingleton
 
-    def __repr__(self):
-        return "LongType"
 
-
-class ShortType(object):
+class ShortType(PrimitiveType):
     """Spark SQL ShortType
 
     The data type representing int values with 2 signed bytes.
-
     """
-    __metaclass__ = PrimitiveTypeSingleton
-
-    def __repr__(self):
-        return "ShortType"
 
 
-class ArrayType(object):
+class ArrayType(DataType):
     """Spark SQL ArrayType
 
-    The data type representing list values.
-    An ArrayType object comprises two fields, elementType (a DataType) and containsNull (a bool).
+    The data type representing list values. An ArrayType object
+    comprises two fields, elementType (a DataType) and containsNull (a bool).
     The field of elementType is used to specify the type of array elements.
     The field of containsNull is used to specify if the array has None values.
 
     """
+
     def __init__(self, elementType, containsNull=False):
         """Creates an ArrayType
 
@@ -194,40 +181,39 @@ class ArrayType(object):
         self.elementType = elementType
         self.containsNull = containsNull
 
-    def __repr__(self):
-        return "ArrayType(" + self.elementType.__repr__() + "," + \
-               str(self.containsNull).lower() + ")"
-
-    def __eq__(self, other):
-        return (isinstance(other, self.__class__) and
-                self.elementType == other.elementType and
-                self.containsNull == other.containsNull)
-
-    def __ne__(self, other):
-        return not self.__eq__(other)
+    def __str__(self):
+        return "ArrayType(%s,%s)" % (self.elementType,
+               str(self.containsNull).lower())
 
 
-class MapType(object):
+class MapType(DataType):
     """Spark SQL MapType
 
-    The data type representing dict values.
-    A MapType object comprises three fields,
-    keyType (a DataType), valueType (a DataType) and valueContainsNull (a bool).
+    The data type representing dict values. A MapType object comprises
+    three fields, keyType (a DataType), valueType (a DataType) and
+    valueContainsNull (a bool).
+
     The field of keyType is used to specify the type of keys in the map.
     The field of valueType is used to specify the type of values in the map.
-    The field of valueContainsNull is used to specify if values of this map has None values.
+    The field of valueContainsNull is used to specify if values of this
+    map has None values.
+
     For values of a MapType column, keys are not allowed to have None values.
 
     """
+
     def __init__(self, keyType, valueType, valueContainsNull=True):
         """Creates a MapType
         :param keyType: the data type of keys.
         :param valueType: the data type of values.
-        :param valueContainsNull: indicates whether values contains null values.
+        :param valueContainsNull: indicates whether values contains
+        null values.
 
-        >>> MapType(StringType, IntegerType) == MapType(StringType, IntegerType, True)
+        >>> (MapType(StringType, IntegerType)
+        ...        == MapType(StringType, IntegerType, True))
         True
-        >>> MapType(StringType, IntegerType, False) == MapType(StringType, FloatType)
+        >>> (MapType(StringType, IntegerType, False)
+        ...        == MapType(StringType, FloatType))
         False
         """
         self.keyType = keyType
@@ -235,39 +221,36 @@ class MapType(object):
         self.valueContainsNull = valueContainsNull
 
     def __repr__(self):
-        return "MapType(" + self.keyType.__repr__() + "," + \
-               self.valueType.__repr__() + "," + \
-               str(self.valueContainsNull).lower() + ")"
+        return "MapType(%s,%s,%s)" % (self.keyType, self.valueType,
+               str(self.valueContainsNull).lower())
 
-    def __eq__(self, other):
-        return (isinstance(other, self.__class__) and
-                self.keyType == other.keyType and
-                self.valueType == other.valueType and
-                self.valueContainsNull == other.valueContainsNull)
 
-    def __ne__(self, other):
-        return not self.__eq__(other)
-
-
-class StructField(object):
+class StructField(DataType):
     """Spark SQL StructField
 
     Represents a field in a StructType.
-    A StructField object comprises three fields, name (a string), dataType (a DataType),
-    and nullable (a bool). The field of name is the name of a StructField. The field of
-    dataType specifies the data type of a StructField.
-    The field of nullable specifies if values of a StructField can contain None values.
+    A StructField object comprises three fields, name (a string),
+    dataType (a DataType) and nullable (a bool). The field of name
+    is the name of a StructField. The field of dataType specifies
+    the data type of a StructField.
+
+    The field of nullable specifies if values of a StructField can
+    contain None values.
 
     """
+
     def __init__(self, name, dataType, nullable):
         """Creates a StructField
         :param name: the name of this field.
         :param dataType: the data type of this field.
-        :param nullable: indicates whether values of this field can be null.
+        :param nullable: indicates whether values of this field
+                         can be null.
 
-        >>> StructField("f1", StringType, True) == StructField("f1", StringType, True)
+        >>> (StructField("f1", StringType, True)
+        ...      == StructField("f1", StringType, True))
         True
-        >>> StructField("f1", StringType, True) == StructField("f2", StringType, True)
+        >>> (StructField("f1", StringType, True)
+        ...      == StructField("f2", StringType, True))
         False
         """
         self.name = name
@@ -275,27 +258,18 @@ class StructField(object):
         self.nullable = nullable
 
     def __repr__(self):
-        return "StructField(" + self.name + "," + \
-               self.dataType.__repr__() + "," + \
-               str(self.nullable).lower() + ")"
+        return "StructField(%s,%s,%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.
+    The data type representing rows.
     A StructType object comprises a list of L{StructField}s.
 
     """
+
     def __init__(self, fields):
         """Creates a StructType
 
@@ -312,15 +286,8 @@ class StructType(object):
         self.fields = fields
 
     def __repr__(self):
-        return "StructType(List(" + \
-               ",".join([field.__repr__() for field in self.fields]) + "))"
-
-    def __eq__(self, other):
-        return (isinstance(other, self.__class__) and
-                self.fields == other.fields)
-
-    def __ne__(self, other):
-        return not self.__eq__(other)
+        return ("StructType(List(%s))" %
+                    ",".join(str(field) for field in self.fields))
 
 
 def _parse_datatype_list(datatype_list_string):
@@ -347,34 +314,19 @@ def _parse_datatype_list(datatype_list_string):
     return datatype_list
 
 
+_all_primitive_types = dict((k, v) for k, v in globals().iteritems()
+    if type(v) is PrimitiveTypeSingleton and v.__base__ == PrimitiveType)
+
+
 def _parse_datatype_string(datatype_string):
     """Parses the given data type string.
 
     >>> def check_datatype(datatype):
-    ...     scala_datatype = sqlCtx._ssql_ctx.parseDataType(datatype.__repr__())
-    ...     python_datatype = _parse_datatype_string(scala_datatype.toString())
+    ...     scala_datatype = sqlCtx._ssql_ctx.parseDataType(str(datatype))
+    ...     python_datatype = _parse_datatype_string(
+    ...                          scala_datatype.toString())
     ...     return datatype == python_datatype
-    >>> check_datatype(StringType())
-    True
-    >>> check_datatype(BinaryType())
-    True
-    >>> check_datatype(BooleanType())
-    True
-    >>> check_datatype(TimestampType())
-    True
-    >>> check_datatype(DecimalType())
-    True
-    >>> check_datatype(DoubleType())
-    True
-    >>> check_datatype(FloatType())
-    True
-    >>> check_datatype(ByteType())
-    True
-    >>> check_datatype(IntegerType())
-    True
-    >>> check_datatype(LongType())
-    True
-    >>> check_datatype(ShortType())
+    >>> all(check_datatype(cls()) for cls in _all_primitive_types.values())
     True
     >>> # Simple ArrayType.
     >>> simple_arraytype = ArrayType(StringType(), True)
@@ -405,70 +357,525 @@ 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)):
+        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 hasattr(row, "__FIELDS__"): # Row
+            items = zip(row.__FIELDS__, tuple(row))
+        elif all(isinstance(x, tuple) and len(x) == 2 for x in row):
+            items = row
+        else:
+            raise ValueError("Can't infer schema from tuple")
+
+    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 hasattr(obj, "__FIELDS__"):
+            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)
+        else:
+            raise ValueError("unexpected tuple")
+
+    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 _drop_schema(rows, schema):
+    """ 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)
+
+
+_BRACKETS = {'(': ')', '[': ']', '{': '}'}
+
+
+def _split_schema_abstract(s):
+    """
+    split the schema abstract into fields
+
+    >>> _split_schema_abstract("a b  c")
+    ['a', 'b', 'c']
+    >>> _split_schema_abstract("a(a b)")
+    ['a(a b)']
+    >>> _split_schema_abstract("a b[] c{a b}")
+    ['a', 'b[]', 'c{a b}']
+    >>> _split_schema_abstract(" ")
+    []
+    """
+
+    r = []
+    w = ''
+    brackets = []
+    for c in s:
+        if c == ' ' and not brackets:
+            if w:
+                r.append(w)
+            w = ''
+        else:
+            w += c
+            if c in _BRACKETS:
+                brackets.append(c)
+            elif c in _BRACKETS.values():
+                if not brackets or c != _BRACKETS[brackets.pop()]:
+                    raise ValueError("unexpected " + c)
+
+    if brackets:
+        raise ValueError("brackets not closed: %s" % brackets)
+    if w:
+        r.append(w)
+    return r
+
+
+def _parse_field_abstract(s):
+    """
+    Parse a field in schema abstract
+
+    >>> _parse_field_abstract("a")
+    StructField(a,None,true)
+    >>> _parse_field_abstract("b(c d)")
+    StructField(b,StructType(...c,None,true),StructField(d...
+    >>> _parse_field_abstract("a[]")
+    StructField(a,ArrayType(None,true),true)
+    >>> _parse_field_abstract("a{[]}")
+    StructField(a,MapType(None,ArrayType(None,true),true),true)
+    """
+    if set(_BRACKETS.keys()) & set(s):
+        idx = min((s.index(c) for c in _BRACKETS if c in s))
+        name = s[:idx]
+        return StructField(name, _parse_schema_abstract(s[idx:]), True)
+    else:
+        return StructField(s, None, True)
+
+
+def _parse_schema_abstract(s):
+    """
+    parse abstract into schema
+
+    >>> _parse_schema_abstract("a b  c")
+    StructType...a...b...c...
+    >>> _parse_schema_abstract("a[b c] b{}")
+    StructType...a,ArrayType...b...c...b,MapType...
+    >>> _parse_schema_abstract("c{} d{a b}")
+    StructType...c,MapType...d,MapType...a...b...
+    >>> _parse_schema_abstract("a b(t)").fields[1]
+    StructField(b,StructType(List(StructField(t,None,true))),true)
+    """
+    s = s.strip()
+    if not s:
+        return
+
+    elif s.startswith('('):
+        return _parse_schema_abstract(s[1:-1])
+
+    elif s.startswith('['):
+        return ArrayType(_parse_schema_abstract(s[1:-1]), True)
+
+    elif s.startswith('{'):
+        return MapType(None, _parse_schema_abstract(s[1:-1]))
+
+    parts = _split_schema_abstract(s)
+    fields = [_parse_field_abstract(p) for p in parts]
+    return StructType(fields)
+
+
+def _infer_schema_type(obj, dataType):
+    """
+    Fill the dataType with types infered from obj
+
+    >>> schema = _parse_schema_abstract("a b c")
+    >>> row = (1, 1.0, "str")
+    >>> _infer_schema_type(row, schema)
+    StructType...IntegerType...DoubleType...StringType...
+    >>> row = [[1], {"key": (1, 2.0)}]
+    >>> schema = _parse_schema_abstract("a[] b{c d}")
+    >>> _infer_schema_type(row, schema)
+    StructType...a,ArrayType...b,MapType(StringType,...c,IntegerType...
+    """
+    if dataType is None:
+        return _infer_type(obj)
+
+    if not obj:
+        raise ValueError("Can not infer type from empty value")
+
+    if isinstance(dataType, ArrayType):
+        eType = _infer_schema_type(obj[0], dataType.elementType)
+        return ArrayType(eType, True)
+
+    elif isinstance(dataType, MapType):
+        k, v = obj.iteritems().next()
+        return MapType(_infer_type(k),
+                       _infer_schema_type(v, dataType.valueType))
+
+    elif isinstance(dataType, StructType):
+        fs = dataType.fields
+        assert len(fs) == len(obj), \
+            "Obj(%s) have different length with fields(%s)" % (obj, fs)
+        fields = [StructField(f.name, _infer_schema_type(o, f.dataType), True)
+                    for o, f in zip(obj, fs)]
+        return StructType(fields)
+
+    else:
+        raise ValueError("Unexpected dataType: %s" % dataType)
+
+
+_acceptable_types = {
+    BooleanType: (bool,),
+    ByteType: (int, long),
+    ShortType: (int, long),
+    IntegerType: (int, long),
+    LongType: (int, long),
+    FloatType: (float,),
+    DoubleType: (float,),
+    DecimalType: (decimal.Decimal,),
+    StringType: (str, unicode),
+    TimestampType: (datetime.datetime, datetime.time, datetime.date),
+    ArrayType: (list, tuple, array),
+    MapType: (dict,),
+    StructType: (tuple, list),
+}
+
+def _verify_type(obj, dataType):
+    """
+    Verify the type of obj against dataType, raise an exception if
+    they do not match.
+
+    >>> _verify_type(None, StructType([]))
+    >>> _verify_type("", StringType())
+    >>> _verify_type(0, IntegerType())
+    >>> _verify_type(range(3), ArrayType(ShortType()))
+    >>> _verify_type(set(), ArrayType(StringType())) # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    TypeError:...
+    >>> _verify_type({}, MapType(StringType(), IntegerType()))
+    >>> _verify_type((), StructType([]))
+    >>> _verify_type([], StructType([]))
+    >>> _verify_type([1], StructType([])) # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    ValueError:...
+    """
+    # all objects are nullable
+    if obj is None:
+        return
+
+    _type = type(dataType)
+    if _type not in _acceptable_types:
+        return
+
+    if type(obj) not in _acceptable_types[_type]:
+        raise TypeError("%s can not accept abject in type %s"
+                        % (dataType, type(obj)))
+
+    if isinstance(dataType, ArrayType):
+        for i in obj:
+            _verify_type(i, dataType.elementType)
+
+    elif isinstance(dataType, MapType):
+        for k, v in obj.iteritems():
+            _verify_type(k, dataType.keyType)
+            _verify_type(v, dataType.valueType)
+
+    elif isinstance(dataType, StructType):
+        if len(obj) != len(dataType.fields):
+            raise ValueError("Length of object (%d) does not match with"
+                "length of fields (%d)" % (len(obj), len(dataType.fields)))
+        for v, f in zip(obj, dataType.fields):
+            _verify_type(v, f.dataType)
+
+
+_cached_cls = {}
+
+
+def _restore_object(dataType, obj):
+    """ Restore object during unpickling. """
+    # use id(dataType) as key to speed up lookup in dict
+    # Because of batched pickling, dataType will be the
+    # same object in mose cases.
+    k = id(dataType)
+    cls = _cached_cls.get(k)
+    if cls is None:
+        # use dataType as key to avoid create multiple class
+        cls = _cached_cls.get(dataType)
+        if cls is None:
+            cls = _create_cls(dataType)
+            _cached_cls[dataType] = cls
+        _cached_cls[k] = 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_properties(fields):
+    """Create properties according to fields"""
+    ps = {}
+    for i, f in enumerate(fields):
+        name = f.name
+        if (name.startswith("__") and name.endswith("__")
+                or keyword.iskeyword(name)):
+            warnings.warn("field name %s can not be accessed in Python,"
+                          "use position to access it instead" % name)
+        if _has_struct(f.dataType):
+            # delay creating object until accessing it
+            getter = _create_getter(f.dataType, i)
+        else:
+            getter = itemgetter(i)
+        ps[name] = property(getter)
+    return ps
+
+
+def _create_cls(dataType):
+    """
+    Create an class by dataType
+
+    The created class is similar to namedtuple, but can have nested schema.
+
+    >>> schema = _parse_schema_abstract("a b c")
+    >>> row = (1, 1.0, "str")
+    >>> schema = _infer_schema_type(row, schema)
+    >>> obj = _create_cls(schema)(row)
+    >>> import pickle
+    >>> pickle.loads(pickle.dumps(obj))
+    Row(a=1, b=1.0, c='str')
+
+    >>> row = [[1], {"key": (1, 2.0)}]
+    >>> schema = _parse_schema_abstract("a[] b{c d}")
+    >>> schema = _infer_schema_type(row, schema)
+    >>> obj = _create_cls(schema)(row)
+    >>> pickle.loads(pickle.dumps(obj))
+    Row(a=[1], b={'key': Row(c=1, d=2.0)})
+    """
+
+    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):
+                return list.__reduce__(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):
+                return dict.__reduce__(self)
+
+        return Dict
+
+    elif not isinstance(dataType, StructType):
+        raise Exception("unexpected data type: %s" % dataType)
+
+    class Row(tuple):
+        """ Row in SchemaRDD """
+        __DATATYPE__ = dataType
+        __FIELDS__ = tuple(f.name for f in dataType.fields)
+        __slots__ = ()
+
+        # create property for fast access
+        locals().update(_create_properties(dataType.fields))
+
+        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):
+            return (_restore_object, (self.__DATATYPE__, tuple(self)))
+
+    return Row
+
+
 class SQLContext:
     """Main entry point for SparkSQL functionality.
 
@@ -485,7 +892,7 @@ class SQLContext:
         >>> sqlCtx.inferSchema(srdd) # doctest: +IGNORE_EXCEPTION_DETAIL
         Traceback (most recent call last):
             ...
-        ValueError:...
+        TypeError:...
 
         >>> bad_rdd = sc.parallelize([1,2,3])
         >>> sqlCtx.inferSchema(bad_rdd) # doctest: +IGNORE_EXCEPTION_DETAIL
@@ -494,18 +901,22 @@ class SQLContext:
         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))
-        >>> srdd.collect()[0]
-        (1, u'string', 1.0, 1, True, datetime.datetime(2010, 1, 1, 1, 1, 1), 1, [1, 2, 3])
+        >>> allTypes = sc.parallelize([Row(i=1, s="string", d=1.0, l=1L,
+        ...     b=True, list=[1, 2, 3], dict={"s": 0}, row=Row(a=1),
+        ...     time=datetime(2014, 8, 1, 14, 1, 5))])
+        >>> srdd = sqlCtx.inferSchema(allTypes)
+        >>> srdd.registerAsTable("allTypes")
+        >>> sqlCtx.sql('select i+1, d+1, not b, list[1], dict["s"], time, row.a '
+        ...            'from allTypes where b and i > 0').collect()
+        [Row(c0=2, c1=2.0, c2=False, c3=2, c4=0...8, 1, 14, 1, 5), a=1)]
+        >>> srdd.map(lambda x: (x.i, x.s, x.d, x.l, x.b, x.time,
+        ...                     x.row.a, x.list)).collect()
+        [(1, u'string', 1.0, 1, True, ...(2014, 8, 1, 14, 1, 5), 1, [1, 2, 3])]
         """
         self._sc = sparkContext
         self._jsc = self._sc._jsc
         self._jvm = self._sc._jvm
-        self._pythonToJavaMap = self._jvm.PythonRDD.pythonToJavaMap
+        self._pythonToJava = self._jvm.PythonRDD.pythonToJavaArray
 
         if sqlContext:
             self._scala_SQLContext = sqlContext
@@ -522,71 +933,123 @@ class SQLContext:
         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.
+        All the rows in `rdd` should have the same type with the first one,
+        or it will cause runtime exceptions.
 
+        Each row could be L{pyspark.sql.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.
+
+        These tuples or lists can contain complex nested structures like
+        lists, maps or nested rows.
+
+        The schema should be a StructType.
 
+        It is important that the schema matches the types of the objects
+        in each row or exceptions could be thrown at runtime.
+
+        >>> rdd2 = sc.parallelize([(1, "row1"), (2, "row2"), (3, "row3")])
         >>> schema = StructType([StructField("field1", IntegerType(), False),
         ...     StructField("field2", StringType(), False)])
-        >>> srdd = sqlCtx.applySchema(rdd, schema)
+        >>> srdd = sqlCtx.applySchema(rdd2, schema)
         >>> sqlCtx.registerRDDAsTable(srdd, "table1")
         >>> srdd2 = sqlCtx.sql("SELECT * from table1")
-        >>> srdd2.collect() == [{"field1" : 1, "field2" : "row1"}, {"field1" : 2, "field2": "row2"},
-        ...                    {"field1" : 3, "field2": "row3"}]
-        True
+        >>> srdd2.collect()
+        [Row(field1=1, field2=u'row1'),..., Row(field1=3, field2=u'row3')]
+
         >>> from datetime import datetime
-        >>> rdd = sc.parallelize([{"byte": 127, "short": -32768, "float": 1.0,
-        ... "time": datetime(2010, 1, 1, 1, 1, 1), "map": {"a": 1}, "struct": {"b": 2},
-        ... "list": [1, 2, 3]}])
+        >>> rdd = sc.parallelize([(127, -32768, 1.0,
+        ...     datetime(2010, 1, 1, 1, 1, 1),
+        ...     {"a": 1}, (2,), [1, 2, 3], None)])
         >>> schema = StructType([
         ...     StructField("byte", ByteType(), False),
         ...     StructField("short", ShortType(), False),
         ...     StructField("float", FloatType(), False),
         ...     StructField("time", TimestampType(), False),
-        ...     StructField("map", MapType(StringType(), IntegerType(), False), False),
-        ...     StructField("struct", StructType([StructField("b", ShortType(), False)]), False),
+        ...     StructField("map",
+        ...         MapType(StringType(), IntegerType(), False), False),
+        ...     StructField("struct",
+        ...         StructType([StructField("b", ShortType(), False)]), False),
         ...     StructField("list", ArrayType(ByteType(), False), False),
         ...     StructField("null", DoubleType(), True)])
         >>> srdd = sqlCtx.applySchema(rdd, schema).map(
-        ...     lambda x: (
-        ...         x.byte, x.short, x.float, x.time, x.map["a"], x.struct["b"], x.list, x.null))
+        ...     lambda x: (x.byte, x.short, x.float, x.time,
+        ...         x.map["a"], x.struct.b, x.list, x.null))
         >>> srdd.collect()[0]
-        (127, -32768, 1.0, datetime.datetime(2010, 1, 1, 1, 1, 1), 1, 2, [1, 2, 3], None)
+        (127, -32768, 1.0, ...(2010, 1, 1, 1, 1, 1), 1, 2, [1, 2, 3], None)
+
+        >>> rdd = sc.parallelize([(127, -32768, 1.0,
+        ...     datetime(2010, 1, 1, 1, 1, 1),
+        ...     {"a": 1}, (2,), [1, 2, 3])])
+        >>> abstract = "byte short float time map{} struct(b) list[]"
+        >>> schema = _parse_schema_abstract(abstract)
+        >>> typedSchema = _infer_schema_type(rdd.first(), schema)
+        >>> srdd = sqlCtx.applySchema(rdd, typedSchema)
+        >>> srdd.collect()
+        [Row(byte=127, short=-32768, float=1.0, time=..., list=[1, 2, 3])]
         """
-        jrdd = self._pythonToJavaMap(rdd._jrdd)
-        srdd = self._ssql_ctx.applySchemaToPythonRDD(jrdd.rdd(), schema.__repr__())
+
+        if isinstance(rdd, SchemaRDD):
+            raise TypeError("Cannot apply schema to SchemaRDD")
+
+        if not isinstance(schema, StructType):
+            raise TypeError("schema should be StructType")
+
+        # take the first few rows to verify schema
+        rows = rdd.take(10)
+        for row in rows:
+            _verify_type(row, schema)
+
+        batched = isinstance(rdd._jrdd_deserializer, BatchedSerializer)
+        jrdd = self._pythonToJava(rdd._jrdd, batched)
+        srdd = self._ssql_ctx.applySchemaToPythonRDD(jrdd.rdd(), str(schema))
         return SchemaRDD(srdd, self)
 
     def registerRDDAsTable(self, rdd, tableName):
@@ -620,10 +1083,15 @@ class SQLContext:
         return SchemaRDD(jschema_rdd, self)
 
     def jsonFile(self, path, schema=None):
-        """Loads a text file storing one JSON object per line as a L{SchemaRDD}.
+        """
+        Loads a text file storing one JSON object per line as a
+        L{SchemaRDD}.
 
-        If the schema is provided, applies the given schema to this JSON dataset.
-        Otherwise, it goes through the entire dataset once to determine the schema.
+        If the schema is provided, applies the given schema to this
+        JSON dataset.
+
+        Otherwise, it goes through the entire dataset once to determine
+        the schema.
 
         >>> import tempfile, shutil
         >>> jsonFile = tempfile.mkdtemp()
@@ -635,94 +1103,100 @@ class SQLContext:
         >>> srdd1 = sqlCtx.jsonFile(jsonFile)
         >>> 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
+        ...   "SELECT field1 AS f1, field2 as f2, field3 as f3, "
+        ...   "field6 as f4 from table1")
+        >>> for r in srdd2.collect():
+        ...     print r
+        Row(f1=1, f2=u'row1', f3=Row(field4=11, field5=None), f4=None)
+        Row(f1=2, f2=None, f3=Row(field4=22,..., f4=[Row(field7=u'row2')])
+        Row(f1=None, f2=u'row3', f3=Row(field4=33, field5=[]), f4=None)
         >>> srdd3 = sqlCtx.jsonFile(jsonFile, srdd1.schema())
         >>> sqlCtx.registerRDDAsTable(srdd3, "table2")
         >>> srdd4 = sqlCtx.sql(
-        ...   "SELECT field1 AS f1, field2 as f2, field3 as f3, field6 as f4 from table2")
-        >>> srdd4.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
+        ...   "SELECT field1 AS f1, field2 as f2, field3 as f3, "
+        ...   "field6 as f4 from table2")
+        >>> for r in srdd4.collect():
+        ...    print r
+        Row(f1=1, f2=u'row1', f3=Row(field4=11, field5=None), f4=None)
+        Row(f1=2, f2=None, f3=Row(field4=22,..., f4=[Row(field7=u'row2')])
+        Row(f1=None, f2=u'row3', f3=Row(field4=33, field5=[]), f4=None)
         >>> schema = StructType([
         ...     StructField("field2", StringType(), True),
         ...     StructField("field3",
         ...         StructType([
-        ...             StructField("field5", ArrayType(IntegerType(), False), True)]), False)])
+        ...             StructField("field5",
+        ...                 ArrayType(IntegerType(), False), True)]), False)])
         >>> srdd5 = sqlCtx.jsonFile(jsonFile, schema)
         >>> sqlCtx.registerRDDAsTable(srdd5, "table3")
         >>> srdd6 = sqlCtx.sql(
-        ...   "SELECT field2 AS f1, field3.field5 as f2, field3.field5[0] as f3 from table3")
-        >>> srdd6.collect() == [
-        ... {"f1": "row1", "f2": None, "f3": None},
-        ... {"f1": None, "f2": [10, 11], "f3": 10},
-        ... {"f1": "row3", "f2": [], "f3": None}]
-        True
+        ...   "SELECT field2 AS f1, field3.field5 as f2, "
+        ...   "field3.field5[0] as f3 from table3")
+        >>> srdd6.collect()
+        [Row(f1=u'row1', f2=None, f3=None)...Row(f1=u'row3', f2=[], f3=None)]
         """
         if schema is None:
             jschema_rdd = self._ssql_ctx.jsonFile(path)
         else:
-            scala_datatype = self._ssql_ctx.parseDataType(schema.__repr__())
+            scala_datatype = self._ssql_ctx.parseDataType(str(schema))
             jschema_rdd = self._ssql_ctx.jsonFile(path, scala_datatype)
         return SchemaRDD(jschema_rdd, self)
 
     def jsonRDD(self, rdd, schema=None):
         """Loads an RDD storing one JSON object per string as a L{SchemaRDD}.
 
-        If the schema is provided, applies the given schema to this JSON dataset.
-        Otherwise, it goes through the entire dataset once to determine the schema.
+        If the schema is provided, applies the given schema to this
+        JSON dataset.
+
+        Otherwise, it goes through the entire dataset once to determine
+        the schema.
 
         >>> srdd1 = sqlCtx.jsonRDD(json)
         >>> 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
+        ...   "SELECT field1 AS f1, field2 as f2, field3 as f3, "
+        ...   "field6 as f4 from table1")
+        >>> for r in srdd2.collect():
+        ...     print r
+        Row(f1=1, f2=u'row1', f3=Row(field4=11, field5=None), f4=None)
+        Row(f1=2, f2=None, f3=Row(field4=22..., f4=[Row(field7=u'row2')])
+        Row(f1=None, f2=u'row3', f3=Row(field4=33, field5=[]), f4=None)
         >>> srdd3 = sqlCtx.jsonRDD(json, srdd1.schema())
         >>> sqlCtx.registerRDDAsTable(srdd3, "table2")
         >>> srdd4 = sqlCtx.sql(
-        ...   "SELECT field1 AS f1, field2 as f2, field3 as f3, field6 as f4 from table2")
-        >>> srdd4.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
+        ...   "SELECT field1 AS f1, field2 as f2, field3 as f3, "
+        ...   "field6 as f4 from table2")
+        >>> for r in srdd4.collect():
+        ...     print r
+        Row(f1=1, f2=u'row1', f3=Row(field4=11, field5=None), f4=None)
+        Row(f1=2, f2=None, f3=Row(field4=22..., f4=[Row(field7=u'row2')])
+        Row(f1=None, f2=u'row3', f3=Row(field4=33, field5=[]), f4=None)
         >>> schema = StructType([
         ...     StructField("field2", StringType(), True),
         ...     StructField("field3",
         ...         StructType([
-        ...             StructField("field5", ArrayType(IntegerType(), False), True)]), False)])
+        ...             StructField("field5",
+        ...                 ArrayType(IntegerType(), False), True)]), False)])
         >>> srdd5 = sqlCtx.jsonRDD(json, schema)
         >>> sqlCtx.registerRDDAsTable(srdd5, "table3")
         >>> srdd6 = sqlCtx.sql(
-        ...   "SELECT field2 AS f1, field3.field5 as f2, field3.field5[0] as f3 from table3")
-        >>> srdd6.collect() == [
-        ... {"f1": "row1", "f2": None, "f3": None},
-        ... {"f1": None, "f2": [10, 11], "f3": 10},
-        ... {"f1": "row3", "f2": [], "f3": None}]
-        True
+        ...   "SELECT field2 AS f1, field3.field5 as f2, "
+        ...   "field3.field5[0] as f3 from table3")
+        >>> srdd6.collect()
+        [Row(f1=u'row1', f2=None,...Row(f1=u'row3', f2=[], f3=None)]
         """
-        def func(split, iterator):
+
+        def func(iterator):
             for x in iterator:
                 if not isinstance(x, basestring):
                     x = unicode(x)
                 yield x.encode("utf-8")
-        keyed = PipelinedRDD(rdd, func)
+        keyed = rdd.mapPartitions(func)
         keyed._bypass_serializer = True
         jrdd = keyed._jrdd.map(self._jvm.BytesToString())
         if schema is None:
             jschema_rdd = self._ssql_ctx.jsonRDD(jrdd.rdd())
         else:
-            scala_datatype = self._ssql_ctx.parseDataType(schema.__repr__())
+            scala_datatype = self._ssql_ctx.parseDataType(str(schema))
             jschema_rdd = self._ssql_ctx.jsonRDD(jrdd.rdd(), scala_datatype)
         return SchemaRDD(jschema_rdd, self)
 
@@ -732,9 +1206,8 @@ class SQLContext:
         >>> srdd = sqlCtx.inferSchema(rdd)
         >>> sqlCtx.registerRDDAsTable(srdd, "table1")
         >>> srdd2 = sqlCtx.sql("SELECT field1 AS f1, field2 as f2 from table1")
-        >>> srdd2.collect() == [{"f1" : 1, "f2" : "row1"}, {"f1" : 2, "f2": "row2"},
-        ...                     {"f1" : 3, "f2": "row3"}]
-        True
+        >>> srdd2.collect()
+        [Row(f1=1, f2=u'row1'), Row(f1=2, f2=u'row2'), Row(f1=3, f2=u'row3')]
         """
         return SchemaRDD(self._ssql_ctx.sql(sqlQuery), self)
 
@@ -772,7 +1245,8 @@ class HiveContext(SQLContext):
                 self._scala_HiveContext = self._get_hive_ctx()
             return self._scala_HiveContext
         except Py4JError as e:
-            raise Exception("You must build Spark with Hive. Export 'SPARK_HIVE=true' and run "
+            raise Exception("You must build Spark with Hive. "
+                            "Export 'SPARK_HIVE=true' and run "
                             "sbt/sbt assembly", e)
 
     def _get_hive_ctx(self):
@@ -780,13 +1254,15 @@ class HiveContext(SQLContext):
 
     def hiveql(self, hqlQuery):
         """
-        Runs a query expressed in HiveQL, returning the result as a L{SchemaRDD}.
+        Runs a query expressed in HiveQL, returning the result as
+        a L{SchemaRDD}.
         """
         return SchemaRDD(self._ssql_ctx.hiveql(hqlQuery), self)
 
     def hql(self, hqlQuery):
         """
-        Runs a query expressed in HiveQL, returning the result as a L{SchemaRDD}.
+        Runs a query expressed in HiveQL, returning the result as
+        a L{SchemaRDD}.
         """
         return self.hiveql(hqlQuery)
 
@@ -803,10 +1279,14 @@ class LocalHiveContext(HiveContext):
     ...     supress = hiveCtx.hql("DROP TABLE src")
     ... except Exception:
     ...     pass
-    >>> kv1 = os.path.join(os.environ["SPARK_HOME"], 'examples/src/main/resources/kv1.txt')
-    >>> supress = hiveCtx.hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
-    >>> supress = hiveCtx.hql("LOAD DATA LOCAL INPATH '%s' INTO TABLE src" % kv1)
-    >>> results = hiveCtx.hql("FROM src SELECT value").map(lambda r: int(r.value.split('_')[1]))
+    >>> kv1 = os.path.join(os.environ["SPARK_HOME"],
+    ...        'examples/src/main/resources/kv1.txt')
+    >>> supress = hiveCtx.hql(
+    ...     "CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
+    >>> supress = hiveCtx.hql("LOAD DATA LOCAL INPATH '%s' INTO TABLE src"
+    ...        % kv1)
+    >>> results = hiveCtx.hql("FROM src SELECT value"
+    ...      ).map(lambda r: int(r.value.split('_')[1]))
     >>> num = results.count()
     >>> reduce_sum = results.reduce(lambda x, y: x + y)
     >>> num
@@ -816,8 +1296,9 @@ class LocalHiveContext(HiveContext):
     """
 
     def __init__(self, sparkContext, sqlContext=None):
-      HiveContext.__init__(self, sparkContext, sqlContext)
-      warnings.warn("LocalHiveContext is deprecated.  Use HiveContext instead.", DeprecationWarning)
+        HiveContext.__init__(self, sparkContext, sqlContext)
+        warnings.warn("LocalHiveContext is deprecated. "
+                "Use HiveContext instead.", DeprecationWarning)
 
     def _get_hive_ctx(self):
         return self._jvm.LocalHiveContext(self._jsc.sc())
@@ -829,25 +1310,83 @@ class TestHiveContext(HiveContext):
         return self._jvm.TestHiveContext(self._jsc.sc())
 
 
-# TODO: Investigate if it is more efficient to use a namedtuple. One problem is that named tuples
-# are custom classes that must be generated per Schema.
-class Row(dict):
-    """A row in L{SchemaRDD}.
+def _create_row(fields, values):
+    row = Row(*values)
+    row.__FIELDS__ = fields
+    return row
+
+
+class Row(tuple):
+    """
+    A row in L{SchemaRDD}. The fields in it can be accessed like attributes.
+
+    Row can be used to create a row object by using named arguments,
+    the fields will be sorted by names.
+
+    >>> row = Row(name="Alice", age=11)
+    >>> row
+    Row(age=11, name='Alice')
+    >>> row.name, row.age
+    ('Alice', 11)
 
-    An extended L{dict} that takes a L{dict} in its constructor, and
-    exposes those items as fields.
+    Row also can be used to create another Row like class, then it
+    could be used to create Row objects, such as
 
-    >>> r = Row({"hello" : "world", "foo" : "bar"})
-    >>> r.hello
-    'world'
-    >>> r.foo
-    'bar'
+    >>> Person = Row("name", "age")
+    >>> Person
+    <Row(name, age)>
+    >>> Person("Alice", 11)
+    Row(name='Alice', age=11)
     """
 
-    def __init__(self, d):
-        d.update(self.__dict__)
-        self.__dict__ = d
-        dict.__init__(self, d)
+    def __new__(self, *args, **kwargs):
+        if args and kwargs:
+            raise ValueError("Can not use both args "
+                             "and kwargs to create Row")
+        if args:
+            # create row class or objects
+            return tuple.__new__(self, args)
+
+        elif kwargs:
+            # create row objects
+            names = sorted(kwargs.keys())
+            values = tuple(kwargs[n] for n in names)
+            row = tuple.__new__(self, values)
+            row.__FIELDS__ = names
+            return row
+
+        else:
+            raise ValueError("No args or kwargs")
+
+
+    # let obect acs like class
+    def __call__(self, *args):
+        """create new Row object"""
+        return _create_row(self, args)
+
+    def __getattr__(self, item):
+        if item.startswith("__"):
+            raise AttributeError(item)
+        try:
+            # it will be slow when it has many fields,
+            # but this will not be used in normal cases
+            idx = self.__FIELDS__.index(item)
+            return self[idx]
+        except IndexError:
+            raise AttributeError(item)
+
+    def __reduce__(self):
+        if hasattr(self, "__FIELDS__"):
+            return (_create_row, (self.__FIELDS__, tuple(self)))
+        else:
+            return tuple.__reduce__(self)
+
+    def __repr__(self):
+        if hasattr(self, "__FIELDS__"):
+            return "Row(%s)" % ", ".join("%s=%r" % (k, v)
+                for k, v in zip(self.__FIELDS__, self))
+        else:
+            return "<Row(%s)>" % ", ".join(self)
 
 
 class SchemaRDD(RDD):
@@ -861,6 +1400,10 @@ class SchemaRDD(RDD):
     implementation is an RDD composed of Java objects. Instead it is
     converted to a PythonRDD in the JVM, on which Python operations can
     be done.
+
+    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.
     """
 
     def __init__(self, jschema_rdd, sql_ctx):
@@ -871,7 +1414,8 @@ class SchemaRDD(RDD):
         self.is_cached = False
         self.is_checkpointed = False
         self.ctx = self.sql_ctx._sc
-        self._jrdd_deserializer = self.ctx.serializer
+        # the _jrdd is created by javaToPython(), serialized by pickle
+        self._jrdd_deserializer = BatchedSerializer(PickleSerializer())
 
     @property
     def _jrdd(self):
@@ -881,7 +1425,7 @@ class SchemaRDD(RDD):
         L{pyspark.rdd.RDD} super class (map, filter, etc.).
         """
         if not hasattr(self, '_lazy_jrdd'):
-            self._lazy_jrdd = self._toPython()._jrdd
+            self._lazy_jrdd = self._jschema_rdd.javaToPython()
         return self._lazy_jrdd
 
     @property
@@ -931,7 +1475,8 @@ class SchemaRDD(RDD):
         self._jschema_rdd.saveAsTable(tableName)
 
     def schema(self):
-        """Returns the schema of this SchemaRDD (represented by a L{StructType})."""
+        """Returns the schema of this SchemaRDD (represented by
+        a L{StructType})."""
         return _parse_datatype_string(self._jschema_rdd.schema().toString())
 
     def schemaString(self):
@@ -957,19 +1502,45 @@ class SchemaRDD(RDD):
         """
         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))
-
-    # We override the default cache/persist/checkpoint behavior as we want to cache the underlying
-    # SchemaRDD object in the JVM, not the PythonRDD checkpointed by the super class
+    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 each object in the RDD to a Row with the right class
+    # for this SchemaRDD, so that fields can be accessed as attributes.
+    def mapPartitionsWithIndex(self, f, preservesPartitioning=False):
+        """
+        Return a new RDD by applying a function to each partition of this RDD,
+        while tracking the index of the original partition.
+
+        >>> rdd = sc.parallelize([1, 2, 3, 4], 4)
+        >>> def f(splitIndex, iterator): yield splitIndex
+        >>> rdd.mapPartitionsWithIndex(f).sum()
+        6
+        """
+        rdd = RDD(self._jrdd, self._sc, self._jrdd_deserializer)
+
+        schema = self.schema()
+        import pickle
+        pickle.loads(pickle.dumps(schema))
+
+        def applySchema(_, it):
+            cls = _create_cls(schema)
+            return itertools.imap(cls, it)
+
+        objrdd = rdd.mapPartitionsWithIndex(applySchema, preservesPartitioning)
+        return objrdd.mapPartitionsWithIndex(f, preservesPartitioning)
+
+    # We override the default cache/persist/checkpoint behavior
+    # as we want to cache the underlying SchemaRDD object in the JVM,
+    # not the PythonRDD checkpointed by the super class
     def cache(self):
         self.is_cached = True
         self._jschema_rdd.cache()
@@ -1024,7 +1595,8 @@ class SchemaRDD(RDD):
             if numPartitions is None:
                 rdd = self._jschema_rdd.subtract(other._jschema_rdd)
             else:
-                rdd = self._jschema_rdd.subtract(other._jschema_rdd, numPartitions)
+                rdd = self._jschema_rdd.subtract(other._jschema_rdd,
+                        numPartitions)
             return SchemaRDD(rdd, self.sql_ctx)
         else:
             raise ValueError("Can only subtract another SchemaRDD")
@@ -1034,31 +1606,31 @@ def _test():
     import doctest
     from array import array
     from pyspark.context import SparkContext
-    globs = globals().copy()
+    # let doctest run in pyspark.sql, so DataTypes can be picklable
+    import pyspark.sql
+    from pyspark.sql import Row, SQLContext
+    globs = pyspark.sql.__dict__.copy()
     # The small batch size here ensures that we see multiple batches,
     # even in these small test examples:
     sc = SparkContext('local[4]', 'PythonTest', batchSize=2)
     globs['sc'] = sc
     globs['sqlCtx'] = SQLContext(sc)
     globs['rdd'] = sc.parallelize(
-        [{"field1": 1, "field2": "row1"},
-         {"field1": 2, "field2": "row2"},
-         {"field1": 3, "field2": "row3"}]
+        [Row(field1=1, field2="row1"),
+         Row(field1=2, field2="row2"),
+         Row(field1=3, field2="row3")]
     )
     jsonStrings = [
         '{"field1": 1, "field2": "row1", "field3":{"field4":11}}',
-        '{"field1" : 2, "field3":{"field4":22, "field5": [10, 11]}, "field6":[{"field7": "row2"}]}',
-        '{"field1" : null, "field2": "row3", "field3":{"field4":33, "field5": []}}'
+        '{"field1" : 2, "field3":{"field4":22, "field5": [10, 11]},'
+            '"field6":[{"field7": "row2"}]}',
+        '{"field1" : null, "field2": "row3", '
+            '"field3":{"field4":33, "field5": []}}'
     ]
     globs['jsonStrings'] = jsonStrings
     globs['json'] = sc.parallelize(jsonStrings)
-    globs['nestedRdd1'] = sc.parallelize([
-        {"f1": array('i', [1, 2]), "f2": {"row1": 1.0}},
-        {"f1": array('i', [2, 3]), "f2": {"row2": 2.0}}])
-    globs['nestedRdd2'] = sc.parallelize([
-        {"f1": [[1, 2], [2, 3]], "f2": [1, 2]},
-        {"f1": [[2, 3], [3, 4]], "f2": [2, 3]}])
-    (failure_count, test_count) = doctest.testmod(globs=globs, optionflags=doctest.ELLIPSIS)
+    (failure_count, test_count) = doctest.testmod(
+        pyspark.sql, globs=globs, optionflags=doctest.ELLIPSIS)
     globs['sc'].stop()
     if failure_count:
         exit(-1)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala b/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
index 86338752a2..dad71079c2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
@@ -412,35 +412,6 @@ class SQLContext(@transient val sparkContext: SparkContext)
   }
 
   /**
-   * Peek at the first row of the RDD and infer its schema.
-   * It is only used by PySpark.
-   */
-  private[sql] def inferSchema(rdd: RDD[Map[String, _]]): SchemaRDD = {
-    import scala.collection.JavaConversions._
-
-    def typeOfComplexValue: PartialFunction[Any, DataType] = {
-      case c: java.util.Calendar => TimestampType
-      case c: java.util.List[_] =>
-        ArrayType(typeOfObject(c.head))
-      case c: java.util.Map[_, _] =>
-        val (key, value) = c.head
-        MapType(typeOfObject(key), typeOfObject(value))
-      case c if c.getClass.isArray =>
-        val elem = c.asInstanceOf[Array[_]].head
-        ArrayType(typeOfObject(elem))
-      case c => throw new Exception(s"Object of type $c cannot be used")
-    }
-    def typeOfObject = ScalaReflection.typeOfObject orElse typeOfComplexValue
-
-    val firstRow = rdd.first()
-    val fields = firstRow.map {
-      case (fieldName, obj) => StructField(fieldName, typeOfObject(obj), true)
-    }.toSeq
-
-    applySchemaToPythonRDD(rdd, StructType(fields))
-  }
-
-  /**
    * Parses the data type in our internal string representation. The data type string should
    * have the same format as the one generated by `toString` in scala.
    * It is only used by PySpark.
@@ -454,7 +425,7 @@ class SQLContext(@transient val sparkContext: SparkContext)
    * Apply a schema defined by the schemaString to an RDD. It is only used by PySpark.
    */
   private[sql] def applySchemaToPythonRDD(
-      rdd: RDD[Map[String, _]],
+      rdd: RDD[Array[Any]],
       schemaString: String): SchemaRDD = {
     val schema = parseDataType(schemaString).asInstanceOf[StructType]
     applySchemaToPythonRDD(rdd, schema)
@@ -464,10 +435,8 @@ class SQLContext(@transient val sparkContext: SparkContext)
    * Apply a schema defined by the schema to an RDD. It is only used by PySpark.
    */
   private[sql] def applySchemaToPythonRDD(
-      rdd: RDD[Map[String, _]],
+      rdd: RDD[Array[Any]],
       schema: StructType): SchemaRDD = {
-    // TODO: We should have a better implementation once we do not turn a Python side record
-    // to a Map.
     import scala.collection.JavaConversions._
     import scala.collection.convert.Wrappers.{JListWrapper, JMapWrapper}
 
@@ -494,55 +463,39 @@ class SQLContext(@transient val sparkContext: SparkContext)
         val converted = c.map { e => convert(e, elementType)}
         JListWrapper(converted)
 
-      case (c: java.util.Map[_, _], struct: StructType) =>
-        val row = new GenericMutableRow(struct.fields.length)
-        struct.fields.zipWithIndex.foreach {
-          case (field, i) =>
-            val value = convert(c.get(field.name), field.dataType)
-            row.update(i, value)
-        }
-        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, ArrayType(elementType, _)) if c.getClass.isArray =>
-        val converted = c.asInstanceOf[Array[_]].map(e => convert(e, elementType))
-        converted: Seq[Any]
+        c.asInstanceOf[Array[_]].map(e => convert(e, elementType)): Seq[Any]
+
+      case (c: java.util.Map[_, _], MapType(keyType, valueType, _)) => c.map {
+          case (key, value) => (convert(key, keyType), convert(value, valueType))
+        }.toMap
+
+      case (c, StructType(fields)) if c.getClass.isArray =>
+        new GenericRow(c.asInstanceOf[Array[_]].zip(fields).map {
+          case (e, f) => convert(e, f.dataType)
+        }): Row
+
+      case (c: java.util.Calendar, TimestampType) =>
+        new java.sql.Timestamp(c.getTime().getTime())
 
-      case (c: java.util.Calendar, TimestampType) => new java.sql.Timestamp(c.getTime().getTime())
       case (c: Int, ByteType) => c.toByte
       case (c: Int, ShortType) => c.toShort
       case (c: Double, FloatType) => c.toFloat
+      case (c, StringType) if !c.isInstanceOf[String] => c.toString
 
       case (c, _) => c
     }
 
     val convertedRdd = if (schema.fields.exists(f => needsConversion(f.dataType))) {
-      rdd.map(m => m.map { case (key, value) => (key, convert(value, schema(key).dataType)) })
+      rdd.map(m => m.zip(schema.fields).map {
+        case (value, field) => convert(value, field.dataType)
+      })
     } else {
       rdd
     }
 
     val rowRdd = convertedRdd.mapPartitions { iter =>
-      val row = new GenericMutableRow(schema.fields.length)
-      val fieldsWithIndex = schema.fields.zipWithIndex
-      iter.map { m =>
-        // We cannot use m.values because the order of values returned by m.values may not
-        // match fields order.
-        fieldsWithIndex.foreach {
-          case (field, i) =>
-            val value =
-              m.get(field.name).flatMap(v => Option(v)).map(v => convert(v, field.dataType)).orNull
-            row.update(i, value)
-        }
-
-        row: Row
-      }
+      iter.map { m => new GenericRow(m): Row}
     }
 
     new SchemaRDD(this, SparkLogicalPlan(ExistingRdd(schema.toAttributes, rowRdd))(self))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala
index 420f21fb9c..d34f62dc88 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala
@@ -383,7 +383,7 @@ class SchemaRDD(
     import scala.collection.Map
 
     def toJava(obj: Any, dataType: DataType): Any = dataType match {
-      case struct: StructType => rowToMap(obj.asInstanceOf[Row], struct)
+      case struct: StructType => rowToArray(obj.asInstanceOf[Row], struct)
       case array: ArrayType => obj match {
         case seq: Seq[Any] => seq.map(x => toJava(x, array.elementType)).asJava
         case list: JList[_] => list.map(x => toJava(x, array.elementType)).asJava
@@ -397,21 +397,19 @@ class SchemaRDD(
       // Pyrolite can handle Timestamp
       case other => obj
     }
-    def rowToMap(row: Row, structType: StructType): JMap[String, Any] = {
-      val fields = structType.fields.map(field => (field.name, field.dataType))
-      val map: JMap[String, Any] = new java.util.HashMap
-      row.zip(fields).foreach {
-        case (obj, (attrName, dataType)) => map.put(attrName, toJava(obj, dataType))
-      }
-      map
+    def rowToArray(row: Row, structType: StructType): Array[Any] = {
+      val fields = structType.fields.map(field => field.dataType)
+      row.zip(fields).map {
+        case (obj, dataType) => toJava(obj, dataType)
+      }.toArray
     }
 
     val rowSchema = StructType.fromAttributes(this.queryExecution.analyzed.output)
     this.mapPartitions { iter =>
       val pickle = new Pickler
       iter.map { row =>
-        rowToMap(row, rowSchema)
-      }.grouped(10).map(batched => pickle.dumps(batched.toArray))
+        rowToArray(row, rowSchema)
+      }.grouped(100).map(batched => pickle.dumps(batched.toArray))
     }
   }