[SPARK-5193][SQL] Remove Spark SQL Java-specific API. After the following patches, the main (Scala) API is now usable for Java users directly.
https://github.com/apache/spark/pull/4056 https://github.com/apache/spark/pull/4054 https://github.com/apache/spark/pull/4049 https://github.com/apache/spark/pull/4030 https://github.com/apache/spark/pull/3965 https://github.com/apache/spark/pull/3958 Author: Reynold Xin <r...@databricks.com> Closes #4065 from rxin/sql-java-api and squashes the following commits: b1fd860 [Reynold Xin] Fix Mima 6d86578 [Reynold Xin] Ok one more attempt in fixing Python... e8f1455 [Reynold Xin] Fix Python again... 3e53f91 [Reynold Xin] Fixed Python. 83735da [Reynold Xin] Fix BigDecimal test. e9f1de3 [Reynold Xin] Use scala BigDecimal. 500d2c4 [Reynold Xin] Fix Decimal. ba3bfa2 [Reynold Xin] Updated javadoc for RowFactory. c4ae1c5 [Reynold Xin] [SPARK-5193][SQL] Remove Spark SQL Java-specific API. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/61b427d4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/61b427d4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/61b427d4 Branch: refs/heads/master Commit: 61b427d4b1c4934bd70ed4da844b64f0e9a377aa Parents: ee1c1f3 Author: Reynold Xin <r...@databricks.com> Authored: Fri Jan 16 21:09:06 2015 -0800 Committer: Reynold Xin <r...@databricks.com> Committed: Fri Jan 16 21:09:06 2015 -0800 ---------------------------------------------------------------------- bin/spark-class | 2 +- .../examples/ml/JavaCrossValidatorExample.java | 15 +- .../examples/ml/JavaSimpleParamsExample.java | 14 +- .../JavaSimpleTextClassificationPipeline.java | 17 +- .../apache/spark/examples/sql/JavaSparkSQL.java | 32 +-- .../scala/org/apache/spark/ml/Estimator.scala | 38 --- .../scala/org/apache/spark/ml/Transformer.scala | 24 -- .../org/apache/spark/ml/JavaPipelineSuite.java | 17 +- .../JavaLogisticRegressionSuite.java | 21 +- .../ml/tuning/JavaCrossValidatorSuite.java | 13 +- project/MimaExcludes.scala | 4 + python/pyspark/sql.py | 48 +--- .../java/org/apache/spark/sql/RowFactory.java | 6 +- .../org/apache/spark/sql/types/Decimal.scala | 2 + .../scala/org/apache/spark/sql/SQLContext.scala | 4 +- .../scala/org/apache/spark/sql/SchemaRDD.scala | 16 +- .../spark/sql/api/java/JavaSQLContext.scala | 241 ------------------ .../spark/sql/api/java/JavaSchemaRDD.scala | 225 ----------------- .../org/apache/spark/sql/api/java/Row.scala | 153 ----------- .../spark/sql/api/java/UDFRegistration.scala | 251 ------------------- .../apache/spark/sql/api/java/JavaAPISuite.java | 16 +- .../sql/api/java/JavaApplySchemaSuite.java | 34 +-- .../apache/spark/sql/api/java/JavaRowSuite.java | 11 +- .../sql/api/java/JavaUserDefinedTypeSuite.java | 88 ------- .../spark/sql/api/java/JavaSQLSuite.scala | 209 --------------- .../sql/hive/api/java/JavaHiveContext.scala | 49 ---- .../sql/hive/api/java/JavaHiveQLSuite.scala | 91 ------- 27 files changed, 125 insertions(+), 1516 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/bin/spark-class ---------------------------------------------------------------------- diff --git a/bin/spark-class b/bin/spark-class index 0d58d95..79af42c 100755 --- a/bin/spark-class +++ b/bin/spark-class @@ -148,7 +148,7 @@ fi if [[ "$1" =~ org.apache.spark.tools.* ]]; then if test -z "$SPARK_TOOLS_JAR"; then echo "Failed to find Spark Tools Jar in $FWDIR/tools/target/scala-$SPARK_SCALA_VERSION/" 1>&2 - echo "You need to build Spark before running $1." 1>&2 + echo "You need to run \"build/sbt tools/package\" before running $1." 1>&2 exit 1 fi CLASSPATH="$CLASSPATH:$SPARK_TOOLS_JAR" http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/examples/src/main/java/org/apache/spark/examples/ml/JavaCrossValidatorExample.java ---------------------------------------------------------------------- diff --git a/examples/src/main/java/org/apache/spark/examples/ml/JavaCrossValidatorExample.java b/examples/src/main/java/org/apache/spark/examples/ml/JavaCrossValidatorExample.java index f4b4f8d..247d2a5 100644 --- a/examples/src/main/java/org/apache/spark/examples/ml/JavaCrossValidatorExample.java +++ b/examples/src/main/java/org/apache/spark/examples/ml/JavaCrossValidatorExample.java @@ -33,9 +33,9 @@ import org.apache.spark.ml.param.ParamMap; import org.apache.spark.ml.tuning.CrossValidator; import org.apache.spark.ml.tuning.CrossValidatorModel; import org.apache.spark.ml.tuning.ParamGridBuilder; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import org.apache.spark.sql.api.java.Row; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.SQLContext; +import org.apache.spark.sql.Row; /** * A simple example demonstrating model selection using CrossValidator. @@ -55,7 +55,7 @@ public class JavaCrossValidatorExample { public static void main(String[] args) { SparkConf conf = new SparkConf().setAppName("JavaCrossValidatorExample"); JavaSparkContext jsc = new JavaSparkContext(conf); - JavaSQLContext jsql = new JavaSQLContext(jsc); + SQLContext jsql = new SQLContext(jsc); // Prepare training documents, which are labeled. List<LabeledDocument> localTraining = Lists.newArrayList( @@ -71,8 +71,7 @@ public class JavaCrossValidatorExample { new LabeledDocument(9L, "a e c l", 0.0), new LabeledDocument(10L, "spark compile", 1.0), new LabeledDocument(11L, "hadoop software", 0.0)); - JavaSchemaRDD training = - jsql.applySchema(jsc.parallelize(localTraining), LabeledDocument.class); + SchemaRDD training = jsql.applySchema(jsc.parallelize(localTraining), LabeledDocument.class); // Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr. Tokenizer tokenizer = new Tokenizer() @@ -113,11 +112,11 @@ public class JavaCrossValidatorExample { new Document(5L, "l m n"), new Document(6L, "mapreduce spark"), new Document(7L, "apache hadoop")); - JavaSchemaRDD test = jsql.applySchema(jsc.parallelize(localTest), Document.class); + SchemaRDD test = jsql.applySchema(jsc.parallelize(localTest), Document.class); // Make predictions on test documents. cvModel uses the best model found (lrModel). cvModel.transform(test).registerAsTable("prediction"); - JavaSchemaRDD predictions = jsql.sql("SELECT id, text, score, prediction FROM prediction"); + SchemaRDD predictions = jsql.sql("SELECT id, text, score, prediction FROM prediction"); for (Row r: predictions.collect()) { System.out.println("(" + r.get(0) + ", " + r.get(1) + ") --> score=" + r.get(2) + ", prediction=" + r.get(3)); http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleParamsExample.java ---------------------------------------------------------------------- diff --git a/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleParamsExample.java b/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleParamsExample.java index e25b271..5b92655 100644 --- a/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleParamsExample.java +++ b/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleParamsExample.java @@ -28,9 +28,9 @@ import org.apache.spark.ml.param.ParamMap; import org.apache.spark.ml.classification.LogisticRegression; import org.apache.spark.mllib.linalg.Vectors; import org.apache.spark.mllib.regression.LabeledPoint; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import org.apache.spark.sql.api.java.Row; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.SQLContext; +import org.apache.spark.sql.Row; /** * A simple example demonstrating ways to specify parameters for Estimators and Transformers. @@ -44,7 +44,7 @@ public class JavaSimpleParamsExample { public static void main(String[] args) { SparkConf conf = new SparkConf().setAppName("JavaSimpleParamsExample"); JavaSparkContext jsc = new JavaSparkContext(conf); - JavaSQLContext jsql = new JavaSQLContext(jsc); + SQLContext jsql = new SQLContext(jsc); // Prepare training data. // We use LabeledPoint, which is a JavaBean. Spark SQL can convert RDDs of JavaBeans @@ -54,7 +54,7 @@ public class JavaSimpleParamsExample { new LabeledPoint(0.0, Vectors.dense(2.0, 1.0, -1.0)), new LabeledPoint(0.0, Vectors.dense(2.0, 1.3, 1.0)), new LabeledPoint(1.0, Vectors.dense(0.0, 1.2, -0.5))); - JavaSchemaRDD training = jsql.applySchema(jsc.parallelize(localTraining), LabeledPoint.class); + SchemaRDD training = jsql.applySchema(jsc.parallelize(localTraining), LabeledPoint.class); // Create a LogisticRegression instance. This instance is an Estimator. LogisticRegression lr = new LogisticRegression(); @@ -94,14 +94,14 @@ public class JavaSimpleParamsExample { new LabeledPoint(1.0, Vectors.dense(-1.0, 1.5, 1.3)), new LabeledPoint(0.0, Vectors.dense(3.0, 2.0, -0.1)), new LabeledPoint(1.0, Vectors.dense(0.0, 2.2, -1.5))); - JavaSchemaRDD test = jsql.applySchema(jsc.parallelize(localTest), LabeledPoint.class); + SchemaRDD test = jsql.applySchema(jsc.parallelize(localTest), LabeledPoint.class); // Make predictions on test documents using the Transformer.transform() method. // LogisticRegression.transform will only use the 'features' column. // Note that model2.transform() outputs a 'probability' column instead of the usual 'score' // column since we renamed the lr.scoreCol parameter previously. model2.transform(test).registerAsTable("results"); - JavaSchemaRDD results = + SchemaRDD results = jsql.sql("SELECT features, label, probability, prediction FROM results"); for (Row r: results.collect()) { System.out.println("(" + r.get(0) + ", " + r.get(1) + ") -> prob=" + r.get(2) http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleTextClassificationPipeline.java ---------------------------------------------------------------------- diff --git a/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleTextClassificationPipeline.java b/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleTextClassificationPipeline.java index 54f1801..74db449 100644 --- a/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleTextClassificationPipeline.java +++ b/examples/src/main/java/org/apache/spark/examples/ml/JavaSimpleTextClassificationPipeline.java @@ -21,6 +21,7 @@ import java.util.List; import com.google.common.collect.Lists; +import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.ml.Pipeline; import org.apache.spark.ml.PipelineModel; @@ -28,10 +29,9 @@ import org.apache.spark.ml.PipelineStage; import org.apache.spark.ml.classification.LogisticRegression; import org.apache.spark.ml.feature.HashingTF; import org.apache.spark.ml.feature.Tokenizer; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import org.apache.spark.sql.api.java.Row; -import org.apache.spark.SparkConf; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.SQLContext; +import org.apache.spark.sql.Row; /** * A simple text classification pipeline that recognizes "spark" from input text. It uses the Java @@ -46,7 +46,7 @@ public class JavaSimpleTextClassificationPipeline { public static void main(String[] args) { SparkConf conf = new SparkConf().setAppName("JavaSimpleTextClassificationPipeline"); JavaSparkContext jsc = new JavaSparkContext(conf); - JavaSQLContext jsql = new JavaSQLContext(jsc); + SQLContext jsql = new SQLContext(jsc); // Prepare training documents, which are labeled. List<LabeledDocument> localTraining = Lists.newArrayList( @@ -54,8 +54,7 @@ public class JavaSimpleTextClassificationPipeline { new LabeledDocument(1L, "b d", 0.0), new LabeledDocument(2L, "spark f g h", 1.0), new LabeledDocument(3L, "hadoop mapreduce", 0.0)); - JavaSchemaRDD training = - jsql.applySchema(jsc.parallelize(localTraining), LabeledDocument.class); + SchemaRDD training = jsql.applySchema(jsc.parallelize(localTraining), LabeledDocument.class); // Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr. Tokenizer tokenizer = new Tokenizer() @@ -80,11 +79,11 @@ public class JavaSimpleTextClassificationPipeline { new Document(5L, "l m n"), new Document(6L, "mapreduce spark"), new Document(7L, "apache hadoop")); - JavaSchemaRDD test = jsql.applySchema(jsc.parallelize(localTest), Document.class); + SchemaRDD test = jsql.applySchema(jsc.parallelize(localTest), Document.class); // Make predictions on test documents. model.transform(test).registerAsTable("prediction"); - JavaSchemaRDD predictions = jsql.sql("SELECT id, text, score, prediction FROM prediction"); + SchemaRDD predictions = jsql.sql("SELECT id, text, score, prediction FROM prediction"); for (Row r: predictions.collect()) { System.out.println("(" + r.get(0) + ", " + r.get(1) + ") --> score=" + r.get(2) + ", prediction=" + r.get(3)); http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/examples/src/main/java/org/apache/spark/examples/sql/JavaSparkSQL.java ---------------------------------------------------------------------- diff --git a/examples/src/main/java/org/apache/spark/examples/sql/JavaSparkSQL.java b/examples/src/main/java/org/apache/spark/examples/sql/JavaSparkSQL.java index 01c77bd..b708046 100644 --- a/examples/src/main/java/org/apache/spark/examples/sql/JavaSparkSQL.java +++ b/examples/src/main/java/org/apache/spark/examples/sql/JavaSparkSQL.java @@ -26,9 +26,9 @@ import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import org.apache.spark.sql.api.java.Row; +import org.apache.spark.sql.SQLContext; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.Row; public class JavaSparkSQL { public static class Person implements Serializable { @@ -55,7 +55,7 @@ public class JavaSparkSQL { public static void main(String[] args) throws Exception { SparkConf sparkConf = new SparkConf().setAppName("JavaSparkSQL"); JavaSparkContext ctx = new JavaSparkContext(sparkConf); - JavaSQLContext sqlCtx = new JavaSQLContext(ctx); + SQLContext sqlCtx = new SQLContext(ctx); System.out.println("=== Data source: RDD ==="); // Load a text file and convert each line to a Java Bean. @@ -74,15 +74,15 @@ public class JavaSparkSQL { }); // Apply a schema to an RDD of Java Beans and register it as a table. - JavaSchemaRDD schemaPeople = sqlCtx.applySchema(people, Person.class); + SchemaRDD schemaPeople = sqlCtx.applySchema(people, Person.class); schemaPeople.registerTempTable("people"); // SQL can be run over RDDs that have been registered as tables. - JavaSchemaRDD teenagers = sqlCtx.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19"); + SchemaRDD teenagers = sqlCtx.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19"); // The results of SQL queries are SchemaRDDs and support all the normal RDD operations. // The columns of a row in the result can be accessed by ordinal. - List<String> teenagerNames = teenagers.map(new Function<Row, String>() { + List<String> teenagerNames = teenagers.toJavaRDD().map(new Function<Row, String>() { @Override public String call(Row row) { return "Name: " + row.getString(0); @@ -99,13 +99,13 @@ public class JavaSparkSQL { // Read in the parquet file created above. // Parquet files are self-describing so the schema is preserved. // The result of loading a parquet file is also a JavaSchemaRDD. - JavaSchemaRDD parquetFile = sqlCtx.parquetFile("people.parquet"); + SchemaRDD parquetFile = sqlCtx.parquetFile("people.parquet"); //Parquet files can also be registered as tables and then used in SQL statements. parquetFile.registerTempTable("parquetFile"); - JavaSchemaRDD teenagers2 = + SchemaRDD teenagers2 = sqlCtx.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19"); - teenagerNames = teenagers2.map(new Function<Row, String>() { + teenagerNames = teenagers2.toJavaRDD().map(new Function<Row, String>() { @Override public String call(Row row) { return "Name: " + row.getString(0); @@ -120,7 +120,7 @@ public class JavaSparkSQL { // The path can be either a single text file or a directory storing text files. String path = "examples/src/main/resources/people.json"; // Create a JavaSchemaRDD from the file(s) pointed by path - JavaSchemaRDD peopleFromJsonFile = sqlCtx.jsonFile(path); + SchemaRDD peopleFromJsonFile = sqlCtx.jsonFile(path); // Because the schema of a JSON dataset is automatically inferred, to write queries, // it is better to take a look at what is the schema. @@ -134,11 +134,11 @@ public class JavaSparkSQL { peopleFromJsonFile.registerTempTable("people"); // SQL statements can be run by using the sql methods provided by sqlCtx. - JavaSchemaRDD teenagers3 = sqlCtx.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19"); + SchemaRDD teenagers3 = sqlCtx.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19"); // The results of SQL queries are JavaSchemaRDDs and support all the normal RDD operations. // The columns of a row in the result can be accessed by ordinal. - teenagerNames = teenagers3.map(new Function<Row, String>() { + teenagerNames = teenagers3.toJavaRDD().map(new Function<Row, String>() { @Override public String call(Row row) { return "Name: " + row.getString(0); } }).collect(); @@ -151,7 +151,7 @@ public class JavaSparkSQL { List<String> jsonData = Arrays.asList( "{\"name\":\"Yin\",\"address\":{\"city\":\"Columbus\",\"state\":\"Ohio\"}}"); JavaRDD<String> anotherPeopleRDD = ctx.parallelize(jsonData); - JavaSchemaRDD peopleFromJsonRDD = sqlCtx.jsonRDD(anotherPeopleRDD); + SchemaRDD peopleFromJsonRDD = sqlCtx.jsonRDD(anotherPeopleRDD.rdd()); // Take a look at the schema of this new JavaSchemaRDD. peopleFromJsonRDD.printSchema(); @@ -164,8 +164,8 @@ public class JavaSparkSQL { peopleFromJsonRDD.registerTempTable("people2"); - JavaSchemaRDD peopleWithCity = sqlCtx.sql("SELECT name, address.city FROM people2"); - List<String> nameAndCity = peopleWithCity.map(new Function<Row, String>() { + SchemaRDD peopleWithCity = sqlCtx.sql("SELECT name, address.city FROM people2"); + List<String> nameAndCity = peopleWithCity.toJavaRDD().map(new Function<Row, String>() { @Override public String call(Row row) { return "Name: " + row.getString(0) + ", City: " + row.getString(1); http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/mllib/src/main/scala/org/apache/spark/ml/Estimator.scala ---------------------------------------------------------------------- diff --git a/mllib/src/main/scala/org/apache/spark/ml/Estimator.scala b/mllib/src/main/scala/org/apache/spark/ml/Estimator.scala index fdbee74..77d230e 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/Estimator.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/Estimator.scala @@ -18,12 +18,10 @@ package org.apache.spark.ml import scala.annotation.varargs -import scala.collection.JavaConverters._ import org.apache.spark.annotation.AlphaComponent import org.apache.spark.ml.param.{ParamMap, ParamPair, Params} import org.apache.spark.sql.SchemaRDD -import org.apache.spark.sql.api.java.JavaSchemaRDD /** * :: AlphaComponent :: @@ -66,40 +64,4 @@ abstract class Estimator[M <: Model[M]] extends PipelineStage with Params { def fit(dataset: SchemaRDD, paramMaps: Array[ParamMap]): Seq[M] = { paramMaps.map(fit(dataset, _)) } - - // Java-friendly versions of fit. - - /** - * Fits a single model to the input data with optional parameters. - * - * @param dataset input dataset - * @param paramPairs optional list of param pairs (overwrite embedded params) - * @return fitted model - */ - @varargs - def fit(dataset: JavaSchemaRDD, paramPairs: ParamPair[_]*): M = { - fit(dataset.schemaRDD, paramPairs: _*) - } - - /** - * Fits a single model to the input data with provided parameter map. - * - * @param dataset input dataset - * @param paramMap parameter map - * @return fitted model - */ - def fit(dataset: JavaSchemaRDD, paramMap: ParamMap): M = { - fit(dataset.schemaRDD, paramMap) - } - - /** - * Fits multiple models to the input data with multiple sets of parameters. - * - * @param dataset input dataset - * @param paramMaps an array of parameter maps - * @return fitted models, matching the input parameter maps - */ - def fit(dataset: JavaSchemaRDD, paramMaps: Array[ParamMap]): java.util.List[M] = { - fit(dataset.schemaRDD, paramMaps).asJava - } } http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/mllib/src/main/scala/org/apache/spark/ml/Transformer.scala ---------------------------------------------------------------------- diff --git a/mllib/src/main/scala/org/apache/spark/ml/Transformer.scala b/mllib/src/main/scala/org/apache/spark/ml/Transformer.scala index 1331b91..af56f9c 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/Transformer.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/Transformer.scala @@ -23,7 +23,6 @@ import org.apache.spark.Logging import org.apache.spark.annotation.AlphaComponent import org.apache.spark.ml.param._ import org.apache.spark.sql.SchemaRDD -import org.apache.spark.sql.api.java.JavaSchemaRDD import org.apache.spark.sql.catalyst.analysis.Star import org.apache.spark.sql.catalyst.expressions.ScalaUdf import org.apache.spark.sql.types._ @@ -55,29 +54,6 @@ abstract class Transformer extends PipelineStage with Params { * @return transformed dataset */ def transform(dataset: SchemaRDD, paramMap: ParamMap): SchemaRDD - - // Java-friendly versions of transform. - - /** - * Transforms the dataset with optional parameters. - * @param dataset input datset - * @param paramPairs optional list of param pairs, overwrite embedded params - * @return transformed dataset - */ - @varargs - def transform(dataset: JavaSchemaRDD, paramPairs: ParamPair[_]*): JavaSchemaRDD = { - transform(dataset.schemaRDD, paramPairs: _*).toJavaSchemaRDD - } - - /** - * Transforms the dataset with provided parameter map as additional parameters. - * @param dataset input dataset - * @param paramMap additional parameters, overwrite embedded params - * @return transformed dataset - */ - def transform(dataset: JavaSchemaRDD, paramMap: ParamMap): JavaSchemaRDD = { - transform(dataset.schemaRDD, paramMap).toJavaSchemaRDD - } } /** http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/mllib/src/test/java/org/apache/spark/ml/JavaPipelineSuite.java ---------------------------------------------------------------------- diff --git a/mllib/src/test/java/org/apache/spark/ml/JavaPipelineSuite.java b/mllib/src/test/java/org/apache/spark/ml/JavaPipelineSuite.java index 4284667..47f1f46 100644 --- a/mllib/src/test/java/org/apache/spark/ml/JavaPipelineSuite.java +++ b/mllib/src/test/java/org/apache/spark/ml/JavaPipelineSuite.java @@ -26,10 +26,9 @@ import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.mllib.regression.LabeledPoint; import org.apache.spark.ml.classification.LogisticRegression; import org.apache.spark.ml.feature.StandardScaler; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import static org.apache.spark.mllib.classification.LogisticRegressionSuite - .generateLogisticInputAsList; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.SQLContext; +import static org.apache.spark.mllib.classification.LogisticRegressionSuite.generateLogisticInputAsList; /** * Test Pipeline construction and fitting in Java. @@ -37,13 +36,13 @@ import static org.apache.spark.mllib.classification.LogisticRegressionSuite public class JavaPipelineSuite { private transient JavaSparkContext jsc; - private transient JavaSQLContext jsql; - private transient JavaSchemaRDD dataset; + private transient SQLContext jsql; + private transient SchemaRDD dataset; @Before public void setUp() { jsc = new JavaSparkContext("local", "JavaPipelineSuite"); - jsql = new JavaSQLContext(jsc); + jsql = new SQLContext(jsc); JavaRDD<LabeledPoint> points = jsc.parallelize(generateLogisticInputAsList(1.0, 1.0, 100, 42), 2); dataset = jsql.applySchema(points, LabeledPoint.class); @@ -66,7 +65,7 @@ public class JavaPipelineSuite { .setStages(new PipelineStage[] {scaler, lr}); PipelineModel model = pipeline.fit(dataset); model.transform(dataset).registerTempTable("prediction"); - JavaSchemaRDD predictions = jsql.sql("SELECT label, score, prediction FROM prediction"); - predictions.collect(); + SchemaRDD predictions = jsql.sql("SELECT label, score, prediction FROM prediction"); + predictions.collectAsList(); } } http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/mllib/src/test/java/org/apache/spark/ml/classification/JavaLogisticRegressionSuite.java ---------------------------------------------------------------------- diff --git a/mllib/src/test/java/org/apache/spark/ml/classification/JavaLogisticRegressionSuite.java b/mllib/src/test/java/org/apache/spark/ml/classification/JavaLogisticRegressionSuite.java index 76eb7f0..2eba833 100644 --- a/mllib/src/test/java/org/apache/spark/ml/classification/JavaLogisticRegressionSuite.java +++ b/mllib/src/test/java/org/apache/spark/ml/classification/JavaLogisticRegressionSuite.java @@ -26,21 +26,20 @@ import org.junit.Test; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.mllib.regression.LabeledPoint; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import static org.apache.spark.mllib.classification.LogisticRegressionSuite - .generateLogisticInputAsList; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.SQLContext; +import static org.apache.spark.mllib.classification.LogisticRegressionSuite.generateLogisticInputAsList; public class JavaLogisticRegressionSuite implements Serializable { private transient JavaSparkContext jsc; - private transient JavaSQLContext jsql; - private transient JavaSchemaRDD dataset; + private transient SQLContext jsql; + private transient SchemaRDD dataset; @Before public void setUp() { jsc = new JavaSparkContext("local", "JavaLogisticRegressionSuite"); - jsql = new JavaSQLContext(jsc); + jsql = new SQLContext(jsc); List<LabeledPoint> points = generateLogisticInputAsList(1.0, 1.0, 100, 42); dataset = jsql.applySchema(jsc.parallelize(points, 2), LabeledPoint.class); } @@ -56,8 +55,8 @@ public class JavaLogisticRegressionSuite implements Serializable { LogisticRegression lr = new LogisticRegression(); LogisticRegressionModel model = lr.fit(dataset); model.transform(dataset).registerTempTable("prediction"); - JavaSchemaRDD predictions = jsql.sql("SELECT label, score, prediction FROM prediction"); - predictions.collect(); + SchemaRDD predictions = jsql.sql("SELECT label, score, prediction FROM prediction"); + predictions.collectAsList(); } @Test @@ -68,8 +67,8 @@ public class JavaLogisticRegressionSuite implements Serializable { LogisticRegressionModel model = lr.fit(dataset); model.transform(dataset, model.threshold().w(0.8)) // overwrite threshold .registerTempTable("prediction"); - JavaSchemaRDD predictions = jsql.sql("SELECT label, score, prediction FROM prediction"); - predictions.collect(); + SchemaRDD predictions = jsql.sql("SELECT label, score, prediction FROM prediction"); + predictions.collectAsList(); } @Test http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/mllib/src/test/java/org/apache/spark/ml/tuning/JavaCrossValidatorSuite.java ---------------------------------------------------------------------- diff --git a/mllib/src/test/java/org/apache/spark/ml/tuning/JavaCrossValidatorSuite.java b/mllib/src/test/java/org/apache/spark/ml/tuning/JavaCrossValidatorSuite.java index a266ebd..a9f1c4a 100644 --- a/mllib/src/test/java/org/apache/spark/ml/tuning/JavaCrossValidatorSuite.java +++ b/mllib/src/test/java/org/apache/spark/ml/tuning/JavaCrossValidatorSuite.java @@ -30,21 +30,20 @@ import org.apache.spark.ml.classification.LogisticRegression; import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator; import org.apache.spark.ml.param.ParamMap; import org.apache.spark.mllib.regression.LabeledPoint; -import org.apache.spark.sql.api.java.JavaSQLContext; -import org.apache.spark.sql.api.java.JavaSchemaRDD; -import static org.apache.spark.mllib.classification.LogisticRegressionSuite - .generateLogisticInputAsList; +import org.apache.spark.sql.SchemaRDD; +import org.apache.spark.sql.SQLContext; +import static org.apache.spark.mllib.classification.LogisticRegressionSuite.generateLogisticInputAsList; public class JavaCrossValidatorSuite implements Serializable { private transient JavaSparkContext jsc; - private transient JavaSQLContext jsql; - private transient JavaSchemaRDD dataset; + private transient SQLContext jsql; + private transient SchemaRDD dataset; @Before public void setUp() { jsc = new JavaSparkContext("local", "JavaCrossValidatorSuite"); - jsql = new JavaSQLContext(jsc); + jsql = new SQLContext(jsc); List<LabeledPoint> points = generateLogisticInputAsList(1.0, 1.0, 100, 42); dataset = jsql.applySchema(jsc.parallelize(points, 2), LabeledPoint.class); } http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/project/MimaExcludes.scala ---------------------------------------------------------------------- diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index d3ea594..0ccbfcb 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -78,6 +78,10 @@ object MimaExcludes { "org.apache.spark.TaskContext.taskAttemptId"), ProblemFilters.exclude[MissingMethodProblem]( "org.apache.spark.TaskContext.attemptNumber") + ) ++ Seq( + // SPARK-5166 Spark SQL API stabilization + ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.ml.Transformer.transform"), + ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.ml.Estimator.fit") ) case v if v.startsWith("1.2") => http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/python/pyspark/sql.py ---------------------------------------------------------------------- diff --git a/python/pyspark/sql.py b/python/pyspark/sql.py index dcd3b60..1990323 100644 --- a/python/pyspark/sql.py +++ b/python/pyspark/sql.py @@ -1458,7 +1458,7 @@ class SQLContext(object): jrdd = self._jvm.SerDeUtil.toJavaArray(rdd._to_java_object_rdd()) srdd = self._ssql_ctx.applySchemaToPythonRDD(jrdd.rdd(), schema.json()) - return SchemaRDD(srdd.toJavaSchemaRDD(), self) + return SchemaRDD(srdd, self) def registerRDDAsTable(self, rdd, tableName): """Registers the given RDD as a temporary table in the catalog. @@ -1487,7 +1487,7 @@ class SQLContext(object): >>> sorted(srdd.collect()) == sorted(srdd2.collect()) True """ - jschema_rdd = self._ssql_ctx.parquetFile(path).toJavaSchemaRDD() + jschema_rdd = self._ssql_ctx.parquetFile(path) return SchemaRDD(jschema_rdd, self) def jsonFile(self, path, schema=None, samplingRatio=1.0): @@ -1549,7 +1549,7 @@ class SQLContext(object): else: scala_datatype = self._ssql_ctx.parseDataType(schema.json()) srdd = self._ssql_ctx.jsonFile(path, scala_datatype) - return SchemaRDD(srdd.toJavaSchemaRDD(), self) + return SchemaRDD(srdd, self) def jsonRDD(self, rdd, schema=None, samplingRatio=1.0): """Loads an RDD storing one JSON object per string as a L{SchemaRDD}. @@ -1619,7 +1619,7 @@ class SQLContext(object): else: scala_datatype = self._ssql_ctx.parseDataType(schema.json()) srdd = self._ssql_ctx.jsonRDD(jrdd.rdd(), scala_datatype) - return SchemaRDD(srdd.toJavaSchemaRDD(), self) + return SchemaRDD(srdd, self) def sql(self, sqlQuery): """Return a L{SchemaRDD} representing the result of the given query. @@ -1630,7 +1630,7 @@ class SQLContext(object): >>> 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).toJavaSchemaRDD(), self) + return SchemaRDD(self._ssql_ctx.sql(sqlQuery), self) def table(self, tableName): """Returns the specified table as a L{SchemaRDD}. @@ -1641,7 +1641,7 @@ class SQLContext(object): >>> sorted(srdd.collect()) == sorted(srdd2.collect()) True """ - return SchemaRDD(self._ssql_ctx.table(tableName).toJavaSchemaRDD(), self) + return SchemaRDD(self._ssql_ctx.table(tableName), self) def cacheTable(self, tableName): """Caches the specified table in-memory.""" @@ -1686,24 +1686,6 @@ class HiveContext(SQLContext): def _get_hive_ctx(self): return self._jvm.HiveContext(self._jsc.sc()) - def hiveql(self, hqlQuery): - """ - DEPRECATED: Use sql() - """ - warnings.warn("hiveql() is deprecated as the sql function now parses using HiveQL by" + - "default. The SQL dialect for parsing can be set using 'spark.sql.dialect'", - DeprecationWarning) - return SchemaRDD(self._ssql_ctx.hiveql(hqlQuery).toJavaSchemaRDD(), self) - - def hql(self, hqlQuery): - """ - DEPRECATED: Use sql() - """ - warnings.warn("hql() is deprecated as the sql function now parses using HiveQL by" + - "default. The SQL dialect for parsing can be set using 'spark.sql.dialect'", - DeprecationWarning) - return self.hiveql(hqlQuery) - class LocalHiveContext(HiveContext): @@ -1716,12 +1698,6 @@ class LocalHiveContext(HiveContext): return self._jvm.LocalHiveContext(self._jsc.sc()) -class TestHiveContext(HiveContext): - - def _get_hive_ctx(self): - return self._jvm.TestHiveContext(self._jsc.sc()) - - def _create_row(fields, values): row = Row(*values) row.__FIELDS__ = fields @@ -1846,7 +1822,7 @@ class SchemaRDD(RDD): self.sql_ctx = sql_ctx self._sc = sql_ctx._sc clsName = jschema_rdd.getClass().getName() - assert clsName.endswith("JavaSchemaRDD"), "jschema_rdd must be JavaSchemaRDD" + assert clsName.endswith("SchemaRDD"), "jschema_rdd must be SchemaRDD" self._jschema_rdd = jschema_rdd self._id = None self.is_cached = False @@ -1880,7 +1856,7 @@ class SchemaRDD(RDD): >>> srdd.limit(0).collect() [] """ - rdd = self._jschema_rdd.baseSchemaRDD().limit(num).toJavaSchemaRDD() + rdd = self._jschema_rdd.baseSchemaRDD().limit(num) return SchemaRDD(rdd, self.sql_ctx) def toJSON(self, use_unicode=False): @@ -2059,18 +2035,18 @@ class SchemaRDD(RDD): def getCheckpointFile(self): checkpointFile = self._jschema_rdd.getCheckpointFile() - if checkpointFile.isPresent(): + if checkpointFile.isDefined(): return checkpointFile.get() def coalesce(self, numPartitions, shuffle=False): - rdd = self._jschema_rdd.coalesce(numPartitions, shuffle) + rdd = self._jschema_rdd.coalesce(numPartitions, shuffle, None) return SchemaRDD(rdd, self.sql_ctx) def distinct(self, numPartitions=None): if numPartitions is None: rdd = self._jschema_rdd.distinct() else: - rdd = self._jschema_rdd.distinct(numPartitions) + rdd = self._jschema_rdd.distinct(numPartitions, None) return SchemaRDD(rdd, self.sql_ctx) def intersection(self, other): @@ -2081,7 +2057,7 @@ class SchemaRDD(RDD): raise ValueError("Can only intersect with another SchemaRDD") def repartition(self, numPartitions): - rdd = self._jschema_rdd.repartition(numPartitions) + rdd = self._jschema_rdd.repartition(numPartitions, None) return SchemaRDD(rdd, self.sql_ctx) def subtract(self, other, numPartitions=None): http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/catalyst/src/main/java/org/apache/spark/sql/RowFactory.java ---------------------------------------------------------------------- diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/RowFactory.java b/sql/catalyst/src/main/java/org/apache/spark/sql/RowFactory.java index 62fcec8..5ed60fe 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/RowFactory.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/RowFactory.java @@ -25,10 +25,10 @@ import org.apache.spark.sql.catalyst.expressions.GenericRow; public class RowFactory { /** - * Create a {@link Row} from an array of values. Position i in the array becomes position i - * in the created {@link Row} object. + * Create a {@link Row} from the given arguments. Position i in the argument list becomes + * position i in the created {@link Row} object. */ - public static Row create(Object[] values) { + public static Row create(Object ... values) { return new GenericRow(values); } } http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala ---------------------------------------------------------------------- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala index 3744d77..a85c431 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala @@ -143,6 +143,8 @@ final class Decimal extends Ordered[Decimal] with Serializable { } } + def toJavaBigDecimal: java.math.BigDecimal = toBigDecimal.bigDecimal + def toUnscaledLong: Long = { if (decimalVal.ne(null)) { decimalVal.underlying().unscaledValue().longValue() http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---------------------------------------------------------------------- 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 8ad1753..f23cb18 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 @@ -26,7 +26,7 @@ import scala.reflect.runtime.universe.TypeTag import org.apache.spark.SparkContext import org.apache.spark.annotation.{AlphaComponent, DeveloperApi, Experimental} -import org.apache.spark.api.java.JavaRDD +import org.apache.spark.api.java.{JavaSparkContext, JavaRDD} import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.ScalaReflection import org.apache.spark.sql.catalyst.analysis._ @@ -58,6 +58,8 @@ class SQLContext(@transient val sparkContext: SparkContext) self => + def this(sparkContext: JavaSparkContext) = this(sparkContext.sc) + // Note that this is a lazy val so we can override the default value in subclasses. protected[sql] lazy val conf: SQLConf = new SQLConf http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala ---------------------------------------------------------------------- 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 686bcdf..ae4d8ba 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 @@ -30,7 +30,6 @@ import org.apache.spark.annotation.{AlphaComponent, Experimental} import org.apache.spark.api.java.JavaRDD import org.apache.spark.api.python.SerDeUtil import org.apache.spark.rdd.RDD -import org.apache.spark.sql.api.java.JavaSchemaRDD import org.apache.spark.sql.catalyst.ScalaReflection import org.apache.spark.sql.catalyst.analysis._ import org.apache.spark.sql.catalyst.expressions._ @@ -409,13 +408,6 @@ class SchemaRDD( def toSchemaRDD = this /** - * Returns this RDD as a JavaSchemaRDD. - * - * @group schema - */ - def toJavaSchemaRDD: JavaSchemaRDD = new JavaSchemaRDD(sqlContext, logicalPlan) - - /** * Converts a JavaRDD to a PythonRDD. It is used by pyspark. */ private[sql] def javaToPython: JavaRDD[Array[Byte]] = { @@ -470,6 +462,8 @@ class SchemaRDD( override def collect(): Array[Row] = queryExecution.executedPlan.executeCollect() + def collectAsList(): java.util.List[Row] = java.util.Arrays.asList(collect() : _*) + override def take(num: Int): Array[Row] = limit(num).collect() // ======================================================================= @@ -482,13 +476,15 @@ class SchemaRDD( (implicit ord: Ordering[Row] = null): SchemaRDD = applySchema(super.coalesce(numPartitions, shuffle)(ord)) - override def distinct(): SchemaRDD = - applySchema(super.distinct()) + override def distinct(): SchemaRDD = applySchema(super.distinct()) override def distinct(numPartitions: Int) (implicit ord: Ordering[Row] = null): SchemaRDD = applySchema(super.distinct(numPartitions)(ord)) + def distinct(numPartitions: Int): SchemaRDD = + applySchema(super.distinct(numPartitions)(null)) + override def filter(f: Row => Boolean): SchemaRDD = applySchema(super.filter(f)) http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSQLContext.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSQLContext.scala b/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSQLContext.scala deleted file mode 100644 index a75f559..0000000 --- a/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSQLContext.scala +++ /dev/null @@ -1,241 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one or more -* contributor license agreements. See the NOTICE file distributed with -* this work for additional information regarding copyright ownership. -* The ASF licenses this file to You under the Apache License, Version 2.0 -* (the "License"); you may not use this file except in compliance with -* the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, software -* distributed under the License is distributed on an "AS IS" BASIS, -* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -* See the License for the specific language governing permissions and -* limitations under the License. -*/ - -package org.apache.spark.sql.api.java - -import java.beans.Introspector - -import org.apache.hadoop.conf.Configuration - -import org.apache.spark.annotation.{DeveloperApi, Experimental} -import org.apache.spark.api.java.{JavaRDD, JavaSparkContext} -import org.apache.spark.sql.SQLContext -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericRow, Row => ScalaRow} -import org.apache.spark.sql.execution.LogicalRDD -import org.apache.spark.sql.json.JsonRDD -import org.apache.spark.sql.parquet.ParquetRelation -import org.apache.spark.sql.sources.{LogicalRelation, BaseRelation} -import org.apache.spark.sql.types._ -import org.apache.spark.util.Utils - -/** - * The entry point for executing Spark SQL queries from a Java program. - */ -class JavaSQLContext(val sqlContext: SQLContext) extends UDFRegistration { - - def this(sparkContext: JavaSparkContext) = this(new SQLContext(sparkContext.sc)) - - def baseRelationToSchemaRDD(baseRelation: BaseRelation): JavaSchemaRDD = { - new JavaSchemaRDD(sqlContext, LogicalRelation(baseRelation)) - } - - /** - * Executes a SQL query using Spark, returning the result as a SchemaRDD. The dialect that is - * used for SQL parsing can be configured with 'spark.sql.dialect'. - * - * @group userf - */ - def sql(sqlText: String): JavaSchemaRDD = { - if (sqlContext.conf.dialect == "sql") { - new JavaSchemaRDD(sqlContext, sqlContext.parseSql(sqlText)) - } else { - sys.error(s"Unsupported SQL dialect: $sqlContext.dialect") - } - } - - /** - * :: Experimental :: - * Creates an empty parquet file with the schema of class `beanClass`, which can be registered as - * a table. This registered table can be used as the target of future `insertInto` operations. - * - * {{{ - * JavaSQLContext sqlCtx = new JavaSQLContext(...) - * - * sqlCtx.createParquetFile(Person.class, "path/to/file.parquet").registerTempTable("people") - * sqlCtx.sql("INSERT INTO people SELECT 'michael', 29") - * }}} - * - * @param beanClass A java bean class object that will be used to determine the schema of the - * parquet file. - * @param path The path where the directory containing parquet metadata should be created. - * Data inserted into this table will also be stored at this location. - * @param allowExisting When false, an exception will be thrown if this directory already exists. - * @param conf A Hadoop configuration object that can be used to specific options to the parquet - * output format. - */ - @Experimental - def createParquetFile( - beanClass: Class[_], - path: String, - allowExisting: Boolean = true, - conf: Configuration = new Configuration()): JavaSchemaRDD = { - new JavaSchemaRDD( - sqlContext, - ParquetRelation.createEmpty(path, getSchema(beanClass), allowExisting, conf, sqlContext)) - } - - /** - * Applies a schema to an RDD of Java Beans. - * - * WARNING: Since there is no guaranteed ordering for fields in a Java Bean, - * SELECT * queries will return the columns in an undefined order. - */ - def applySchema(rdd: JavaRDD[_], beanClass: Class[_]): JavaSchemaRDD = { - val attributeSeq = getSchema(beanClass) - val className = beanClass.getName - val rowRdd = rdd.rdd.mapPartitions { iter => - // BeanInfo is not serializable so we must rediscover it remotely for each partition. - val localBeanInfo = Introspector.getBeanInfo( - Class.forName(className, true, Utils.getContextOrSparkClassLoader)) - val extractors = - localBeanInfo.getPropertyDescriptors.filterNot(_.getName == "class").map(_.getReadMethod) - - iter.map { row => - new GenericRow( - extractors.zip(attributeSeq).map { case (e, attr) => - DataTypeConversions.convertJavaToCatalyst(e.invoke(row), attr.dataType) - }.toArray[Any] - ): ScalaRow - } - } - new JavaSchemaRDD(sqlContext, LogicalRDD(attributeSeq, rowRdd)(sqlContext)) - } - - /** - * :: DeveloperApi :: - * Creates a JavaSchemaRDD from an RDD containing Rows by applying a schema to this RDD. - * It is important to make sure that the structure of every Row of the provided RDD matches the - * provided schema. Otherwise, there will be runtime exception. - */ - @DeveloperApi - def applySchema(rowRDD: JavaRDD[Row], schema: StructType): JavaSchemaRDD = { - val scalaRowRDD = rowRDD.rdd.map(r => r.row) - val logicalPlan = - LogicalRDD(schema.toAttributes, scalaRowRDD)(sqlContext) - new JavaSchemaRDD(sqlContext, logicalPlan) - } - - /** - * Loads a parquet file from regular path or files that match file patterns in path, - * returning the result as a [[JavaSchemaRDD]]. - * Supported glob file pattern information at ([[http://tinyurl.com/kcqrzn8]]). - */ - def parquetFile(path: String): JavaSchemaRDD = - new JavaSchemaRDD( - sqlContext, - ParquetRelation(path, Some(sqlContext.sparkContext.hadoopConfiguration), sqlContext)) - - /** - * Loads a JSON file (one object per line), returning the result as a JavaSchemaRDD. - * It goes through the entire dataset once to determine the schema. - */ - def jsonFile(path: String): JavaSchemaRDD = - jsonRDD(sqlContext.sparkContext.textFile(path)) - - /** - * :: Experimental :: - * Loads a JSON file (one object per line) and applies the given schema, - * returning the result as a JavaSchemaRDD. - */ - @Experimental - def jsonFile(path: String, schema: StructType): JavaSchemaRDD = - jsonRDD(sqlContext.sparkContext.textFile(path), schema) - - /** - * Loads an RDD[String] storing JSON objects (one object per record), returning the result as a - * JavaSchemaRDD. - * It goes through the entire dataset once to determine the schema. - */ - def jsonRDD(json: JavaRDD[String]): JavaSchemaRDD = { - val columnNameOfCorruptJsonRecord = sqlContext.conf.columnNameOfCorruptRecord - val appliedScalaSchema = - JsonRDD.nullTypeToStringType( - JsonRDD.inferSchema(json.rdd, 1.0, columnNameOfCorruptJsonRecord)) - val scalaRowRDD = - JsonRDD.jsonStringToRow(json.rdd, appliedScalaSchema, columnNameOfCorruptJsonRecord) - val logicalPlan = - LogicalRDD(appliedScalaSchema.toAttributes, scalaRowRDD)(sqlContext) - new JavaSchemaRDD(sqlContext, logicalPlan) - } - - /** - * :: Experimental :: - * Loads an RDD[String] storing JSON objects (one object per record) and applies the given schema, - * returning the result as a JavaSchemaRDD. - */ - @Experimental - def jsonRDD(json: JavaRDD[String], schema: StructType): JavaSchemaRDD = { - val columnNameOfCorruptJsonRecord = sqlContext.conf.columnNameOfCorruptRecord - val appliedScalaSchema = - Option(schema).getOrElse( - JsonRDD.nullTypeToStringType( - JsonRDD.inferSchema( - json.rdd, 1.0, columnNameOfCorruptJsonRecord))) - val scalaRowRDD = JsonRDD.jsonStringToRow( - json.rdd, appliedScalaSchema, columnNameOfCorruptJsonRecord) - val logicalPlan = - LogicalRDD(appliedScalaSchema.toAttributes, scalaRowRDD)(sqlContext) - new JavaSchemaRDD(sqlContext, logicalPlan) - } - - /** - * Registers the given RDD as a temporary table in the catalog. Temporary tables exist only - * during the lifetime of this instance of SQLContext. - */ - def registerRDDAsTable(rdd: JavaSchemaRDD, tableName: String): Unit = { - sqlContext.registerRDDAsTable(rdd.baseSchemaRDD, tableName) - } - - /** - * Returns a Catalyst Schema for the given java bean class. - */ - protected def getSchema(beanClass: Class[_]): Seq[AttributeReference] = { - // TODO: All of this could probably be moved to Catalyst as it is mostly not Spark specific. - val beanInfo = Introspector.getBeanInfo(beanClass) - - // Note: The ordering of elements may differ from when the schema is inferred in Scala. - // This is because beanInfo.getPropertyDescriptors gives no guarantees about - // element ordering. - val fields = beanInfo.getPropertyDescriptors.filterNot(_.getName == "class") - fields.map { property => - val (dataType, nullable) = property.getPropertyType match { - case c: Class[_] if c.isAnnotationPresent(classOf[SQLUserDefinedType]) => - (c.getAnnotation(classOf[SQLUserDefinedType]).udt().newInstance(), true) - case c: Class[_] if c == classOf[java.lang.String] => (StringType, true) - case c: Class[_] if c == java.lang.Short.TYPE => (ShortType, false) - case c: Class[_] if c == java.lang.Integer.TYPE => (IntegerType, false) - case c: Class[_] if c == java.lang.Long.TYPE => (LongType, false) - case c: Class[_] if c == java.lang.Double.TYPE => (DoubleType, false) - case c: Class[_] if c == java.lang.Byte.TYPE => (ByteType, false) - case c: Class[_] if c == java.lang.Float.TYPE => (FloatType, false) - case c: Class[_] if c == java.lang.Boolean.TYPE => (BooleanType, false) - - case c: Class[_] if c == classOf[java.lang.Short] => (ShortType, true) - case c: Class[_] if c == classOf[java.lang.Integer] => (IntegerType, true) - case c: Class[_] if c == classOf[java.lang.Long] => (LongType, true) - case c: Class[_] if c == classOf[java.lang.Double] => (DoubleType, true) - case c: Class[_] if c == classOf[java.lang.Byte] => (ByteType, true) - case c: Class[_] if c == classOf[java.lang.Float] => (FloatType, true) - case c: Class[_] if c == classOf[java.lang.Boolean] => (BooleanType, true) - case c: Class[_] if c == classOf[java.math.BigDecimal] => (DecimalType(), true) - case c: Class[_] if c == classOf[java.sql.Date] => (DateType, true) - case c: Class[_] if c == classOf[java.sql.Timestamp] => (TimestampType, true) - } - AttributeReference(property.getName, dataType, nullable)() - } - } -} http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSchemaRDD.scala ---------------------------------------------------------------------- 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 deleted file mode 100644 index 9e10e53..0000000 --- a/sql/core/src/main/scala/org/apache/spark/sql/api/java/JavaSchemaRDD.scala +++ /dev/null @@ -1,225 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.api.java - -import java.util.{List => JList} - -import org.apache.spark.Partitioner -import org.apache.spark.api.java.{JavaRDD, JavaRDDLike} -import org.apache.spark.api.java.function.{Function => JFunction} -import org.apache.spark.rdd.RDD -import org.apache.spark.sql.{SQLContext, SchemaRDD, SchemaRDDLike} -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan -import org.apache.spark.sql.types.StructType -import org.apache.spark.storage.StorageLevel - -/** - * An RDD of [[Row]] objects that is returned as the result of a Spark SQL query. In addition to - * standard RDD operations, a JavaSchemaRDD can also be registered as a table in the JavaSQLContext - * that was used to create. Registering a JavaSchemaRDD allows its contents to be queried in - * future SQL statement. - * - * @groupname schema SchemaRDD Functions - * @groupprio schema -1 - * @groupname Ungrouped Base RDD Functions - */ -class JavaSchemaRDD( - @transient val sqlContext: SQLContext, - @transient val baseLogicalPlan: LogicalPlan) - extends JavaRDDLike[Row, JavaRDD[Row]] - with SchemaRDDLike { - - private[sql] val baseSchemaRDD = new SchemaRDD(sqlContext, logicalPlan) - - /** Returns the underlying Scala SchemaRDD. */ - val schemaRDD: SchemaRDD = baseSchemaRDD - - override val classTag = scala.reflect.classTag[Row] - - override def wrapRDD(rdd: RDD[Row]): JavaRDD[Row] = JavaRDD.fromRDD(rdd) - - val rdd = baseSchemaRDD.map(new Row(_)) - - override def toString: String = baseSchemaRDD.toString - - /** Returns the schema of this JavaSchemaRDD (represented by a StructType). */ - def schema: StructType = baseSchemaRDD.schema.asInstanceOf[StructType] - - // ======================================================================= - // Base RDD functions that do NOT change schema - // ======================================================================= - - // Common RDD functions - - /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */ - def cache(): JavaSchemaRDD = { - baseSchemaRDD.cache() - this - } - - /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */ - def persist(): JavaSchemaRDD = { - baseSchemaRDD.persist() - this - } - - /** - * Set this RDD's storage level to persist its values across operations after the first time - * it is computed. This can only be used to assign a new storage level if the RDD does not - * have a storage level set yet.. - */ - def persist(newLevel: StorageLevel): JavaSchemaRDD = { - baseSchemaRDD.persist(newLevel) - this - } - - /** - * Mark the RDD as non-persistent, and remove all blocks for it from memory and disk. - * - * @param blocking Whether to block until all blocks are deleted. - * @return This RDD. - */ - def unpersist(blocking: Boolean = true): JavaSchemaRDD = { - baseSchemaRDD.unpersist(blocking) - this - } - - /** Assign a name to this RDD */ - def setName(name: String): JavaSchemaRDD = { - baseSchemaRDD.setName(name) - this - } - - // Overridden actions from JavaRDDLike. - - override def collect(): JList[Row] = { - import scala.collection.JavaConversions._ - val arr: java.util.Collection[Row] = baseSchemaRDD.collect().toSeq.map(new Row(_)) - new java.util.ArrayList(arr) - } - - override def count(): Long = baseSchemaRDD.count - - override def take(num: Int): JList[Row] = { - import scala.collection.JavaConversions._ - val arr: java.util.Collection[Row] = baseSchemaRDD.take(num).toSeq.map(new Row(_)) - new java.util.ArrayList(arr) - } - - // Transformations (return a new RDD) - - /** - * Returns a new RDD with each row transformed to a JSON string. - */ - def toJSON(): JavaRDD[String] = - baseSchemaRDD.toJSON.toJavaRDD - - /** - * Return a new RDD that is reduced into `numPartitions` partitions. - */ - def coalesce(numPartitions: Int, shuffle: Boolean = false): JavaSchemaRDD = - baseSchemaRDD.coalesce(numPartitions, shuffle).toJavaSchemaRDD - - /** - * Return a new RDD containing the distinct elements in this RDD. - */ - def distinct(): JavaSchemaRDD = - baseSchemaRDD.distinct().toJavaSchemaRDD - - /** - * Return a new RDD containing the distinct elements in this RDD. - */ - def distinct(numPartitions: Int): JavaSchemaRDD = - baseSchemaRDD.distinct(numPartitions).toJavaSchemaRDD - - /** - * Return a new RDD containing only the elements that satisfy a predicate. - */ - def filter(f: JFunction[Row, java.lang.Boolean]): JavaSchemaRDD = - baseSchemaRDD.filter(x => f.call(new Row(x)).booleanValue()).toJavaSchemaRDD - - /** - * Return the intersection of this RDD and another one. The output will not contain any - * duplicate elements, even if the input RDDs did. - * - * Note that this method performs a shuffle internally. - */ - def intersection(other: JavaSchemaRDD): JavaSchemaRDD = - this.baseSchemaRDD.intersection(other.baseSchemaRDD).toJavaSchemaRDD - - /** - * Return the intersection of this RDD and another one. The output will not contain any - * duplicate elements, even if the input RDDs did. - * - * Note that this method performs a shuffle internally. - * - * @param partitioner Partitioner to use for the resulting RDD - */ - def intersection(other: JavaSchemaRDD, partitioner: Partitioner): JavaSchemaRDD = - this.baseSchemaRDD.intersection(other.baseSchemaRDD, partitioner).toJavaSchemaRDD - - /** - * Return the intersection of this RDD and another one. The output will not contain any - * duplicate elements, even if the input RDDs did. Performs a hash partition across the cluster - * - * Note that this method performs a shuffle internally. - * - * @param numPartitions How many partitions to use in the resulting RDD - */ - def intersection(other: JavaSchemaRDD, numPartitions: Int): JavaSchemaRDD = - this.baseSchemaRDD.intersection(other.baseSchemaRDD, numPartitions).toJavaSchemaRDD - - /** - * Return a new RDD that has exactly `numPartitions` partitions. - * - * Can increase or decrease the level of parallelism in this RDD. Internally, this uses - * a shuffle to redistribute data. - * - * If you are decreasing the number of partitions in this RDD, consider using `coalesce`, - * which can avoid performing a shuffle. - */ - def repartition(numPartitions: Int): JavaSchemaRDD = - baseSchemaRDD.repartition(numPartitions).toJavaSchemaRDD - - /** - * Return an RDD with the elements from `this` that are not in `other`. - * - * Uses `this` partitioner/partition size, because even if `other` is huge, the resulting - * RDD will be <= us. - */ - def subtract(other: JavaSchemaRDD): JavaSchemaRDD = - this.baseSchemaRDD.subtract(other.baseSchemaRDD).toJavaSchemaRDD - - /** - * Return an RDD with the elements from `this` that are not in `other`. - */ - def subtract(other: JavaSchemaRDD, numPartitions: Int): JavaSchemaRDD = - this.baseSchemaRDD.subtract(other.baseSchemaRDD, numPartitions).toJavaSchemaRDD - - /** - * Return an RDD with the elements from `this` that are not in `other`. - */ - def subtract(other: JavaSchemaRDD, p: Partitioner): JavaSchemaRDD = - this.baseSchemaRDD.subtract(other.baseSchemaRDD, p).toJavaSchemaRDD - - /** - * Return a SchemaRDD with a sampled version of the underlying dataset. - */ - def sample(withReplacement: Boolean, fraction: Double, seed: Long): JavaSchemaRDD = - this.baseSchemaRDD.sample(withReplacement, fraction, seed).toJavaSchemaRDD -} http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/main/scala/org/apache/spark/sql/api/java/Row.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/api/java/Row.scala b/sql/core/src/main/scala/org/apache/spark/sql/api/java/Row.scala deleted file mode 100644 index 4faa79a..0000000 --- a/sql/core/src/main/scala/org/apache/spark/sql/api/java/Row.scala +++ /dev/null @@ -1,153 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.api.java - -import scala.annotation.varargs -import scala.collection.convert.Wrappers.{JListWrapper, JMapWrapper} -import scala.collection.JavaConversions -import scala.math.BigDecimal - -import org.apache.spark.api.java.JavaUtils.mapAsSerializableJavaMap -import org.apache.spark.sql.{Row => ScalaRow} - -/** - * A result row from a Spark SQL query. - */ -class Row(private[spark] val row: ScalaRow) extends Serializable { - - /** Returns the number of columns present in this Row. */ - def length: Int = row.length - - /** Returns the value of column `i`. */ - def get(i: Int): Any = - Row.toJavaValue(row(i)) - - /** Returns true if value at column `i` is NULL. */ - def isNullAt(i: Int) = get(i) == null - - /** - * Returns the value of column `i` as an int. This function will throw an exception if the value - * is at `i` is not an integer, or if it is null. - */ - def getInt(i: Int): Int = - row.getInt(i) - - /** - * Returns the value of column `i` as a long. This function will throw an exception if the value - * is at `i` is not a long, or if it is null. - */ - def getLong(i: Int): Long = - row.getLong(i) - - /** - * Returns the value of column `i` as a double. This function will throw an exception if the - * value is at `i` is not a double, or if it is null. - */ - def getDouble(i: Int): Double = - row.getDouble(i) - - /** - * Returns the value of column `i` as a bool. This function will throw an exception if the value - * is at `i` is not a boolean, or if it is null. - */ - def getBoolean(i: Int): Boolean = - row.getBoolean(i) - - /** - * Returns the value of column `i` as a short. This function will throw an exception if the value - * is at `i` is not a short, or if it is null. - */ - def getShort(i: Int): Short = - row.getShort(i) - - /** - * Returns the value of column `i` as a byte. This function will throw an exception if the value - * is at `i` is not a byte, or if it is null. - */ - def getByte(i: Int): Byte = - row.getByte(i) - - /** - * Returns the value of column `i` as a float. This function will throw an exception if the value - * is at `i` is not a float, or if it is null. - */ - def getFloat(i: Int): Float = - row.getFloat(i) - - /** - * Returns the value of column `i` as a String. This function will throw an exception if the - * value is at `i` is not a String. - */ - def getString(i: Int): String = - row.getString(i) - - def canEqual(other: Any): Boolean = other.isInstanceOf[Row] - - override def equals(other: Any): Boolean = other match { - case that: Row => - (that canEqual this) && - row == that.row - case _ => false - } - - override def hashCode(): Int = row.hashCode() - - override def toString: String = row.toString -} - -object Row { - - private def toJavaValue(value: Any): Any = value match { - // For values of this ScalaRow, we will do the conversion when - // they are actually accessed. - case row: ScalaRow => new Row(row) - case map: scala.collection.Map[_, _] => - mapAsSerializableJavaMap( - map.map { - case (key, value) => (toJavaValue(key), toJavaValue(value)) - } - ) - case seq: scala.collection.Seq[_] => - JavaConversions.seqAsJavaList(seq.map(toJavaValue)) - case decimal: BigDecimal => decimal.underlying() - case other => other - } - - // TODO: Consolidate the toScalaValue at here with the scalafy in JsonRDD? - private def toScalaValue(value: Any): Any = value match { - // Values of this row have been converted to Scala values. - case row: Row => row.row - case map: java.util.Map[_, _] => - JMapWrapper(map).map { - case (key, value) => (toScalaValue(key), toScalaValue(value)) - } - case list: java.util.List[_] => - JListWrapper(list).map(toScalaValue) - case decimal: java.math.BigDecimal => BigDecimal(decimal) - case other => other - } - - /** - * Creates a Row with the given values. - */ - @varargs def create(values: Any*): Row = { - // Right now, we cannot use @varargs to annotate the constructor of - // org.apache.spark.sql.api.java.Row. See https://issues.scala-lang.org/browse/SI-8383. - new Row(ScalaRow(values.map(toScalaValue):_*)) - } -} http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/main/scala/org/apache/spark/sql/api/java/UDFRegistration.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/api/java/UDFRegistration.scala b/sql/core/src/main/scala/org/apache/spark/sql/api/java/UDFRegistration.scala deleted file mode 100644 index 4186c27..0000000 --- a/sql/core/src/main/scala/org/apache/spark/sql/api/java/UDFRegistration.scala +++ /dev/null @@ -1,251 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one or more -* contributor license agreements. See the NOTICE file distributed with -* this work for additional information regarding copyright ownership. -* The ASF licenses this file to You under the Apache License, Version 2.0 -* (the "License"); you may not use this file except in compliance with -* the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, software -* distributed under the License is distributed on an "AS IS" BASIS, -* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -* See the License for the specific language governing permissions and -* limitations under the License. -*/ - -package org.apache.spark.sql.api.java - -import org.apache.spark.sql.catalyst.expressions.{Expression, ScalaUdf} -import org.apache.spark.sql.types.DataType - -/** - * A collection of functions that allow Java users to register UDFs. In order to handle functions - * of varying airities with minimal boilerplate for our users, we generate classes and functions - * for each airity up to 22. The code for this generation can be found in comments in this trait. - */ -private[java] trait UDFRegistration { - self: JavaSQLContext => - - /* The following functions and required interfaces are generated with these code fragments: - - (1 to 22).foreach { i => - val extTypeArgs = (1 to i).map(_ => "_").mkString(", ") - val anyTypeArgs = (1 to i).map(_ => "Any").mkString(", ") - val anyCast = s".asInstanceOf[UDF$i[$anyTypeArgs, Any]]" - val anyParams = (1 to i).map(_ => "_: Any").mkString(", ") - println(s""" - |def registerFunction( - | name: String, f: UDF$i[$extTypeArgs, _], @transient dataType: DataType) = { - | sqlContext.functionRegistry.registerFunction( - | name, - | (e: Seq[Expression]) => ScalaUdf(f$anyCast.call($anyParams), dataType, e)) - |} - """.stripMargin) - } - - import java.io.File - import org.apache.spark.sql.catalyst.util.stringToFile - val directory = new File("sql/core/src/main/java/org/apache/spark/sql/api/java/") - (1 to 22).foreach { i => - val typeArgs = (1 to i).map(i => s"T$i").mkString(", ") - val args = (1 to i).map(i => s"T$i t$i").mkString(", ") - - val contents = - s"""/* - | * Licensed to the Apache Software Foundation (ASF) under one or more - | * contributor license agreements. See the NOTICE file distributed with - | * this work for additional information regarding copyright ownership. - | * The ASF licenses this file to You under the Apache License, Version 2.0 - | * (the "License"); you may not use this file except in compliance with - | * the License. You may obtain a copy of the License at - | * - | * http://www.apache.org/licenses/LICENSE-2.0 - | * - | * Unless required by applicable law or agreed to in writing, software - | * distributed under the License is distributed on an "AS IS" BASIS, - | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - | * See the License for the specific language governing permissions and - | * limitations under the License. - | */ - | - |package org.apache.spark.sql.api.java; - | - |import java.io.Serializable; - | - |// ************************************************** - |// THIS FILE IS AUTOGENERATED BY CODE IN - |// org.apache.spark.sql.api.java.FunctionRegistration - |// ************************************************** - | - |/** - | * A Spark SQL UDF that has $i arguments. - | */ - |public interface UDF$i<$typeArgs, R> extends Serializable { - | public R call($args) throws Exception; - |} - |""".stripMargin - - stringToFile(new File(directory, s"UDF$i.java"), contents) - } - - */ - - // scalastyle:off - def registerFunction( - name: String, f: UDF1[_, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF1[Any, Any]].call(_: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF2[_, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF2[Any, Any, Any]].call(_: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF3[_, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF3[Any, Any, Any, Any]].call(_: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF4[_, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF4[Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF5[_, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF5[Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF6[_, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF6[Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF7[_, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF7[Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF8[_, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF8[Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF9[_, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF9[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF10[_, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF10[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - - def registerFunction( - name: String, f: UDF11[_, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF11[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF12[_, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF12[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF13[_, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF13[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF14[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF14[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF15[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF15[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF16[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF16[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF17[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF17[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF18[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF18[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF19[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF19[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF20[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF20[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF21[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF21[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - - def registerFunction( - name: String, f: UDF22[_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _], @transient dataType: DataType) = { - sqlContext.functionRegistry.registerFunction( - name, - (e: Seq[Expression]) => ScalaUdf(f.asInstanceOf[UDF22[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any]].call(_: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any, _: Any), dataType, e)) - } - // scalastyle:on -} http://git-wip-us.apache.org/repos/asf/spark/blob/61b427d4/sql/core/src/test/java/org/apache/spark/sql/api/java/JavaAPISuite.java ---------------------------------------------------------------------- diff --git a/sql/core/src/test/java/org/apache/spark/sql/api/java/JavaAPISuite.java b/sql/core/src/test/java/org/apache/spark/sql/api/java/JavaAPISuite.java index 88017eb..9ff4047 100644 --- a/sql/core/src/test/java/org/apache/spark/sql/api/java/JavaAPISuite.java +++ b/sql/core/src/test/java/org/apache/spark/sql/api/java/JavaAPISuite.java @@ -24,6 +24,8 @@ import org.junit.Before; import org.junit.Test; import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SQLContext; import org.apache.spark.sql.types.DataTypes; // The test suite itself is Serializable so that anonymous Function implementations can be @@ -31,12 +33,12 @@ import org.apache.spark.sql.types.DataTypes; // see http://stackoverflow.com/questions/758570/. public class JavaAPISuite implements Serializable { private transient JavaSparkContext sc; - private transient JavaSQLContext sqlContext; + private transient SQLContext sqlContext; @Before public void setUp() { sc = new JavaSparkContext("local", "JavaAPISuite"); - sqlContext = new JavaSQLContext(sc); + sqlContext = new SQLContext(sc); } @After @@ -52,15 +54,14 @@ public class JavaAPISuite implements Serializable { // sqlContext.registerFunction( // "stringLengthTest", (String str) -> str.length(), DataType.IntegerType); - sqlContext.registerFunction("stringLengthTest", new UDF1<String, Integer>() { + sqlContext.udf().register("stringLengthTest", new UDF1<String, Integer>() { @Override public Integer call(String str) throws Exception { return str.length(); } }, DataTypes.IntegerType); - // TODO: Why do we need this cast? - Row result = (Row) sqlContext.sql("SELECT stringLengthTest('test')").first(); + Row result = sqlContext.sql("SELECT stringLengthTest('test')").first(); assert(result.getInt(0) == 4); } @@ -73,15 +74,14 @@ public class JavaAPISuite implements Serializable { // (String str1, String str2) -> str1.length() + str2.length, // DataType.IntegerType); - sqlContext.registerFunction("stringLengthTest", new UDF2<String, String, Integer>() { + sqlContext.udf().register("stringLengthTest", new UDF2<String, String, Integer>() { @Override public Integer call(String str1, String str2) throws Exception { return str1.length() + str2.length(); } }, DataTypes.IntegerType); - // TODO: Why do we need this cast? - Row result = (Row) sqlContext.sql("SELECT stringLengthTest('test', 'test2')").first(); + Row result = sqlContext.sql("SELECT stringLengthTest('test', 'test2')").first(); assert(result.getInt(0) == 9); } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org