even better

witgo · Apr 15, 2014 · 4886052 · 4886052
1 parent c0fb1c6
commit 4886052
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 1 deletion.
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
@@ -1398,7 +1398,8 @@ def registerAsTable(self, name):
         self._jschema_rdd.registerAsTable(name)
 
     def toPython(self):
-        jrdd = self._sc._javaToPython(self._jschema_rdd)
+        jrdd = self._jschema_rdd.javaToPython()
+        #jrdd = self._sc._javaToPython(self._jschema_rdd)
         return RDD(jrdd, self._sc, self._sc.serializer)
 
 def _test():

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSchemaRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSchemaRDD.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.api.java
 
+import net.razorvine.pickle.{Pickler, Unpickler}
+
 import org.apache.spark.api.java.{JavaRDDLike, JavaRDD}
 import org.apache.spark.sql.{SQLContext, SchemaRDD, SchemaRDDLike}
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -45,4 +47,14 @@ class JavaSchemaRDD(
   override def wrapRDD(rdd: RDD[Row]): JavaRDD[Row] = JavaRDD.fromRDD(rdd)
 
   val rdd = baseSchemaRDD.map(new Row(_))
+
+  def javaToPython: JavaRDD[Array[Byte]] = {
+    this.rdd.mapPartitions { iter =>
+      val unpickle = new Pickler
+      iter.map { row =>
+        val fields: Array[Any] = (for (i <- 0 to row.length - 1) yield row.get(i)).toArray
+        unpickle.dumps(fields)
+      }
+    }
+  }
 }