Apache Spark - A unified analytics engine for large-scale data processing
Go to file
Herman van Hovell a9a0d0cebf [SPARK-8638] [SQL] Window Function Performance Improvements
## Description
Performance improvements for Spark Window functions. This PR will also serve as the basis for moving away from Hive UDAFs to Spark UDAFs. See JIRA tickets SPARK-8638 and SPARK-7712 for more information.

## Improvements
* Much better performance (10x) in running cases (e.g. BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) and UNBOUDED FOLLOWING cases. The current implementation in spark uses a sliding window approach in these cases. This means that an aggregate is maintained for every row, so space usage is N (N being the number of rows). This also means that all these aggregates all need to be updated separately, this takes N*(N-1)/2 updates. The running case differs from the Sliding case because we are only adding data to an aggregate function (no reset is required), we only need to maintain one aggregate (like in the UNBOUNDED PRECEDING AND UNBOUNDED case), update the aggregate for each row, and get the aggregate value after each update. This is what the new implementation does. This approach only uses 1 buffer, and only requires N updates; I am currently working on data with window sizes of 500-1000 doing running sums and this saves a lot of time. The CURRENT ROW AND UNBOUNDED FOLLOWING case also uses this approach and the fact that aggregate operations are communitative, there is one twist though it will process the input buffer in reverse.
* Fewer comparisons in the sliding case. The current implementation determines frame boundaries for every input row. The new implementation makes more use of the fact that the window is sorted, maintains the boundaries, and only moves them when the current row order changes. This is a minor improvement.
* A single Window node is able to process all types of Frames for the same Partitioning/Ordering. This saves a little time/memory spent buffering and managing partitions. This will be enabled in a follow-up PR.
* A lot of the staging code is moved from the execution phase to the initialization phase. Minor performance improvement, and improves readability of the execution code.

