Sample Tables — Simple Data
Venn view of 7 JOIN types — refer to df1 (4 rows) / df2 (3 rows) sample.
All PySpark JOIN examples use these two small DataFrames. Visualize them before reading each Venn diagram.
df1 — employees
| id | name | dept_id |
|---|---|---|
| 1 | Alice | 10 |
| 2 | Bob | 20 |
| 3 | Charlie | 10 |
| 4 | David | NULL |
df2 — departments
| dept_id | dept_name |
|---|---|
| 10 | HR |
| 20 | Engineering |
| 30 | Marketing |
Spark Join Types
Spark supports several types of joins for combining DataFrames. Understanding when to use each type is crucial for both interviews and production code.
Inner Join
Returns only rows that have matching keys in both DataFrames.
df1.join(df2, df1.id == df2.id, "inner")Result Example:
Left Join (Left Outer)
Returns all rows from the left DataFrame, and matching rows from the right DataFrame. NULL for non-matching right rows.
df1.join(df2, df1.id == df2.id, "left")Result Example:
Right Join (Right Outer)
Returns all rows from the right DataFrame, and matching rows from the left DataFrame. NULL for non-matching left rows.
df1.join(df2, df1.id == df2.id, "right")Result Example:
Full Outer Join
Returns all rows from both DataFrames. NULL where there is no match.
df1.join(df2, df1.id == df2.id, "outer")Left Semi Join
Returns rows from the left DataFrame where a match exists in the right DataFrame, but returns only columns from the left side.
df1.join(df2, df1.id == df2.id, "left_semi")Expected Output (left_semi - only left columns, only matched):
Bob (id 2, no match) and David (id 4, NULL) excluded. Only left side columns returned.
Left Anti Join
Returns rows from the left DataFrame where no match exists in the right DataFrame.
df1.join(df2, df1.id == df2.id, "left_anti")Expected Output (left_anti - only unmatched left):
Alice and Charlie have matches, so excluded. Only Bob and David remain.
Broadcast Join
Broadcast join sends the smaller DataFrame to all executors so that the large DataFrame does not need to perform a large shuffle.
Key B
Key B
Key B
from pyspark.sql.functions import broadcast
# Broadcast the smaller DataFrame
result = large_df.join(
broadcast(small_df),
"key"
)Expected Output (same as inner join, but faster):
Result identical to INNER JOIN on sample tables (David excluded), but no shuffle of small_df.
Configuration:
# Set broadcast threshold (default 10MB)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10m")✅ Advantages
- Can avoid large shuffle
- Fast when the smaller DataFrame fits in executor memory
❌ Disadvantages
- Not suitable if the supposedly small DataFrame is too large
- Memory pressure can occur
💡 Interview Tip: Broadcast joins are the fastest join strategy but only work when one side is small enough to fit in memory.
Sort-Merge Join
Sort-Merge Join is commonly used when joining two large DataFrames. Both sides are partitioned, sorted, and then merged.
Process Steps:
- Both DataFrames are hash-partitioned on the join key
- Each partition is sorted
- Sorted partitions are merged
Expected Output (same as inner, via sort-merge):
Both DataFrames were partitioned and sorted on dept_id before merging. David (NULL) excluded.
✅ Advantages
- Suitable for large datasets
- Default join strategy for large-large joins
❌ Disadvantages
- Requires sorting
- Can require significant shuffling
- Sorting/shuffling can increase execution time
💡 Interview Tip: Sort-merge join is Spark's default for large datasets. Understanding its mechanics helps optimize join performance.
Join Optimization Tips
Use Broadcast when possible
Broadcast the smaller DataFrame to avoid shuffle.
from pyspark.sql.functions import broadcast
# Force broadcast join
result = large_df.join(broadcast(small_df), "key")Partition by join key
Pre-partition data to reduce shuffle.
# Repartition by join key before joining
df1_repartitioned = df1.repartition("join_key")
df2_repartitioned = df2.repartition("join_key")
result = df1_repartitioned.join(df2_repartitioned, "join_key")Handle data skew
Use salting for skewed join keys.
import pyspark.sql.functions as F
import random
# Add salt to handle skew
salt_buckets = 10
df1_salted = df1.withColumn(
"salted_key",
F.concat(F.col("key"), F.lit("_"), (F.rand() * salt_buckets).cast("int"))
)
# Explode the salted keys on the other side
df2_exploded = df2.withColumn(
"salt", F.explode(F.array([F.lit(i) for i in range(salt_buckets)]))
).withColumn(
"salted_key",
F.concat(F.col("key"), F.lit("_"), F.col("salt"))
)
result = df1_salted.join(df2_exploded, "salted_key")Use broadcast hint
Force broadcast join with hint.
# Using broadcast hint
result = df1.join(
df2.hint("broadcast"),
"key"
)
# Or using SQL hint
df1.createOrReplaceTempView("table1")
df2.createOrReplaceTempView("table2")
result = spark.sql("""
SELECT /*+ BROADCAST(table2) */ *
FROM table1
JOIN table2 ON table1.key = table2.key
""")Filter early
Reduce data before joining.
# Filter before join to reduce data size
df1_filtered = df1.filter(F.col("date") >= "2024-01-01")
df2_filtered = df2.filter(F.col("status") == "active")
result = df1_filtered.join(df2_filtered, "key")💡 Interview Tip: Always consider broadcast join first for small-large joins. It can improve performance by orders of magnitude.