Why Spark applies a broadcast join for a file with size larger than the autoBroadcastJoinThreshold?

user_7770010 0
Viewed 417

I am using spark 3.1.1 and joining two Dataframes of file size 8.6Gb and 25.2Mb respectively and not applying any filter. Spark is automatically using BroadcastHashJoin for this, although spark.sql.autoBroadcastJoinThreshold defaults to 10Mb.

How is 25.2Mb converted to 8.1Mb without applying any filter to be eligible for broadcast?

val df1 = spark.read
  .option("header",true)
  .csv("s3a://data/staging/received/data/spark/3/KernelVersionOutputFiles.csv")
  .withColumn("Pid",substring(rand(),3,4).cast("bigint"))


val df2 = spark.read
  .option("header",true)
  .csv("s3a://data/staging/received/data/spark/3/ForumTopics.csv")
  .withColumn("Cid",substring(rand(),3,4).cast("bigint"))

val df3 = df2.coalesce(1)
val joinDf = df1.join(df3, df1("Pid") === df3("Cid"))
val cnt = joinDf.count()

DAG looks like this:

enter image description here

apache-spark-sql apache-spark
1 Answers

Spark applies a broadcast join, because the data of 25MB in the csv ("size of files read") will be lower than 10MB when serialized by Spark ("data size").

The amount shown with "size of files read" is pretty accurate because Spark is able to compute the statistics directly on the files of data. However, the "data size" shown in the DAG suffers from the inaccuracy of the SizeEstimator.

There it says:

"Estimate the number of bytes that the given object takes up on the JVM heap. The estimate includes space taken up by objects referenced by the given object, their references, and so on and so forth.

This is useful for determining the amount of heap space a broadcast variable will occupy on each executor or the amount of space each object will take when caching objects in deserialized form. This is not the same as the serialized size of the object, which will typically be much smaller."

If you want to get the actual size of you 25MB csv file, you could cache it and check the "Storage" tab in the WebUI.

In my test case, although I also left the configuration autoBroadcastJoinThreshold by default to 10MB, Spark applied a broadcast join. The estimation was 66MB for a json file of size 14MB. When I cached it, it showed a size of 3.5MB which is obviously lower than the threshold of 10MB.

The following picture shows my test case (similar to yours):

enter image description here

The following screenshot shows the actual size of the data, which is only 3.5MB:

enter image description here

Another reference is given from Microsoft about this here.

user_122089100
Related