Java API.

Author: yhuai

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

@@ -23,6 +23,7 @@ import org.apache.hadoop.conf.Configuration
 
 import org.apache.spark.annotation.Experimental
 import org.apache.spark.api.java.{JavaRDD, JavaSparkContext}
+import org.apache.spark.sql.json.JsonRDD
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericRow, Row => ScalaRow}
 import org.apache.spark.sql.catalyst.types._
@@ -100,13 +101,28 @@ class JavaSQLContext(val sqlContext: SQLContext) {
     new JavaSchemaRDD(sqlContext, SparkLogicalPlan(ExistingRdd(schema, rowRdd)))
   }
 
-
   /**
    * Loads a parquet file, returning the result as a [[JavaSchemaRDD]].
    */
   def parquetFile(path: String): JavaSchemaRDD =
     new JavaSchemaRDD(sqlContext, ParquetRelation(path))
 
+  /**
+   * Loads a JSON file (one object per line), returning the result as a [[JavaSchemaRDD]].
+   *
+   * @group userf
+   */
+  def jsonFile(path: String): JavaSchemaRDD =
+    jsonRDD(sqlContext.sparkContext.textFile(path))
+
+  /**
+   * Loads a RDD[String] storing JSON objects (one object per record), returning the result as a
+   * [[JavaSchemaRDD]].
+   *
+   * @group userf
+   */
+  def jsonRDD(json: JavaRDD[String]): JavaSchemaRDD =
+    new JavaSchemaRDD(sqlContext, JsonRDD.inferSchema(json, 1.0))
 
   /**
    * Registers the given RDD as a temporary table in the catalog.  Temporary tables exist only

sql/core/src/test/scala/org/apache/spark/sql/api/java/JavaSQLSuite.scala

@@ -17,11 +17,13 @@
 
 package org.apache.spark.sql.api.java
 
+
 import scala.beans.BeanProperty
 
 import org.scalatest.FunSuite
 
 import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.test.TestSQLContext
 
 // Implicits
@@ -111,4 +113,48 @@ class JavaSQLSuite extends FunSuite {
         """.stripMargin).collect.head.row ===
         Seq.fill(8)(null))
   }
+
+  test("loads JSON datasets") {
+    val jsonString =
+      """{"string":"this is a simple string.",
+          "integer":10,
+          "long":21474836470,
+          "bigInteger":92233720368547758070,
+          "double":1.7976931348623157E308,
+          "boolean":true,
+          "null":null
+      }""".replaceAll("\n", " ")
+    val rdd = javaCtx.parallelize(jsonString :: Nil)
+
+    var schemaRDD = javaSqlCtx.jsonRDD(rdd)
+
+    schemaRDD.registerAsTable("jsonTable1")
+
+    assert(
+      javaSqlCtx.sql("select * from jsonTable1").collect.head.row ===
+        Seq(BigDecimal("92233720368547758070"),
+            true,
+            1.7976931348623157E308,
+            10,
+            21474836470L,
+            null,
+            "this is a simple string."))
+
+    val file = getTempFilePath("json")
+    val path = file.toString
+    rdd.saveAsTextFile(path)
+    schemaRDD = javaSqlCtx.jsonFile(path)
+
+    schemaRDD.registerAsTable("jsonTable2")
+
+    assert(
+      javaSqlCtx.sql("select * from jsonTable2").collect.head.row ===
+        Seq(BigDecimal("92233720368547758070"),
+            true,
+            1.7976931348623157E308,
+            10,
+            21474836470L,
+            null,
+            "this is a simple string."))
+  }
 }