How to configure a Spark broadcast join

Spark joins usually shuffle rows with matching keys into the same partitions unless the optimizer can copy one side to every executor. A broadcast join is useful when a large fact DataFrame joins with a small lookup DataFrame, because Spark can send the lookup side once instead of repartitioning both inputs.

Spark can choose broadcast automatically from table statistics when the candidate side is below spark.sql.autoBroadcastJoinThreshold. A PySpark broadcast() hint marks one DataFrame as the build side for one join, which keeps the change local to that query instead of raising the session threshold for unrelated joins.

The local proof run disables automatic broadcast so the physical plan changes only after the hint is added. Use the same pattern only for data that fits executor memory; broadcasting a large dimension table can fail the job or push cached data out of memory.

1. Choose the smaller lookup or dimension DataFrame as the broadcast side.

   Do not broadcast the large fact side. Spark copies the broadcast side to executors, so the selected DataFrame must fit comfortably in executor memory after serialization.

2. Create a PySpark check script.

   $ vi broadcast_join_check.py

3. Build the Spark session and sample DataFrames.

   from pyspark.sql import SparkSession, functions as F
    
   spark = (
       SparkSession.builder
       .appName("sg-broadcast-join-check")
       .config("spark.sql.shuffle.partitions", "2")
       .config("spark.sql.autoBroadcastJoinThreshold", "-1")
       .getOrCreate()
   )
   spark.sparkContext.setLogLevel("ERROR")
    
   orders = spark.createDataFrame(
       [
           (1, "MY", 42.50),
           (2, "SG", 31.20),
           (3, "MY", 18.90),
           (4, "ID", 25.00),
       ],
       ["order_id", "country_code", "amount"],
   )
    
   countries = spark.createDataFrame(
       [
           ("MY", "Malaysia"),
           ("SG", "Singapore"),
           ("ID", "Indonesia"),
       ],
       ["country_code", "country_name"],
   )

   The proof run sets spark.sql.autoBroadcastJoinThreshold to -1 so the explicit hint is the reason Spark changes the join strategy. For session-wide automatic broadcast, set this threshold to the largest small-table size Spark should broadcast.

4. Add the unhinted join and the broadcast-hinted join.

   plain_join = orders.join(countries, "country_code")
   plain_plan = plain_join._jdf.queryExecution().executedPlan().toString()
    
   broadcast_join = orders.join(F.broadcast(countries), "country_code").select(
       "order_id", "country_code", "country_name", "amount"
   )
   broadcast_plan = broadcast_join._jdf.queryExecution().executedPlan().toString()

   F.broadcast(countries) marks the lookup DataFrame as the build side for this join. Spark can still ignore a strategy hint when the join type cannot use that strategy.

5. Print the join-plan signal and a joined-row smoke test.

   print("autoBroadcastJoinThreshold =", spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))
   print("plain join uses BroadcastHashJoin =", "BroadcastHashJoin" in plain_plan)
   print("broadcast join uses BroadcastHashJoin =", "BroadcastHashJoin" in broadcast_plan)
   print()
   broadcast_join.explain(mode="simple")
   broadcast_join.orderBy("order_id").show(truncate=False)
    
   spark.stop()

6. Run the check script in local mode.

   $ spark-submit --master 'local[2]' broadcast_join_check.py
   autoBroadcastJoinThreshold = -1
   plain join uses BroadcastHashJoin = False
   broadcast join uses BroadcastHashJoin = True
   
   == Physical Plan ==
   AdaptiveSparkPlan isFinalPlan=false
   +- Project [order_id#0L, country_code#1, country_name#4, amount#2]
      +- BroadcastHashJoin [country_code#1], [country_code#3], Inner, BuildRight, false
         :- Filter isnotnull(country_code#1)
         :  +- Scan ExistingRDD[order_id#0L,country_code#1,amount#2]
         +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=55]
            +- Filter isnotnull(country_code#3)
               +- Scan ExistingRDD[country_code#3,country_name#4]
   
   +--------+------------+------------+------+
   |order_id|country_code|country_name|amount|
   +--------+------------+------------+------+
   |1       |MY          |Malaysia    |42.5  |
   |2       |SG          |Singapore   |31.2  |
   |3       |MY          |Malaysia    |18.9  |
   |4       |ID          |Indonesia   |25.0  |
   +--------+------------+------------+------+

   The plain join line proves automatic broadcast was disabled for the check. The broadcast join line and BroadcastHashJoin plan node prove the hint changed the physical join strategy.

7. Remove the proof script when it is not part of the application code.

   $ rm broadcast_join_check.py