compiling

Author: ahirreddy

core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala

@@ -25,6 +25,8 @@ import java.util.{List => JList, ArrayList => JArrayList, Map => JMap, Collectio
 import scala.collection.JavaConversions._
 import scala.reflect.ClassTag
 
+import net.razorvine.pickle.Unpickler
+
 import org.apache.spark._
 import org.apache.spark.api.java.{JavaSparkContext, JavaPairRDD, JavaRDD}
 import org.apache.spark.broadcast.Broadcast
@@ -284,6 +286,14 @@ private[spark] object PythonRDD {
     file.close()
   }
 
+  def pythonToJava(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[_] = {
+    pyRDD.rdd.mapPartitions { iter =>
+      val unpickle = new Unpickler
+      iter.map { row =>
+        unpickle.loads(row)
+      }
+    }
+  }
 }
 
 private

project/SparkBuild.scala

@@ -345,7 +345,8 @@ object SparkBuild extends Build {
         "com.twitter"               %% "chill"            % chillVersion excludeAll(excludeAsm),
         "com.twitter"                % "chill-java"       % chillVersion excludeAll(excludeAsm),
         "org.tachyonproject"         % "tachyon"          % "0.4.1-thrift" excludeAll(excludeHadoop, excludeCurator, excludeEclipseJetty, excludePowermock),
-        "com.clearspring.analytics"  % "stream"           % "2.5.1"
+        "com.clearspring.analytics"  % "stream"           % "2.5.1",
+        "net.razorvine" % "pyrolite_2.10" % "1.1"
       ),
     libraryDependencies ++= maybeAvro
   )

sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSQLContext.scala

@@ -82,6 +82,33 @@ class JavaSQLContext(sparkContext: JavaSparkContext) {
     new JavaSchemaRDD(sqlContext, SparkLogicalPlan(ExistingRdd(schema, rowRdd)))
   }
 
+  /**
+   * Applies a schema to an RDD of Array[Any]
+   */
+  def applySchema(rdd: JavaRDD[Array[Any]]): JavaSchemaRDD = {
+    val fields = rdd.first.map(_.getClass)
+    val schema = fields.zipWithIndex.map { case (klass, index) =>
+      val dataType = klass match {
+        case c: Class[_] if c == classOf[java.lang.String] => StringType
+        case c: Class[_] if c == java.lang.Short.TYPE => ShortType
+        case c: Class[_] if c == java.lang.Integer.TYPE => IntegerType
+        case c: Class[_] if c == java.lang.Long.TYPE => LongType
+        case c: Class[_] if c == java.lang.Double.TYPE => DoubleType
+        case c: Class[_] if c == java.lang.Byte.TYPE => ByteType
+        case c: Class[_] if c == java.lang.Float.TYPE => FloatType
+        case c: Class[_] if c == java.lang.Boolean.TYPE => BooleanType
+      }
+
+      AttributeReference(index.toString, dataType, true)()
+    }
+
+    val rowRdd = rdd.rdd.mapPartitions { iter =>
+      iter.map { row =>
+        new GenericRow(row): ScalaRow
+      }
+    }
+    new JavaSchemaRDD(sqlContext, SparkLogicalPlan(ExistingRdd(schema, rowRdd)))
+  }
 
   /**
    * Loads a parquet file, returning the result as a [[JavaSchemaRDD]].