No description, website, or topics provided.
Clone or download
gczsjdy and plusplusjiajia Reduce shuffles for successive full outer join (#63)
* Modify existed Partioning & Distribution to reduce shuffles for full outer join

* Refactor and test
Latest commit 61bd1c9 Sep 16, 2018
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
.github [SPARK-18073][DOCS][WIP] Migrate wiki to spark.apache.org web site Nov 23, 2016
R Preparing Spark release v2.3.2-rc6 Sep 16, 2018
assembly Preparing Spark release v2.3.2-rc6 Sep 16, 2018
bin Fix issue in 'docker-image-tool.sh' Jun 18, 2018
build [SPARK-19810][BUILD][CORE] Remove support for Scala 2.10 Jul 13, 2017
common Optimize shuffle fetch of contiguous partition Sep 27, 2018
conf [SPARK-22466][SPARK SUBMIT] export SPARK_CONF_DIR while conf is default Nov 9, 2017
core Handle skewed join at runtime Sep 27, 2018
data [SPARK-23205][ML] Update ImageSchema.readImages to correctly set alph… Jan 26, 2018
dev [PYSPARK] Updates to pyspark broadcast Sep 13, 2018
docs Preparing Spark release v2.3.2-rc6 Sep 16, 2018
examples Preparing Spark release v2.3.2-rc6 Sep 16, 2018
external Preparing Spark release v2.3.2-rc6 Sep 16, 2018
graphx Preparing Spark release v2.3.2-rc6 Sep 16, 2018
hadoop-cloud Preparing Spark release v2.3.2-rc6 Sep 16, 2018
launcher Preparing Spark release v2.3.2-rc6 Sep 16, 2018
licenses [SPARK-19112][CORE] Support for ZStandard codec Nov 1, 2017
mllib-local Preparing Spark release v2.3.2-rc6 Sep 16, 2018
mllib Preparing Spark release v2.3.2-rc6 Sep 16, 2018
project [SPARK-23732][DOCS] Fix source links in generated scaladoc. Jun 12, 2018
python Preparing Spark release v2.3.2-rc6 Sep 16, 2018
repl Preparing Spark release v2.3.2-rc6 Sep 16, 2018
resource-managers Preparing Spark release v2.3.2-rc6 Sep 16, 2018
sbin [PYSPARK] Update py4j to version 0.10.7. May 10, 2018
sql Reduce shuffles for successive full outer join (#63) Jan 15, 2019
streaming Preparing Spark release v2.3.2-rc6 Sep 16, 2018
tools Preparing Spark release v2.3.2-rc6 Sep 16, 2018
.gitattributes [SPARK-3870] EOL character enforcement Oct 31, 2014
.gitignore [SPARK-21485][SQL][DOCS] Spark SQL documentation generation for built… Jul 26, 2017
.travis.yml [SPARK-18278][SCHEDULER] Spark on Kubernetes - Basic Scheduler Backend Nov 29, 2017
CONTRIBUTING.md [SPARK-18073][DOCS][WIP] Migrate wiki to spark.apache.org web site Nov 23, 2016
LICENSE [PYSPARK] Update py4j to version 0.10.7. May 10, 2018
NOTICE [MINOR][BUILD] Add ECCN notice required by http://www.apache.org/dev/… Aug 10, 2018
README.md Add information about external shuffle service in readme Sep 27, 2018
appveyor.yml [SPARK-22817][R] Use fixed testthat version for SparkR tests in AppVeyor Dec 17, 2017
pom.xml Preparing Spark release v2.3.2-rc6 Sep 16, 2018
scalastyle-config.xml [SPARK-20657][CORE] Speed up rendering of the stages page. Jan 11, 2018

README.md

Spark SQL Adaptive Execution

There are three main features in Adaptive Execution, including auto setting the shuffle partition number, optimizing join strategy at runtime and handling skewed join. These features can be enabled separately. To start with Adaptive Exection on Spark 2.3, please build branch ae-2.3-08 and at least set spark.sql.adaptive.enabled to true. For users who enabled external shuffle service, please also upgrade external shuffle service to use adaptive execution feature.

An Engilish version design doc is available on google doc. A Chinese version blog is available on CSDN that introduces the features and benchmark results. SPARK-23128 is the Jira for contributing this work to Apache Spark.

Auto Setting The Shuffle Partition Number

Property NameDefaultMeaning
spark.sql.adaptive.enabled false When true, enable adaptive query execution.
spark.sql.adaptive.minNumPostShufflePartitions 1 The minimum number of post-shuffle partitions used in adaptive execution. This can be used to control the minimum parallelism.
spark.sql.adaptive.maxNumPostShufflePartitions 500 The maximum number of post-shuffle partitions used in adaptive execution. This is also used as the initial shuffle partition number so please set it to an reasonable value.
spark.sql.adaptive.shuffle.targetPostShuffleInputSize 67108864 The target post-shuffle input size in bytes of a task. By default is 64 MB.
spark.sql.adaptive.shuffle.targetPostShuffleRowCount 20000000 The target post-shuffle row count of a task. This only takes effect if row count information is collected.

Optimizing Join Strategy at Runtime

Property NameDefaultMeaning
spark.sql.adaptive.join.enabled true When true and spark.sql.adaptive.enabled is enabled, a better join strategy is determined at runtime.
spark.sql.adaptiveBroadcastJoinThreshold equals to spark.sql.autoBroadcastJoinThreshold Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join in adaptive exeuction mode. If not set, it equals to spark.sql.autoBroadcastJoinThreshold.

Handling Skewed Join

Property NameDefaultMeaning
spark.sql.adaptive.skewedJoin.enabled false When true and spark.sql.adaptive.enabled is enabled, a skewed join is automatically handled at runtime.
spark.sql.adaptive.skewedPartitionFactor 10 A partition is considered as a skewed partition if its size is larger than this factor multiple the median partition size and also larger than spark.sql.adaptive.skewedPartitionSizeThreshold, or if its row count is larger than this factor multiple the median row count and also larger than spark.sql.adaptive.skewedPartitionRowCountThreshold.
spark.sql.adaptive.skewedPartitionSizeThreshold 67108864 Configures the minimum size in bytes for a partition that is considered as a skewed partition in adaptive skewed join.
spark.sql.adaptive.skewedPartitionRowCountThreshold 10000000 Configures the minimum row count for a partition that is considered as a skewed partition in adaptive skewed join.
spark.shuffle.statistics.verbose false Collect shuffle statistics in verbose mode, including row counts etc. This is required for handling skewed join.