I use Spark 2.1 in local mode and I'm running this simple application.
val N = 10 << 20
sparkSession.conf.set("spark.sql.shuffle.partitions", "5")
sparkSession.conf.set("spark.sql.autoBroadcastJoinThreshold", (N + 1).toString)
sparkSession.conf.set("spark.sql.join.preferSortMergeJoin", "false")
val df1 = sparkSession.range(N).selectExpr(s"id as k1")
val df2 = sparkSession.range(N / 5).selectExpr(s"id * 3 as k2")
df1.join(df2, col("k1") === col("k2")).count()
Here, the range(N) creates a dataset of Long (with unique values), so I assume that the size of
- df1 = N * 8 bytes ~ 80MB
- df2 = N / 5 * 8 bytes ~ 16MB
Ok now let's take df1 as an example. df1 consists of 8 partitions and shuffledRDDs of 5, so I assume that
- # of mappers (M) = 8
- # of reducers (R) = 5
As the # of partitions is low, Spark will use the Hash Shuffle which will create M * R files in the disk but I haven't understood if every file has all the data, thus each_file_size = data_size resulting to M * R * data_size files or all_files = data_size.
However when executing this app, shuffle write of df1 = 160MB which doesn't match either of the above cases.
What am I missing here? Why has the shuffle write data doubled in size?