Number reduce tasks Spark

Question

What is the formula that Spark uses to calculate the number of reduce tasks?

I am running a couple of spark-sql queries and the number of reduce tasks always is 200. The number of map tasks for these queries is 154. I am on Spark 1.4.1.

Is this related to spark.shuffle.sort.bypassMergeThreshold, which defaults to 200

sgvd · Accepted Answer · 2021-05-11T13:56:59.683

It's spark.sql.shuffle.partitions that you're after. According to the Spark SQL performance tuning guide:

| Property Name                 | Default | Meaning                                        |
+-------------------------------+---------+------------------------------------------------+
| spark.sql.shuffle.partitions  | 200     | Configures the number of partitions to use     |
|                               |         | when shuffling data for joins or aggregations. |

Another option that is related is spark.default.parallelism, which determines the 'default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set by user', however this seems to be ignored by Spark SQL and only relevant when working on plain RDDs.

spark.default.parallelism is giving maximum partition if data is that large to handle otherwise transformations like join,reduceByKey etc can generate couple of partition as output. — Bhavesh Gadoya, Jun 07 '19 at 08:34

score 4 · Answer 2 · answered May 06 '16 at 19:01

4

Yes, @svgd, that is the correct parameter. Here is how you reset it in Scala:

// Set number of shuffle partitions to 3
sqlContext.setConf("spark.sql.shuffle.partitions", "3")
// Verify the setting 
sqlContext.getConf("spark.sql.shuffle.partitions")

answered May 06 '16 at 19:01

pmhargis

713
6
7

score 4 · Answer 3 · answered May 24 '19 at 07:14

4

Nowadays in Spark 2 + to set this parameter do the following

spark.conf.set("spark.sql.shuffle.partitions", 16)

answered May 24 '19 at 07:14

Jack

645
7
14

score 0 · Answer 4 · answered Mar 23 '20 at 12:56

Specifying the min and max split size through mapreduce.input.fileinputformat.split should help. These parameters determine the respective minimum and maximum chunk sizes for splitting the input files to.

val spark = SparkSession.builder
    .config("mapreduce.input.fileinputformat.split.minsize", "1073741824")
    .config("mapreduce.input.fileinputformat.split.maxsize", "1073741824")           
    .enableHiveSupport().getOrCreate()

Here, the split size has been kept 1GB (1073741824 bytes). To remember that parquet, snappy are splittable and gzip, lzo aren't. Refer more here.

Number reduce tasks Spark

4 Answers4

Linked