Join Strategies

Overview

Learn how Spark executes joins under the hood using Sort-Merge, Broadcast Hash, Shuffle Hash, and Nested Loop joins.

What You Will Learn

Detailed Concept Explanation

When you join two DataFrames, Spark select one of several physical Join Strategies based on table sizes, join conditions, and whether keys are sortable.

1. Broadcast Hash Join (BHJ)

• Use Case: One table is small (under 10MB).
• Behavior: Broadcasts the small table to all executors, avoiding network shuffles.
• Performance: Fastest strategy.

2. Sort-Merge Join (SMJ)

• Use Case: Default strategy for joining two large tables.
• Behavior: Shuffles both tables by join keys so matching keys end up in the same partitions, sorts the data in each partition, and merges them.
• Performance: High disk and network I/O overhead due to shuffles.

3. Shuffle Hash Join

• Use Case: Joining large tables where data cannot be sorted, but partitions fit in memory.
• Behavior: Shuffles data and builds a local hash table on each partition.
• Performance: Faster than SMJ when sorting is expensive.

Code Examples

Python (PySpark) Implementation

from pyspark.sql import SparkSession

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

# Force Sort-Merge Join by disabling automatic broadcasting
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

df1 = spark.read.parquet("sales.parquet")
df2 = spark.read.parquet("users.parquet")

joined = df1.join(df2, "userId")
joined.explain()

Expected Output

== Physical Plan ==
+- SortMergeJoin [userId#0], [userId#5], Inner
   :- Sort [userId#0 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(userId#0, 200)
...

Execution Plan Diagram (Python & Scala)

Scala Implementation

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("JoinStrategiesScala").getOrCreate()

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

val df1 = spark.read.parquet("sales.parquet")
val df2 = spark.read.parquet("users.parquet")

val joined = df1.join(df2, "userId")
joined.explain()

Related Topics

• Broadcast Join
• Shuffle
• Joins