## Benchmarking
I have done a small benchmark using [on time performance](http://www.transtats.bts.gov) data of the month april. I have used the origin as a partioning key, as a result there is quite some variation in window sizes. The code for the benchmark can be found in the JIRA ticket. These are the results per Frame type:

Frame | Master | SPARK-8638
----- | ------ | ----------
Entire Frame | 2 s | 1 s
Sliding | 18 s | 1 s
Growing | 14 s | 0.9 s
Shrinking | 13 s | 1 s

Author: Herman van Hovell <hvanhovell@questtec.nl>

Closes #7057 from hvanhovell/SPARK-8638 and squashes the following commits:

3bfdc49 [Herman van Hovell] Fixed Perfomance Regression for Shrinking Window Frames (+Rebase)
2eb3b33 [Herman van Hovell] Corrected reverse range frame processing.
2cd2d5b [Herman van Hovell] Corrected reverse range frame processing.
b0654d7 [Herman van Hovell] Tests for exotic frame specifications.
e75b76e [Herman van Hovell] More docs, added support for reverse sliding range frames, and some reorganization of code.
1fdb558 [Herman van Hovell] Changed Data In HiveDataFrameWindowSuite.
ac2f682 [Herman van Hovell] Added a few more comments.
1938312 [Herman van Hovell] Added Documentation to the createBoundOrdering methods.
bb020e6 [Herman van Hovell] Major overhaul of Window operator.
2015-07-18 23:44:38 -07:00
assembly [SPARK-7801] [BUILD] Updating versions to SPARK 1.5.0 2015-06-03 10:11:27 -07:00
bagel [SPARK-7801] [BUILD] Updating versions to SPARK 1.5.0 2015-06-03 10:11:27 -07:00
bin [SPARK-7733] [CORE] [BUILD] Update build, code to use Java 7 for 1.5.0+ 2015-06-07 20:18:13 +01:00
build [SPARK-8933] [BUILD] Provide a --force flag to build/mvn that always uses downloaded maven 2015-07-14 11:43:26 -07:00
conf [SPARK-3071] Increase default driver memory 2015-07-01 23:11:02 -07:00
core [SPARK-9171][SQL] add and improve tests for nondeterministic expressions 2015-07-18 11:58:53 -07:00
data/mllib [MLLIB] [DOC] Seed fix in mllib naive bayes example 2015-07-18 10:12:48 -07:00
dev [SPARK-7879] [MLLIB] KMeans API for spark.ml Pipelines 2015-07-17 18:30:04 -07:00
docker [SPARK-8954] [BUILD] Remove unneeded deb repository from Dockerfile to fix build error in docker. 2015-07-13 12:01:23 -07:00
docs [SPARK-6284] [MESOS] Add mesos role, principal and secret 2015-07-16 19:37:15 -07:00
ec2 [SPARK-8596] Add module for rstudio link to spark 2015-07-13 08:15:54 -07:00
examples [SPARK-7977] [BUILD] Disallowing println 2015-07-10 11:34:01 +01:00
external [SPARK-8962] Add Scalastyle rule to ban direct use of Class.forName; fix existing uses 2015-07-14 16:08:17 -07:00
extras [SPARK-9030] [STREAMING] Add Kinesis.createStream unit tests that actual sends data 2015-07-17 16:43:18 -07:00
graphx [SPARK-9109] [GRAPHX] Keep the cached edge in the graph 2015-07-17 12:11:32 -07:00
launcher [SPARK-9001] Fixing errors in javadocs that lead to failed build/sbt doc 2015-07-14 00:32:29 -07:00
mllib [SPARK-9118] [ML] Implement IntArrayParam in mllib 2015-07-17 20:02:05 -07:00
network [SPARK-3071] Increase default driver memory 2015-07-01 23:11:02 -07:00
project [SPARK-8278] Remove non-streaming JSON reader. 2015-07-18 20:27:55 -07:00
python [SPARK-8199][SPARK-8184][SPARK-8183][SPARK-8182][SPARK-8181][SPARK-8180][SPARK-8179][SPARK-8177][SPARK-8178][SPARK-9115][SQL] date functions 2015-07-18 22:48:05 -07:00
R [SPARK-9093] [SPARKR] Fix single-quotes strings in SparkR 2015-07-17 17:00:50 +09:00
repl [SPARK-9015] [BUILD] Clean project import in scala ide 2015-07-16 18:42:41 +01:00
sbin [SPARK-5412] [DEPLOY] Cannot bind Master to a specific hostname as per the documentation 2015-05-15 11:30:19 -07:00
sbt Adde LICENSE Header to build/mvn, build/sbt and sbt/sbt 2014-12-29 10:48:53 -08:00
sql [SPARK-8638] [SQL] Window Function Performance Improvements 2015-07-18 23:44:38 -07:00
streaming [SPARK-5681] [STREAMING] Move 'stopReceivers' to the event loop to resolve the race condition 2015-07-17 14:00:31 -07:00
tools [SPARK-9015] [BUILD] Clean project import in scala ide 2015-07-16 18:42:41 +01:00
unsafe [SPARK-8240][SQL] string function: concat 2015-07-18 14:07:56 -07:00
yarn [SPARK-8851] [YARN] In Client mode, make sure the client logs in and updates tokens 2015-07-17 09:38:08 -05:00
.gitattributes [SPARK-3870] EOL character enforcement 2014-10-31 12:39:52 -07:00
.gitignore [SPARK-8495] [SPARKR] Add a .lintr file to validate the SparkR files and the lint-r script 2015-06-20 16:10:14 -07:00
.rat-excludes [SPARK-6123] [SPARK-6775] [SPARK-6776] [SQL] Refactors Parquet read path for interoperability and backwards-compatibility 2015-07-08 15:51:01 -07:00
CONTRIBUTING.md [SPARK-6889] [DOCS] CONTRIBUTING.md updates to accompany contribution doc updates 2015-04-21 22:34:31 -07:00
LICENSE [SPARK-8709] Exclude hadoop-client's mockito-all dependency 2015-06-29 14:07:55 -07:00
make-distribution.sh [SPARK-6797] [SPARKR] Add support for YARN cluster mode. 2015-07-13 08:21:47 -07:00
NOTICE SPARK-1827. LICENSE and NOTICE files need a refresh to contain transitive dependency info 2014-05-14 09:38:33 -07:00
pom.xml [SPARK-9015] [BUILD] Clean project import in scala ide 2015-07-16 18:42:41 +01:00
pylintrc [SPARK-8706] [PYSPARK] [PROJECT INFRA] Add pylint checks to PySpark 2015-07-15 08:25:53 -07:00
README.md Update README to include DataFrames and zinc. 2015-05-31 23:55:45 -07:00
scalastyle-config.xml [SPARK-8962] Add Scalastyle rule to ban direct use of Class.forName; fix existing uses 2015-07-14 16:08:17 -07:00
tox.ini [SPARK-7427] [PYSPARK] Make sharedParams match in Scala, Python 2015-05-10 19:18:32 -07:00

Apache Spark

Spark is a fast and general cluster computing system for Big Data. It provides high-level APIs in Scala, Java, and Python, and an optimized engine that supports general computation graphs for data analysis. It also supports a rich set of higher-level tools including Spark SQL for SQL and DataFrames, MLlib for machine learning, GraphX for graph processing, and Spark Streaming for stream processing.

http://spark.apache.org/

Online Documentation

You can find the latest Spark documentation, including a programming guide, on the project web page and project wiki. This README file only contains basic setup instructions.

Building Spark

Spark is built using Apache Maven. To build Spark and its example programs, run:

build/mvn -DskipTests clean package

(You do not need to do this if you downloaded a pre-built package.) More detailed documentation is available from the project site, at "Building Spark".

Interactive Scala Shell

The easiest way to start using Spark is through the Scala shell:

./bin/spark-shell

Try the following command, which should return 1000:

scala> sc.parallelize(1 to 1000).count()

Interactive Python Shell

Alternatively, if you prefer Python, you can use the Python shell:

./bin/pyspark

And run the following command, which should also return 1000:

>>> sc.parallelize(range(1000)).count()

Example Programs

Spark also comes with several sample programs in the examples directory. To run one of them, use ./bin/run-example <class> [params]. For example:

./bin/run-example SparkPi

will run the Pi example locally.

You can set the MASTER environment variable when running examples to submit examples to a cluster. This can be a mesos:// or spark:// URL, "yarn-cluster" or "yarn-client" to run on YARN, and "local" to run locally with one thread, or "local[N]" to run locally with N threads. You can also use an abbreviated class name if the class is in the examples package. For instance:

MASTER=spark://host:7077 ./bin/run-example SparkPi

Many of the example programs print usage help if no params are given.

Running Tests

Testing first requires building Spark. Once Spark is built, tests can be run using:

./dev/run-tests

Please see the guidance on how to run tests for a module, or individual tests.

A Note About Hadoop Versions

Spark uses the Hadoop core library to talk to HDFS and other Hadoop-supported storage systems. Because the protocols have changed in different versions of Hadoop, you must build Spark against the same version that your cluster runs.

Please refer to the build documentation at "Specifying the Hadoop Version" for detailed guidance on building for a particular distribution of Hadoop, including building for particular Hive and Hive Thriftserver distributions. See also "Third Party Hadoop Distributions" for guidance on building a Spark application that works with a particular distribution.

Configuration

Please refer to the Configuration guide in the online documentation for an overview on how to configure Spark.