You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pw...@apache.org on 2014/04/15 09:08:06 UTC
git commit: SPARK-1374: PySpark API for SparkSQL
Repository: spark
Updated Branches:
refs/heads/master 0247b5c54 -> c99bcb7fe
SPARK-1374: PySpark API for SparkSQL
An initial API that exposes SparkSQL functionality in PySpark. A PythonRDD composed of dictionaries, with string keys and primitive values (boolean, float, int, long, string) can be converted into a SchemaRDD that supports sql queries.
```
from pyspark.context import SQLContext
sqlCtx = SQLContext(sc)
rdd = sc.parallelize([{"field1" : 1, "field2" : "row1"}, {"field1" : 2, "field2": "row2"}, {"field1" : 3, "field2": "row3"}])
srdd = sqlCtx.applySchema(rdd)
sqlCtx.registerRDDAsTable(srdd, "table1")
srdd2 = sqlCtx.sql("SELECT field1 AS f1, field2 as f2 from table1")
srdd2.collect()
```
The last line yields ```[{"f1" : 1, "f2" : "row1"}, {"f1" : 2, "f2": "row2"}, {"f1" : 3, "f2": "row3"}]```
Author: Ahir Reddy <ah...@gmail.com>
Author: Michael Armbrust <mi...@databricks.com>
Closes #363 from ahirreddy/pysql and squashes the following commits:
0294497 [Ahir Reddy] Updated log4j properties to supress Hive Warns
307d6e0 [Ahir Reddy] Style fix
6f7b8f6 [Ahir Reddy] Temporary fix MIMA checker. Since we now assemble Spark jar with Hive, we don't want to check the interfaces of all of our hive dependencies
3ef074a [Ahir Reddy] Updated documentation because classes moved to sql.py
29245bf [Ahir Reddy] Cache underlying SchemaRDD instead of generating and caching PythonRDD
f2312c7 [Ahir Reddy] Moved everything into sql.py
a19afe4 [Ahir Reddy] Doc fixes
6d658ba [Ahir Reddy] Remove the metastore directory created by the HiveContext tests in SparkSQL
521ff6d [Ahir Reddy] Trying to get spark to build with hive
ab95eba [Ahir Reddy] Set SPARK_HIVE=true on jenkins
ded03e7 [Ahir Reddy] Added doc test for HiveContext
22de1d4 [Ahir Reddy] Fixed maven pyrolite dependency
e4da06c [Ahir Reddy] Display message if hive is not built into spark
227a0be [Michael Armbrust] Update API links. Fix Hive example.
58e2aa9 [Michael Armbrust] Build Docs for pyspark SQL Api. Minor fixes.
4285340 [Michael Armbrust] Fix building of Hive API Docs.
38a92b0 [Michael Armbrust] Add note to future non-python developers about python docs.
337b201 [Ahir Reddy] Changed com.clearspring.analytics stream version from 2.4.0 to 2.5.1 to match SBT build, and added pyrolite to maven build
40491c9 [Ahir Reddy] PR Changes + Method Visibility
1836944 [Michael Armbrust] Fix comments.
e00980f [Michael Armbrust] First draft of python sql programming guide.
b0192d3 [Ahir Reddy] Added Long, Double and Boolean as usable types + unit test
f98a422 [Ahir Reddy] HiveContexts
79621cf [Ahir Reddy] cleaning up cruft
b406ba0 [Ahir Reddy] doctest formatting
20936a5 [Ahir Reddy] Added tests and documentation
e4d21b4 [Ahir Reddy] Added pyrolite dependency
79f739d [Ahir Reddy] added more tests
7515ba0 [Ahir Reddy] added more tests :)
d26ec5e [Ahir Reddy] added test
e9f5b8d [Ahir Reddy] adding tests
906d180 [Ahir Reddy] added todo explaining cost of creating Row object in python
251f99d [Ahir Reddy] for now only allow dictionaries as input
09b9980 [Ahir Reddy] made jrdd explicitly lazy
c608947 [Ahir Reddy] SchemaRDD now has all RDD operations
725c91e [Ahir Reddy] awesome row objects
55d1c76 [Ahir Reddy] return row objects
4fe1319 [Ahir Reddy] output dictionaries correctly
be079de [Ahir Reddy] returning dictionaries works
cd5f79f [Ahir Reddy] Switched to using Scala SQLContext
e948bd9 [Ahir Reddy] yippie
4886052 [Ahir Reddy] even better
c0fb1c6 [Ahir Reddy] more working
043ca85 [Ahir Reddy] working
5496f9f [Ahir Reddy] doesn't crash
b8b904b [Ahir Reddy] Added schema rdd class
67ba875 [Ahir Reddy] java to python, and python to java
bcc0f23 [Ahir Reddy] Java to python
ab6025d [Ahir Reddy] compiling
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/c99bcb7f
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/c99bcb7f
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/c99bcb7f
Branch: refs/heads/master
Commit: c99bcb7feaa761c5826f2e1d844d0502a3b79538
Parents: 0247b5c
Author: Ahir Reddy <ah...@gmail.com>
Authored: Tue Apr 15 00:07:55 2014 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Tue Apr 15 00:07:55 2014 -0700
----------------------------------------------------------------------
core/pom.xml | 5 +
.../org/apache/spark/api/python/PythonRDD.scala | 32 ++
dev/run-tests | 1 +
docs/README.md | 2 +-
docs/_plugins/copy_api_dirs.rb | 4 +-
docs/sql-programming-guide.md | 103 +++++-
pom.xml | 2 +-
project/SparkBuild.scala | 3 +-
python/pyspark/__init__.py | 18 +-
python/pyspark/java_gateway.py | 4 +
python/pyspark/sql.py | 363 +++++++++++++++++++
python/run-tests | 4 +
.../scala/org/apache/spark/sql/SQLContext.scala | 27 ++
.../scala/org/apache/spark/sql/SchemaRDD.scala | 23 ++
.../org/apache/spark/sql/hive/TestHive.scala | 3 +-
sql/hive/src/test/resources/log4j.properties | 3 +
.../apache/spark/tools/GenerateMIMAIgnore.scala | 4 +-
17 files changed, 589 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index a1bdd8e..d87e2bc 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -266,6 +266,11 @@
<artifactId>junit-interface</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.spark-project</groupId>
+ <artifactId>pyrolite</artifactId>
+ <version>2.0</version>
+ </dependency>
</dependencies>
<build>
<outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
----------------------------------------------------------------------
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 32f1100..f9d86fe 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,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.{Pickler, Unpickler}
+
import org.apache.spark._
import org.apache.spark.api.java.{JavaSparkContext, JavaPairRDD, JavaRDD}
import org.apache.spark.broadcast.Broadcast
@@ -284,6 +286,36 @@ private[spark] object PythonRDD {
file.close()
}
+ /**
+ * Convert an RDD of serialized Python dictionaries to Scala Maps
+ * TODO: Support more Python types.
+ */
+ def pythonToJavaMap(pyRDD: JavaRDD[Array[Byte]]): JavaRDD[Map[String, _]] = {
+ pyRDD.rdd.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ // TODO: Figure out why flatMap is necessay for pyspark
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ case objs: java.util.ArrayList[JMap[String, _] @unchecked] => objs.map(_.toMap)
+ // Incase the partition doesn't have a collection
+ case obj: JMap[String @unchecked, _] => Seq(obj.toMap)
+ }
+ }
+ }
+ }
+
+ /**
+ * Convert and RDD of Java objects to and RDD of serialized Python objects, that is usable by
+ * PySpark.
+ */
+ def javaToPython(jRDD: JavaRDD[Any]): JavaRDD[Array[Byte]] = {
+ jRDD.rdd.mapPartitions { iter =>
+ val pickle = new Pickler
+ iter.map { row =>
+ pickle.dumps(row)
+ }
+ }
+ }
}
private
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/dev/run-tests
----------------------------------------------------------------------
diff --git a/dev/run-tests b/dev/run-tests
index 6ad674a..0725b68 100755
--- a/dev/run-tests
+++ b/dev/run-tests
@@ -34,6 +34,7 @@ else
fi
JAVA_VERSION=$($java_cmd -version 2>&1 | sed 's/java version "\(.*\)\.\(.*\)\..*"/\1\2/; 1q')
[ "$JAVA_VERSION" -ge 18 ] && echo "" || echo "[Warn] Java 8 tests will not run because JDK version is < 1.8."
+export SPARK_HIVE=true
echo "========================================================================="
echo "Running Apache RAT checks"
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/docs/README.md
----------------------------------------------------------------------
diff --git a/docs/README.md b/docs/README.md
index 0678fc5..75b1811 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -42,7 +42,7 @@ To mark a block of code in your markdown to be syntax highlighted by jekyll duri
You can build just the Spark scaladoc by running `sbt/sbt doc` from the SPARK_PROJECT_ROOT directory.
-Similarly, you can build just the PySpark epydoc by running `epydoc --config epydoc.conf` from the SPARK_PROJECT_ROOT/pyspark directory.
+Similarly, you can build just the PySpark epydoc by running `epydoc --config epydoc.conf` from the SPARK_PROJECT_ROOT/pyspark directory. Documentation is only generated for classes that are listed as public in `__init__.py`.
When you run `jekyll` in the docs directory, it will also copy over the scaladoc for the various Spark subprojects into the docs directory (and then also into the _site directory). We use a jekyll plugin to run `sbt/sbt doc` before building the site so if you haven't run it (recently) it may take some time as it generates all of the scaladoc. The jekyll plugin also generates the PySpark docs using [epydoc](http://epydoc.sourceforge.net/).
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/docs/_plugins/copy_api_dirs.rb
----------------------------------------------------------------------
diff --git a/docs/_plugins/copy_api_dirs.rb b/docs/_plugins/copy_api_dirs.rb
index bbd56d2..05f0bd4 100644
--- a/docs/_plugins/copy_api_dirs.rb
+++ b/docs/_plugins/copy_api_dirs.rb
@@ -32,8 +32,8 @@ if not (ENV['SKIP_API'] == '1' or ENV['SKIP_SCALADOC'] == '1')
curr_dir = pwd
cd("..")
- puts "Running sbt/sbt doc from " + pwd + "; this may take a few minutes..."
- puts `sbt/sbt doc`
+ puts "Running 'sbt/sbt doc hive/doc' from " + pwd + "; this may take a few minutes..."
+ puts `sbt/sbt doc hive/doc`
puts "Moving back into docs dir."
cd("docs")
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/docs/sql-programming-guide.md
----------------------------------------------------------------------
diff --git a/docs/sql-programming-guide.md b/docs/sql-programming-guide.md
index a59393e..6f616fb 100644
--- a/docs/sql-programming-guide.md
+++ b/docs/sql-programming-guide.md
@@ -20,7 +20,7 @@ a schema that describes the data types of each column in the row. A SchemaRDD i
in a traditional relational database. A SchemaRDD can be created from an existing RDD, parquet
file, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
-**All of the examples on this page use sample data included in the Spark distribution and can be run in the spark-shell.**
+**All of the examples on this page use sample data included in the Spark distribution and can be run in the `spark-shell`.**
</div>
@@ -33,6 +33,19 @@ a schema that describes the data types of each column in the row. A JavaSchemaR
in a traditional relational database. A JavaSchemaRDD can be created from an existing RDD, parquet
file, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
</div>
+
+<div data-lang="python" markdown="1">
+
+Spark SQL allows relational queries expressed in SQL or HiveQL to be executed using
+Spark. At the core of this component is a new type of RDD,
+[SchemaRDD](api/pyspark/pyspark.sql.SchemaRDD-class.html). SchemaRDDs are composed
+[Row](api/pyspark/pyspark.sql.Row-class.html) objects along with
+a schema that describes the data types of each column in the row. A SchemaRDD is similar to a table
+in a traditional relational database. A SchemaRDD can be created from an existing RDD, parquet
+file, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
+
+**All of the examples on this page use sample data included in the Spark distribution and can be run in the `pyspark` shell.**
+</div>
</div>
***************************************************************************************************
@@ -44,7 +57,7 @@ file, or by running HiveQL against data stored in [Apache Hive](http://hive.apac
The entry point into all relational functionality in Spark is the
[SQLContext](api/sql/core/index.html#org.apache.spark.sql.SQLContext) class, or one of its
-decendents. To create a basic SQLContext, all you need is a SparkContext.
+descendants. To create a basic SQLContext, all you need is a SparkContext.
{% highlight scala %}
val sc: SparkContext // An existing SparkContext.
@@ -60,7 +73,7 @@ import sqlContext._
The entry point into all relational functionality in Spark is the
[JavaSQLContext](api/sql/core/index.html#org.apache.spark.sql.api.java.JavaSQLContext) class, or one
-of its decendents. To create a basic JavaSQLContext, all you need is a JavaSparkContext.
+of its descendants. To create a basic JavaSQLContext, all you need is a JavaSparkContext.
{% highlight java %}
JavaSparkContext ctx = ...; // An existing JavaSparkContext.
@@ -69,6 +82,19 @@ JavaSQLContext sqlCtx = new org.apache.spark.sql.api.java.JavaSQLContext(ctx);
</div>
+<div data-lang="python" markdown="1">
+
+The entry point into all relational functionality in Spark is the
+[SQLContext](api/pyspark/pyspark.sql.SQLContext-class.html) class, or one
+of its decedents. To create a basic SQLContext, all you need is a SparkContext.
+
+{% highlight python %}
+from pyspark.sql import SQLContext
+sqlCtx = SQLContext(sc)
+{% endhighlight %}
+
+</div>
+
</div>
## Running SQL on RDDs
@@ -81,7 +107,7 @@ One type of table that is supported by Spark SQL is an RDD of Scala case classes
defines the schema of the table. The names of the arguments to the case class are read using
reflection and become the names of the columns. Case classes can also be nested or contain complex
types such as Sequences or Arrays. This RDD can be implicitly converted to a SchemaRDD and then be
-registered as a table. Tables can used in subsequent SQL statements.
+registered as a table. Tables can be used in subsequent SQL statements.
{% highlight scala %}
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
@@ -176,6 +202,34 @@ List<String> teenagerNames = teenagers.map(new Function<Row, String>() {
</div>
+<div data-lang="python" markdown="1">
+
+One type of table that is supported by Spark SQL is an RDD of dictionaries. The keys of the
+dictionary define the columns names of the table, and the types are inferred by looking at the first
+row. Any RDD of dictionaries can converted to a SchemaRDD and then registered as a table. Tables
+can be used in subsequent SQL statements.
+
+{% highlight python %}
+# Load a text file and convert each line to a dictionary.
+lines = sc.textFile("examples/src/main/resources/people.txt")
+parts = lines.map(lambda l: l.split(","))
+people = parts.map(lambda p: {"name": p[0], "age": int(p[1])})
+
+# Infer the schema, and register the SchemaRDD as a table.
+# In future versions of PySpark we would like to add support for registering RDDs with other
+# datatypes as tables
+peopleTable = sqlCtx.inferSchema(people)
+peopleTable.registerAsTable("people")
+
+# SQL can be run over SchemaRDDs that have been registered as a table.
+teenagers = sqlCtx.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
+
+# The results of SQL queries are RDDs and support all the normal RDD operations.
+teenNames = teenagers.map(lambda p: "Name: " + p.name)
+{% endhighlight %}
+
+</div>
+
</div>
**Note that Spark SQL currently uses a very basic SQL parser.**
@@ -235,6 +289,27 @@ JavaSchemaRDD teenagers = sqlCtx.sql("SELECT name FROM parquetFile WHERE age >=
</div>
+<div data-lang="python" markdown="1">
+
+{% highlight python %}
+
+peopleTable # The SchemaRDD from the previous example.
+
+# SchemaRDDs can be saved as parquet files, maintaining the schema information.
+peopleTable.saveAsParquetFile("people.parquet")
+
+# 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 SchemaRDD.
+parquetFile = sqlCtx.parquetFile("people.parquet")
+
+# Parquet files can also be registered as tables and then used in SQL statements.
+parquetFile.registerAsTable("parquetFile");
+teenagers = sqlCtx.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
+
+{% endhighlight %}
+
+</div>
+
</div>
## Writing Language-Integrated Relational Queries
@@ -318,4 +393,24 @@ Row[] results = hiveCtx.hql("FROM src SELECT key, value").collect();
</div>
+<div data-lang="python" markdown="1">
+
+When working with Hive one must construct a `HiveContext`, which inherits from `SQLContext`, and
+adds support for finding tables in in the MetaStore and writing queries using HiveQL. In addition to
+the `sql` method a `HiveContext` also provides an `hql` methods, which allows queries to be
+expressed in HiveQL.
+
+{% highlight python %}
+
+from pyspark.sql import HiveContext
+hiveCtx = HiveContext(sc)
+
+hiveCtx.hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
+hiveCtx.hql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")
+
+# Queries can be expressed in HiveQL.
+results = hiveCtx.hql("FROM src SELECT key, value").collect()
+
+{% endhighlight %}
+
</div>
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0eacedf..cd20437 100644
--- a/pom.xml
+++ b/pom.xml
@@ -262,7 +262,7 @@
<dependency>
<groupId>com.clearspring.analytics</groupId>
<artifactId>stream</artifactId>
- <version>2.4.0</version>
+ <version>2.5.1</version>
</dependency>
<!-- In theory we need not directly depend on protobuf since Spark does not directly
use it. However, when building with Hadoop/YARN 2.2 Maven doesn't correctly bump
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/project/SparkBuild.scala
----------------------------------------------------------------------
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
index aac07b9..09b527c 100644
--- a/project/SparkBuild.scala
+++ b/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",
+ "org.spark-project" % "pyrolite" % "2.0"
),
libraryDependencies ++= maybeAvro
)
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/python/pyspark/__init__.py
----------------------------------------------------------------------
diff --git a/python/pyspark/__init__.py b/python/pyspark/__init__.py
index a51d5af..73fe737 100644
--- a/python/pyspark/__init__.py
+++ b/python/pyspark/__init__.py
@@ -34,6 +34,19 @@ Public classes:
Access files shipped with jobs.
- L{StorageLevel<pyspark.storagelevel.StorageLevel>}
Finer-grained cache persistence levels.
+
+Spark SQL:
+ - L{SQLContext<pyspark.sql.SQLContext>}
+ Main entry point for SQL functionality.
+ - L{SchemaRDD<pyspark.sql.SchemaRDD>}
+ A Resilient Distributed Dataset (RDD) with Schema information for the data contained. In
+ addition to normal RDD operations, SchemaRDDs also support SQL.
+ - L{Row<pyspark.sql.Row>}
+ A Row of data returned by a Spark SQL query.
+
+Hive:
+ - L{HiveContext<pyspark.context.HiveContext>}
+ Main entry point for accessing data stored in Apache Hive..
"""
@@ -45,9 +58,12 @@ sys.path.insert(0, os.path.join(os.environ["SPARK_HOME"], "python/lib/py4j-0.8.1
from pyspark.conf import SparkConf
from pyspark.context import SparkContext
+from pyspark.sql import SQLContext
from pyspark.rdd import RDD
+from pyspark.sql import SchemaRDD
+from pyspark.sql import Row
from pyspark.files import SparkFiles
from pyspark.storagelevel import StorageLevel
-__all__ = ["SparkConf", "SparkContext", "RDD", "SparkFiles", "StorageLevel"]
+__all__ = ["SparkConf", "SparkContext", "SQLContext", "RDD", "SchemaRDD", "SparkFiles", "StorageLevel", "Row"]
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/python/pyspark/java_gateway.py
----------------------------------------------------------------------
diff --git a/python/pyspark/java_gateway.py b/python/pyspark/java_gateway.py
index 6a16756..6bb6c87 100644
--- a/python/pyspark/java_gateway.py
+++ b/python/pyspark/java_gateway.py
@@ -64,5 +64,9 @@ def launch_gateway():
java_import(gateway.jvm, "org.apache.spark.api.java.*")
java_import(gateway.jvm, "org.apache.spark.api.python.*")
java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")
+ java_import(gateway.jvm, "org.apache.spark.sql.SQLContext")
+ java_import(gateway.jvm, "org.apache.spark.sql.hive.HiveContext")
+ java_import(gateway.jvm, "org.apache.spark.sql.hive.LocalHiveContext")
+ java_import(gateway.jvm, "org.apache.spark.sql.hive.TestHiveContext")
java_import(gateway.jvm, "scala.Tuple2")
return gateway
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/python/pyspark/sql.py
----------------------------------------------------------------------
diff --git a/python/pyspark/sql.py b/python/pyspark/sql.py
new file mode 100644
index 0000000..67e6eee
--- /dev/null
+++ b/python/pyspark/sql.py
@@ -0,0 +1,363 @@
+#
+# 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.
+#
+
+from pyspark.rdd import RDD
+
+from py4j.protocol import Py4JError
+
+__all__ = ["SQLContext", "HiveContext", "LocalHiveContext", "TestHiveContext", "SchemaRDD", "Row"]
+
+
+class SQLContext:
+ """
+ Main entry point for SparkSQL functionality. A SQLContext can be used create L{SchemaRDD}s,
+ register L{SchemaRDD}s as tables, execute sql over tables, cache tables, and read parquet files.
+ """
+
+ def __init__(self, sparkContext):
+ """
+ Create a new SQLContext.
+
+ @param sparkContext: The SparkContext to wrap.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> sqlCtx.inferSchema(srdd) # doctest: +IGNORE_EXCEPTION_DETAIL
+ Traceback (most recent call last):
+ ...
+ ValueError:...
+
+ >>> bad_rdd = sc.parallelize([1,2,3])
+ >>> sqlCtx.inferSchema(bad_rdd) # doctest: +IGNORE_EXCEPTION_DETAIL
+ Traceback (most recent call last):
+ ...
+ ValueError:...
+
+ >>> allTypes = sc.parallelize([{"int" : 1, "string" : "string", "double" : 1.0, "long": 1L,
+ ... "boolean" : True}])
+ >>> srdd = sqlCtx.inferSchema(allTypes).map(lambda x: (x.int, x.string, x.double, x.long,
+ ... x.boolean))
+ >>> srdd.collect()[0]
+ (1, u'string', 1.0, 1, True)
+ """
+ self._sc = sparkContext
+ self._jsc = self._sc._jsc
+ self._jvm = self._sc._jvm
+ self._pythonToJavaMap = self._jvm.PythonRDD.pythonToJavaMap
+
+ @property
+ def _ssql_ctx(self):
+ """
+ Accessor for the JVM SparkSQL context. Subclasses can overrite this property to provide
+ their own JVM Contexts.
+ """
+ if not hasattr(self, '_scala_SQLContext'):
+ self._scala_SQLContext = self._jvm.SQLContext(self._jsc.sc())
+ return self._scala_SQLContext
+
+ def inferSchema(self, rdd):
+ """
+ Infer and apply a schema to an RDD of L{dict}s. 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.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> srdd.collect() == [{"field1" : 1, "field2" : "row1"}, {"field1" : 2, "field2": "row2"},
+ ... {"field1" : 3, "field2": "row3"}]
+ True
+ """
+ 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)
+
+ def registerRDDAsTable(self, rdd, tableName):
+ """
+ Registers the given RDD as a temporary table in the catalog. Temporary tables exist only
+ during the lifetime of this instance of SQLContext.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> sqlCtx.registerRDDAsTable(srdd, "table1")
+ """
+ if (rdd.__class__ is SchemaRDD):
+ jschema_rdd = rdd._jschema_rdd
+ self._ssql_ctx.registerRDDAsTable(jschema_rdd, tableName)
+ else:
+ raise ValueError("Can only register SchemaRDD as table")
+
+ def parquetFile(self, path):
+ """
+ Loads a Parquet file, returning the result as a L{SchemaRDD}.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> srdd.saveAsParquetFile("/tmp/tmp.parquet")
+ >>> srdd2 = sqlCtx.parquetFile("/tmp/tmp.parquet")
+ >>> srdd.collect() == srdd2.collect()
+ True
+ """
+ jschema_rdd = self._ssql_ctx.parquetFile(path)
+ return SchemaRDD(jschema_rdd, self)
+
+ def sql(self, sqlQuery):
+ """
+ Executes a SQL query using Spark, returning the result as a L{SchemaRDD}.
+
+ >>> 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
+ """
+ return SchemaRDD(self._ssql_ctx.sql(sqlQuery), self)
+
+ def table(self, tableName):
+ """
+ Returns the specified table as a L{SchemaRDD}.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> sqlCtx.registerRDDAsTable(srdd, "table1")
+ >>> srdd2 = sqlCtx.table("table1")
+ >>> srdd.collect() == srdd2.collect()
+ True
+ """
+ return SchemaRDD(self._ssql_ctx.table(tableName), self)
+
+ def cacheTable(tableName):
+ """
+ Caches the specified table in-memory.
+ """
+ self._ssql_ctx.cacheTable(tableName)
+
+ def uncacheTable(tableName):
+ """
+ Removes the specified table from the in-memory cache.
+ """
+ self._ssql_ctx.uncacheTable(tableName)
+
+
+class HiveContext(SQLContext):
+ """
+ An instance of the Spark SQL execution engine that integrates with data stored in Hive.
+ Configuration for Hive is read from hive-site.xml on the classpath. It supports running both SQL
+ and HiveQL commands.
+ """
+
+ @property
+ def _ssql_ctx(self):
+ try:
+ if not hasattr(self, '_scala_HiveContext'):
+ 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 " \
+ "sbt/sbt assembly" , e)
+
+ def _get_hive_ctx(self):
+ return self._jvm.HiveContext(self._jsc.sc())
+
+ def hiveql(self, hqlQuery):
+ """
+ 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}.
+ """
+ return self.hiveql(hqlQuery)
+
+
+class LocalHiveContext(HiveContext):
+ """
+ Starts up an instance of hive where metadata is stored locally. An in-process metadata data is
+ created with data stored in ./metadata. Warehouse data is stored in in ./warehouse.
+
+ >>> import os
+ >>> hiveCtx = LocalHiveContext(sc)
+ >>> try:
+ ... 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]))
+ >>> num = results.count()
+ >>> reduce_sum = results.reduce(lambda x, y: x + y)
+ >>> num
+ 500
+ >>> reduce_sum
+ 130091
+ """
+
+ def _get_hive_ctx(self):
+ return self._jvm.LocalHiveContext(self._jsc.sc())
+
+
+class TestHiveContext(HiveContext):
+
+ def _get_hive_ctx(self):
+ 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):
+ """
+ An extended L{dict} that takes a L{dict} in its constructor, and exposes those items as fields.
+
+ >>> r = Row({"hello" : "world", "foo" : "bar"})
+ >>> r.hello
+ 'world'
+ >>> r.foo
+ 'bar'
+ """
+
+ def __init__(self, d):
+ d.update(self.__dict__)
+ self.__dict__ = d
+ dict.__init__(self, d)
+
+
+class SchemaRDD(RDD):
+ """
+ An RDD of L{Row} objects that has an associated schema. The underlying JVM object is a SchemaRDD,
+ not a PythonRDD, so we can utilize the relational query api exposed by SparkSQL.
+
+ For normal L{pyspark.rdd.RDD} operations (map, count, etc.) the L{SchemaRDD} is not operated on
+ directly, as it's underlying implementation is a RDD composed of Java objects. Instead it is
+ converted to a PythonRDD in the JVM, on which Python operations can be done.
+ """
+
+ def __init__(self, jschema_rdd, sql_ctx):
+ self.sql_ctx = sql_ctx
+ self._sc = sql_ctx._sc
+ self._jschema_rdd = jschema_rdd
+
+ self.is_cached = False
+ self.is_checkpointed = False
+ self.ctx = self.sql_ctx._sc
+ self._jrdd_deserializer = self.ctx.serializer
+
+ @property
+ def _jrdd(self):
+ """
+ Lazy evaluation of PythonRDD object. Only done when a user calls methods defined by the
+ L{pyspark.rdd.RDD} super class (map, count, etc.).
+ """
+ if not hasattr(self, '_lazy_jrdd'):
+ self._lazy_jrdd = self._toPython()._jrdd
+ return self._lazy_jrdd
+
+ @property
+ def _id(self):
+ return self._jrdd.id()
+
+ def saveAsParquetFile(self, path):
+ """
+ Saves the contents of this L{SchemaRDD} as a parquet file, preserving the schema. Files
+ that are written out using this method can be read back in as a SchemaRDD using the
+ L{SQLContext.parquetFile} method.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> srdd.saveAsParquetFile("/tmp/test.parquet")
+ >>> srdd2 = sqlCtx.parquetFile("/tmp/test.parquet")
+ >>> srdd2.collect() == srdd.collect()
+ True
+ """
+ self._jschema_rdd.saveAsParquetFile(path)
+
+ def registerAsTable(self, name):
+ """
+ Registers this RDD as a temporary table using the given name. The lifetime of this temporary
+ table is tied to the L{SQLContext} that was used to create this SchemaRDD.
+
+ >>> srdd = sqlCtx.inferSchema(rdd)
+ >>> srdd.registerAsTable("test")
+ >>> srdd2 = sqlCtx.sql("select * from test")
+ >>> srdd.collect() == srdd2.collect()
+ True
+ """
+ self._jschema_rdd.registerAsTable(name)
+
+ 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, self._sc.serializer).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 cache(self):
+ self.is_cached = True
+ self._jschema_rdd.cache()
+ return self
+
+ def persist(self, storageLevel):
+ self.is_cached = True
+ javaStorageLevel = self.ctx._getJavaStorageLevel(storageLevel)
+ self._jschema_rdd.persist(javaStorageLevel)
+ return self
+
+ def unpersist(self):
+ self.is_cached = False
+ self._jschema_rdd.unpersist()
+ return self
+
+ def checkpoint(self):
+ self.is_checkpointed = True
+ self._jschema_rdd.checkpoint()
+
+ def isCheckpointed(self):
+ return self._jschema_rdd.isCheckpointed()
+
+ def getCheckpointFile(self):
+ checkpointFile = self._jschema_rdd.getCheckpointFile()
+ if checkpointFile.isDefined():
+ return checkpointFile.get()
+ else:
+ return None
+
+def _test():
+ import doctest
+ from pyspark.context import SparkContext
+ globs = globals().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"}])
+ (failure_count, test_count) = doctest.testmod(globs=globs,optionflags=doctest.ELLIPSIS)
+ globs['sc'].stop()
+ if failure_count:
+ exit(-1)
+
+
+if __name__ == "__main__":
+ _test()
+
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/python/run-tests
----------------------------------------------------------------------
diff --git a/python/run-tests b/python/run-tests
index b2b60f0..dabb714 100755
--- a/python/run-tests
+++ b/python/run-tests
@@ -28,6 +28,9 @@ FAILED=0
rm -f unit-tests.log
+# Remove the metastore and warehouse directory created by the HiveContext tests in SparkSQL
+rm -rf metastore warehouse
+
function run_test() {
SPARK_TESTING=0 $FWDIR/bin/pyspark $1 2>&1 | tee -a > unit-tests.log
FAILED=$((PIPESTATUS[0]||$FAILED))
@@ -46,6 +49,7 @@ function run_test() {
run_test "pyspark/rdd.py"
run_test "pyspark/context.py"
run_test "pyspark/conf.py"
+run_test "pyspark/sql.py"
run_test "-m doctest pyspark/broadcast.py"
run_test "-m doctest pyspark/accumulators.py"
run_test "-m doctest pyspark/serializers.py"
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/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 d3d4c56..24d60ea 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,6 +26,7 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.dsl
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.types._
import org.apache.spark.sql.catalyst.optimizer.Optimizer
import org.apache.spark.sql.catalyst.plans.logical.{Subquery, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
@@ -241,4 +242,30 @@ class SQLContext(@transient val sparkContext: SparkContext)
*/
def debugExec() = DebugQuery(executedPlan).execute().collect()
}
+
+ /**
+ * Peek at the first row of the RDD and infer its schema.
+ * TODO: We only support primitive types, add support for nested types.
+ */
+ private[sql] def inferSchema(rdd: RDD[Map[String, _]]): SchemaRDD = {
+ val schema = rdd.first.map { case (fieldName, obj) =>
+ val dataType = obj.getClass match {
+ case c: Class[_] if c == classOf[java.lang.String] => StringType
+ case c: Class[_] if c == classOf[java.lang.Integer] => IntegerType
+ case c: Class[_] if c == classOf[java.lang.Long] => LongType
+ case c: Class[_] if c == classOf[java.lang.Double] => DoubleType
+ case c: Class[_] if c == classOf[java.lang.Boolean] => BooleanType
+ case c => throw new Exception(s"Object of type $c cannot be used")
+ }
+ AttributeReference(fieldName, dataType, true)()
+ }.toSeq
+
+ val rowRdd = rdd.mapPartitions { iter =>
+ iter.map { map =>
+ new GenericRow(map.values.toArray.asInstanceOf[Array[Any]]): Row
+ }
+ }
+ new SchemaRDD(this, SparkLogicalPlan(ExistingRdd(schema, rowRdd)))
+ }
+
}
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/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 9150041..a771147 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
@@ -17,6 +17,8 @@
package org.apache.spark.sql
+import net.razorvine.pickle.Pickler
+
import org.apache.spark.{Dependency, OneToOneDependency, Partition, TaskContext}
import org.apache.spark.annotation.{AlphaComponent, Experimental}
import org.apache.spark.rdd.RDD
@@ -25,6 +27,8 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
import org.apache.spark.sql.catalyst.types.BooleanType
+import org.apache.spark.api.java.JavaRDD
+import java.util.{Map => JMap}
/**
* :: AlphaComponent ::
@@ -308,4 +312,23 @@ class SchemaRDD(
/** FOR INTERNAL USE ONLY */
def analyze = sqlContext.analyzer(logicalPlan)
+
+ private[sql] def javaToPython: JavaRDD[Array[Byte]] = {
+ val fieldNames: Seq[String] = this.queryExecution.analyzed.output.map(_.name)
+ this.mapPartitions { iter =>
+ val pickle = new Pickler
+ iter.map { row =>
+ val map: JMap[String, Any] = new java.util.HashMap
+ // TODO: We place the map in an ArrayList so that the object is pickled to a List[Dict].
+ // Ideally we should be able to pickle an object directly into a Python collection so we
+ // don't have to create an ArrayList every time.
+ val arr: java.util.ArrayList[Any] = new java.util.ArrayList
+ row.zip(fieldNames).foreach { case (obj, name) =>
+ map.put(name, obj)
+ }
+ arr.add(map)
+ pickle.dumps(arr)
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
----------------------------------------------------------------------
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
index 465e5f1..444bbfb 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
@@ -261,8 +261,9 @@ class TestHiveContext(sc: SparkContext) extends LocalHiveContext(sc) {
testTables.get(name).map(_.commands).getOrElse(sys.error(s"Unknown test table $name"))
createCmds.foreach(_())
- if (cacheTables)
+ if (cacheTables) {
cacheTable(name)
+ }
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/sql/hive/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/sql/hive/src/test/resources/log4j.properties b/sql/hive/src/test/resources/log4j.properties
index 5e17e3b..c07d8fe 100644
--- a/sql/hive/src/test/resources/log4j.properties
+++ b/sql/hive/src/test/resources/log4j.properties
@@ -45,3 +45,6 @@ log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=OFF
log4j.additivity.hive.ql.metadata.Hive=false
log4j.logger.hive.ql.metadata.Hive=OFF
+log4j.additivity.org.apache.hadoop.hive.ql.io.RCFile=false
+log4j.logger.org.apache.hadoop.hive.ql.io.RCFile=ERROR
+
http://git-wip-us.apache.org/repos/asf/spark/blob/c99bcb7f/tools/src/main/scala/org/apache/spark/tools/GenerateMIMAIgnore.scala
----------------------------------------------------------------------
diff --git a/tools/src/main/scala/org/apache/spark/tools/GenerateMIMAIgnore.scala b/tools/src/main/scala/org/apache/spark/tools/GenerateMIMAIgnore.scala
index 5547e9f..3fb85e1 100644
--- a/tools/src/main/scala/org/apache/spark/tools/GenerateMIMAIgnore.scala
+++ b/tools/src/main/scala/org/apache/spark/tools/GenerateMIMAIgnore.scala
@@ -99,7 +99,9 @@ object GenerateMIMAIgnore {
// Heuristic to remove JVM classes that do not correspond to user-facing classes in Scala
name.contains("anon") ||
name.endsWith("$class") ||
- name.contains("$sp")
+ name.contains("$sp") ||
+ name.contains("hive") ||
+ name.contains("Hive")
}
/**