Skip to content

[SPARK-31949][SQL] Add spark.default.parallelism in SQLConf for isolated across session #28778

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ulysses-you wants to merge 5 commits into from
Closed

[SPARK-31949][SQL] Add spark.default.parallelism in SQLConf for isolated across session #28778

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
35b92c2
init
ulysses-you Jun 10, 2020
5c6d661
Add-method-in-SparkSession
ulysses-you Jun 10, 2020
27e113b
replace-defaultParallelism
ulysses-you Jun 10, 2020
e858c27
update doc
ulysses-you Jun 10, 2020
55d367f
add ut
ulysses-you Jun 10, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,14 @@ object SQLConf {
.booleanConf
.createWithDefault(true)

val DEFAULT_PARALLELISM = buildConf("spark.sql.default.parallelism")

@maropu maropu Jun 10, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

spark.sql.default.parallelism -> spark.sql.sessionLocalDefaultParallelism?

@ulysses-you ulysses-you Jun 10, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Em.. is it better to keep similar with spark.default.parallelism? so we can set this config easy. sessionLocalDefaultParallelism seems complex.

.doc("This config behavior is same as spark.default.parallelism, and this value can be " +
"isolated across sessions. Note: always use sc.defaultParallelism as default number.")

@maropu maropu Jun 10, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about this?

The session-local default number of partitions and and this value is widely used inside physical plans.
If not set, the physical plans refer to `spark.default.parallelism` instead.

.version("3.1.0")
.intConf
.checkValue(_ > 0, "The value of spark.sql.default.parallelism must be positive")
.createOptional

val SHUFFLE_PARTITIONS = buildConf("spark.sql.shuffle.partitions")
.doc("The default number of partitions to use when shuffling data for joins or aggregations. " +
"Note: For structured streaming, this configuration cannot be changed between query " +
Expand Down Expand Up @@ -2784,6 +2792,8 @@ class SQLConf extends Serializable with Logging {

def cacheVectorizedReaderEnabled: Boolean = getConf(CACHE_VECTORIZED_READER_ENABLED)

def defaultParallelism: Option[Int] = getConf(DEFAULT_PARALLELISM)

def defaultNumShufflePartitions: Int = getConf(SHUFFLE_PARTITIONS)

def numShufflePartitions: Int = {
Expand Down
9 changes: 9 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,15 @@ class SparkSession private(
*/
@transient lazy val conf: RuntimeConfig = new RuntimeConfig(sessionState.conf)

/**
* Same as `spark.default.parallelism`, can be isolated across sessions.
*
* @since 3.1.0
*/
def defaultParallelism: Int = {

@cloud-fan cloud-fan Jun 11, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd like to not have this API, as SparkSession should provide high-level logical APIs, not physical ones.

sessionState.conf.defaultParallelism.getOrElse(sparkContext.defaultParallelism)

@cloud-fan cloud-fan Jun 10, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

so we add a config, whose only usage is to let users get the config value?

@ulysses-you ulysses-you Jun 10, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As I said above. If add this config, I will move the exists defaultParallelism which in sql module follow up. e.g. FilePartition.maxSplitBytes()

@ulysses-you ulysses-you Jun 10, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just do this in this pr ?

@cloud-fan cloud-fan Jun 10, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please do, otherwise it's a useless config

}

/**
* An interface to register custom [[org.apache.spark.sql.util.QueryExecutionListener]]s
* that listen for execution metrics.
Expand Down