Using Spark Broadcast Joins to Speed Up Large Dataset Joins

When Spark joins two DataFrames larger than a few gigabytes, the default strategy is a hash join or sort-merge join that shuffles both sides across the network. Each executor reads its partition, sends rows to the right reducer, builds a hash table, and only then can the join happen. If one side is an order of magnitude larger than the other, most of that traffic is wasted – the big table is essentially streamed across the cluster only to be discarded after the join builds its side.

In practice, this shows up as high shuffle read/write, increased disk spill, and job latency that grows with data size. For a 12 GB transaction table joined with a 300 MB lookup table, the shuffle cost dwarfs the actual computation.

Enter broadcast variables

Spark broadcast variables let the driver send a read‑only dataset to every executor once, cache it in memory, and make it available to all tasks without per‑task shuffling. The variable is serialized (by default with Kryo) and sent across the network at job startup. Subsequent tasks reference the local copy, eliminating the need to move that data again.

Key points (from the research scope):

• Created on the driver and distributed once.
• Cached in executor memory and reused across all tasks.
• Size limit approximately 10 GB per variable; exceeding it may cause OutOfMemoryError.
• Serialization overhead exists; using Kryo for compact types reduces transmission cost.

Concrete example: broadcast join in Spark SQL

Suppose you have a Spark DataFrame orders (15 GB) and a small lookup customers (500 MB). To force a broadcast join, add the /*+ BROADCAST */ hint or set the configuration property.

-- Spark submit configuration
spark-submit \
  --conf spark.sql.autoBroadcastJoinThreshold=536870912 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  your-job.jar

-- Or within a PySpark session
from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .appName("broadcast-demo") \
    .config("spark.sql.autoBroadcastJoinThreshold", "536870912") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .getOrCreate()

# Assume orders and customers are already DataFrames
# Force the small DataFrame to be broadcast joined
result = orders.join(customers, "customer_id", "inner")

The spark.sql.autoBroadcastJoinThreshold defaults to 10 MB in Spark 3.x. Setting it to a higher value (or explicitly disabling with -1) lets you control when Spark attempts the optimization. In the example, the 500 MB customers table falls below the threshold, so Spark rewrites the logical plan to a broadcast join.

Verifying the improvement

After running the job, open the Spark UI (typically at http://:4040). Navigate to the 'Jobs' tab, click the relevant job, and look for the 'Broadcast' tab or the execution graph. Spark marks shuffle stages that were avoided; you’ll see a smaller number of reduce tasks and reduced shuffle read/write bytes.

You can also tune the block size to control how the broadcast variable is partitioned across executors:

-- Reduce memory pressure by splitting the broadcast into smaller blocks
spark-submit \
  --conf spark.broadcast.blockSize=134217728 \
  your-job.jar

Check executor logs for BroadcastBlockManager messages confirming the variable was fetched and cached.

Trade‑offs and limits

• Size ceiling: Broadcast variables larger than ~10 GB risk OutOfMemoryError on executors. If your dataset grows, reconsider the join strategy or partition the data.
• Dynamic data: Broadcast variables are snapshots. If the source data changes between jobs, you must re‑broadcast; they are not suitable for frequently updating tables.
• Serialization: Without Kryo registration for custom classes, Spark falls back to Java serialization, which is slower and produces larger payloads. Register types upfront or stick to built‑in SQL types.

Practical check: run a job with a dataset near the 10 GB boundary and monitor the executor memory tab in the UI. If you see MemoryPool OVERHEAP_ONLY spikes approaching the executor limit, the broadcast size is too large.

When to use it, when not to

Use a broadcast join when the small side is comfortably under the auto‑broadcast threshold (default 10 MB, adjustable up to ~10 GB) and the data is read‑only for the job’s duration. Avoid it when:

• The small side exceeds 10 GB and cannot be partitioned.
• The table is updated frequently between runs.
• You already have a well‑tuned shuffle join with appropriate partitioning.

Starting with the right threshold and a Kryo‑registered schema often turns a sluggish shuffle into a near‑instant broadcast join – with the only cost being a one‑time network transfer at job startup.