Broadcast Join

Overview

Learn how Broadcast Joins (Broadcast Hash Joins) eliminate the expensive shuffle step when joining large datasets with small lookup tables.

What You Will Learn

In this lesson, you will learn:

Broadcast Hash Join: The mechanics of replication-based joins.
Trigger Thresholds: Configuring table size limits (default 10MB).
Syntax API: Forcing broadcast joins using the broadcast() function wrapper.

Detailed Concept Explanation

Joining two large tables usually requires a Sort-Merge Join, which shuffles all rows with matching join keys to the same worker node over the network. Shuffling is highly resource-intensive.

If one of the tables is small (e.g. a lookup table under 10MB), you can use a Broadcast Hash Join (BHJ) instead.

text

Driver ---> Sends Small Table to ALL Executors ---> Executor A (Large Part join Local Small)
                                               ---> Executor B (Large Part join Local Small)

How it Works

Spark sends (broadcasts) the entire small table to every executor node.
Each executor loads the small table into memory as a hash table.
The executor joins its partition of the large table against the local hash table, performing the join in memory with zero network shuffles.

Code Examples

Input Dataset Preview

Below is the lookup table depts:

deptId	name
1	HR
2	IT

Python (PySpark) Implementation

python

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

spark = SparkSession.builder.appName("BroadcastJoin").getOrCreate()

large_df = spark.read.parquet("sales.parquet")
depts_df = spark.createDataFrame([(1, "HR"), (2, "IT")], ["deptId", "name"])

# Force a Broadcast Hash Join
joined_df = large_df.join(broadcast(depts_df), "deptId")
joined_df.explain()

Expected Output

text

== Physical Plan ==
AdaptiveSparkPlan isAdaptive=true
+- BroadcastHashJoin [deptId#0], [deptId#4], Inner, BuildRight
   :- Scan parquet [deptId#0,amount#1]
   +- BroadcastExchange
      +- LocalTableScan [deptId#4,name#5]

Execution Plan Diagram (Python & Scala)

Execution Plan Diagram

SparkSession.builder

read.parquet

createDataFrame(depts)

join(broadcast(depts_df))

explain()

Scala Implementation

scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.broadcast

val spark = SparkSession.builder().appName("BroadcastJoinScala").getOrCreate()
import spark.implicits._

val largeDF = spark.read.parquet("sales.parquet")
val deptsDF = Seq((1, "HR"), (2, "IT")).toDF("deptId", "name")

val joined = largeDF.join(broadcast(deptsDF), "deptId")
joined.explain()

Interview Perspective

What are the requirements and default threshold for a Broadcast Hash Join?

The default size threshold for broadcasting a table in Spark is 10MB, configured via spark.sql.autoBroadcastJoinThreshold. The small table must fit inside the driver node and executor JVM memory allocations. If the table is larger than available memory, broadcasting it will trigger an Out Of Memory (OOM) error.