Apache Spark - A unified analytics engine for large-scale data processing
Go to file
Yin Huai e3355090d4 [SPARK-10143] [SQL] Use parquet's block size (row group size) setting as the min split size if necessary.
https://issues.apache.org/jira/browse/SPARK-10143

With this PR, we will set min split size to parquet's block size (row group size) set in the conf if the min split size is smaller. So, we can avoid have too many tasks and even useless tasks for reading parquet data.

I tested it locally. The table I have has 343MB and it is in my local FS. Because I did not set any min/max split size, the default split size was 32MB and the map stage had 11 tasks. But there were only three tasks that actually read data. With my PR, there were only three tasks in the map stage. Here is the difference.

Without this PR:
![image](https://cloud.githubusercontent.com/assets/2072857/9399179/8587dba6-4765-11e5-9189-7ebba52a2b6d.png)

With this PR:
![image](https://cloud.githubusercontent.com/assets/2072857/9399185/a4735d74-4765-11e5-8848-1f1e361a6b4b.png)

Even if the block size setting does match the actual block size of parquet file, I think it is still generally good to use parquet's block size setting if min split size is smaller than this block size.

Tested it on a cluster using
```
val count = sqlContext.table("""store_sales""").groupBy().count().queryExecution.executedPlan(3).execute().count
```
Basically, it reads 0 column of table `store_sales`. My table has 1824 parquet files with size from 80MB to 280MB (1 to 3 row group sizes). Without this patch, in a 16 worker cluster, the job had 5023 tasks and spent 102s. With this patch, the job had 2893 tasks and spent 64s. It is still not as good as using one mapper per file (1824 tasks and 42s), but it is much better than our master.

Author: Yin Huai <yhuai@databricks.com>

Closes #8346 from yhuai/parquetMinSplit.
2015-08-21 14:30:00 -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-9270] [PYSPARK] allow --name option in pyspark 2015-07-24 11:56:55 -07:00
build [SPARK-9633] [BUILD] SBT download locations outdated; need an update 2015-08-06 23:43:52 +01:00
conf [SPARK-8118] [SQL] Redirects Parquet JUL logger via SLF4J 2015-08-18 20:15:33 +08:00
core [SPARK-9439] [YARN] External shuffle service robust to NM restarts using leveldb 2015-08-21 08:41:36 -05:00
data/mllib [MLLIB] [DOC] Seed fix in mllib naive bayes example 2015-07-18 10:12:48 -07:00
dev [SPARK-10126] [PROJECT INFRA] Fix typo in release-build.sh which broke snapshot publishing for Scala 2.11 2015-08-20 11:31:03 -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-9846] [DOCS] User guide for Multilayer Perceptron Classifier 2015-08-20 20:02:27 -07:00
ec2 [SPARK-9562] Change reference to amplab/spark-ec2 from mesos/ 2015-08-04 09:40:07 -07:00
examples [SPARK-9812] [STREAMING] Fix Python 3 compatibility issue in PySpark Streaming and some docs 2015-08-19 18:36:01 -07:00
external [SPARK-9574] [STREAMING] Remove unnecessary contents of spark-streaming-XXX-assembly jars 2015-08-18 13:35:45 -07:00
extras [SPARK-9574] [STREAMING] Remove unnecessary contents of spark-streaming-XXX-assembly jars 2015-08-18 13:35:45 -07:00
graphx [SPARK-9960] [GRAPHX] sendMessage type fix in LabelPropagation.scala 2015-08-14 21:28:50 -07:00
launcher [SPARK-9980] [BUILD] Fix SBT publishLocal error due to invalid characters in doc 2015-08-15 10:46:04 +01:00
mllib [SPARK-9864] [DOC] [MLlib] [SQL] Replace since in scaladoc to Since annotation 2015-08-21 14:19:24 -07:00
network [SPARK-9439] [YARN] External shuffle service robust to NM restarts using leveldb 2015-08-21 08:41:36 -05:00
project [SPARK-9580] [SQL] Replace singletons in SQL tests 2015-08-13 17:42:01 -07:00
python [SPARK-10122] [PYSPARK] [STREAMING] Fix getOffsetRanges bug in PySpark-Streaming transform function 2015-08-21 13:15:35 -07:00
R [SPARK-10106] [SPARKR] Add ifelse Column function to SparkR 2015-08-19 12:39:37 -07:00
repl [SPARK-9602] remove "Akka/Actor" words from comments 2015-08-04 14:54:11 -07:00
sbin [SPARK-8064] [SQL] Build against Hive 1.2.1 2015-08-03 15:24:42 -07:00
sbt Adde LICENSE Header to build/mvn, build/sbt and sbt/sbt 2014-12-29 10:48:53 -08:00
sql [SPARK-10143] [SQL] Use parquet's block size (row group size) setting as the min split size if necessary. 2015-08-21 14:30:00 -07:00
streaming [SPARK-10128] [STREAMING] Used correct classloader to deserialize WAL data 2015-08-19 21:15:58 -07:00
tools [SPARK-9015] [BUILD] Clean project import in scala ide 2015-07-16 18:42:41 +01:00
unsafe [SPARK-10095] [SQL] use public API of BigInteger 2015-08-18 20:39:59 -07:00
yarn [SPARK-9439] [YARN] External shuffle service robust to NM restarts using leveldb 2015-08-21 08:41:36 -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-9340] [SQL] Fixes converting unannotated Parquet lists 2015-08-11 12:46:33 +08: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-9199] [CORE] Upgrade Tachyon version from 0.7.0 -> 0.7.1. 2015-08-17 08:28:16 +01: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-9439] [YARN] External shuffle service robust to NM restarts using leveldb 2015-08-21 08:41:36 -05:00
pylintrc [SPARK-9116] [SQL] [PYSPARK] support Python only UDT in __main__ 2015-07-29 22:30:49 -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.