What is the difference between spark.sql.shuffle.partitions and spark.default.parallelism?

Question

What's the difference between spark.sql.shuffle.partitions and spark.default.parallelism?

I have tried to set both of them in SparkSQL, but the task number of the second stage is always 200.

Answer 1

To add onto what some great answers have already posted:

Summary

• spark.sql.shuffle.partitions:
  • Determines how many output partitions you will have after doing wide transformations on Dataframes/Datasets by default.
  • Its default value is 200.
• spark.default.parallelism:
  • Is a more complicated parameter that lives more "deep down" in Spark. It influences:
    • how many partitions you will have after doing wide transformations on RDDs if you don't specify an amount
    • how many partitions sc.parallelize creates
    • how many partitions are read in when doing spark.read.csv, ...
  • Its default value depends on what type of operation you do on which type of cluster.

A bit more detail

spark.sql.shuffle.partitions

From the docs:

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 restarts from the same checkpoint location.

As can be seen in Spark 3.3.1's (newest version at the time of this post) SQLConf.scala, spark.sql.shuffle.partitions has a default value of 200.

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 " +
    "restarts from the same checkpoint location.")
  .version("1.1.0")
  .intConf
  .checkValue(_ > 0, "The value of spark.sql.shuffle.partitions must be positive")
  .createWithDefault(200)

Conclusion: This is a value with which you can have immediate impact on your wide transformations (join, sort, ...) on Dataframes/Datasets. You'll be able to configure the amount of output partitions of those wide transformations with this parameter.

spark.default.parallelism

From the docs:

Default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set by user.

About its default value:

For distributed shuffle operations like reduceByKey and join, the largest number of partitions in a parent RDD. For operations like parallelize with no parent RDDs, it depends on the cluster manager:

• Local mode: number of cores on the local machine
• Mesos fine grained mode: 8
• Others: total number of cores on all executor nodes or 2, whichever is larger

So we already see this parameter is a bit more complicated. It has no real default value, but you can set it. This becomes clearer if we look at a bit of code.

In SparkContext.scala, we see how defaultParallelism is defined:

/** Default level of parallelism to use when not given by user (e.g. parallelize and makeRDD). */
def defaultParallelism: Int = {
  assertNotStopped()
  taskScheduler.defaultParallelism
}

So we see that this defaultParallelism is dependent on the type of taskScheduler (like the docs state). Let's have a look at those:

• local mode:

override def defaultParallelism(): Int =
  scheduler.conf.getInt("spark.default.parallelism", totalCores)

• Mesos fine grained mode:

override def defaultParallelism(): Int = sc.conf.getInt("spark.default.parallelism", 8)

• Others (CoarseGrainSchedulerBackend):

override def defaultParallelism(): Int = {
  conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))
}

Ok, so now we understand that this value is a bit more complicated let's try to figure out when it is relevant:

• When doing wide transformations on RDDs. join, reduceByKey, groupByKey, etc. all make use of a defaultPartitioner if no partitioning is given as input parameter. In that defaultPartitioner, spark.default.parallelism is used to determine the amount of partitions.
• When calling sc.parallelize on a Seq. Depending on your cluster manager, you'll get a number of output partitions as explained above.
• When reading in data (for example spark.read.csv), it will have an impact on how many partitions will be read in. In DataSourceScanExec's createReadRDD function, the amount of output partitions that will be read in are influenced by the maxSplitBytes function, which is in itself influenced by spark.default.parallelism as explained in this SO answer.
• I'm sure it's used in more place but I hope this already gives more intuition about this parameter.

Answer 2

In case someone who might want to know, there's indeed a special situation when the setting spark.sql.shuffle.partitions might become invalid. When you restart a structured streaming spark application with the same checkpoint location, changing this term does not take effect. See more at https://spark.apache.org/docs/latest/configuration.html#runtime-sql-configuration

Answer 3

spark.default.parallelism is the default number of partition set by spark which is by default 200. and if you want to increase the number of partition than you can apply the property spark.sql.shuffle.partitions to set number of partition in the spark configuration or while running spark SQL.

Normally this spark.sql.shuffle.partitions it is being used when we have a memory congestion and we see below error: spark error:java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE

so set your can allocate a partition as 256 MB per partition and that you can use to set for your processes.

also If number of partitions is near to 2000 then increase it to more than 2000. As spark applies different logic for partition < 2000 and > 2000 which will increase your code performance by decreasing the memory footprint as data default is highly compressed if >2000.

Answer 4

From the answer here, spark.sql.shuffle.partitions configures the number of partitions that are used when shuffling data for joins or aggregations.

spark.default.parallelism is the default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set explicitly by the user. Note that spark.default.parallelism seems to only be working for raw RDD and is ignored when working with dataframes.

If the task you are performing is not a join or aggregation and you are working with dataframes then setting these will not have any effect. You could, however, set the number of partitions yourself by calling df.repartition(numOfPartitions) (don't forget to assign it to a new val) in your code.

---

To change the settings in your code you can simply do:

sqlContext.setConf("spark.sql.shuffle.partitions", "300")
sqlContext.setConf("spark.default.parallelism", "300")

Alternatively, you can make the change when submitting the job to a cluster with spark-submit:

./bin/spark-submit --conf spark.sql.shuffle.partitions=300 --conf spark.default.parallelism=300