Joins in PySpark dataframe

 In PySpark, joining DataFrames is very similar to SQL joins, but you use the join() method on a DataFrame. Here’s a clear breakdown:

🔑 Types of Joins in PySpark

• Inner Join: Returns rows with matching keys in both DataFrames.

• Left Outer Join: Returns all rows from the left DataFrame, and matching rows from the right.

• Right Outer Join: Returns all rows from the right DataFrame, and matching rows from the left.

• Full Outer Join: Returns all rows when there is a match in either DataFrame.

• Cross Join: Cartesian product of both DataFrames.

• Semi Join: Returns rows from the left DataFrame where keys exist in the right DataFrame.

• Anti Join: Returns rows from the left DataFrame where keys do not exist in the right DataFrame.

⚙️ Syntax Example

python

# Assume df1 and df2 are DataFrames
# Join on column "id"

# Inner Join
result = df1.join(df2, df1.id == df2.id, "inner")

# Left Outer Join
result = df1.join(df2, df1.id == df2.id, "left")

# Right Outer Join
result = df1.join(df2, df1.id == df2.id, "right")

# Full Outer Join
result = df1.join(df2, df1.id == df2.id, "outer")

📌 Notes

• You can join on multiple columns:

  python

  result = df1.join(df2, (df1.id == df2.id) & (df1.name == df2.name), "inner")
  

• If the column names are the same in both DataFrames, you can pass a string instead of an expression:

  python

  result = df1.join(df2, "id", "inner")
  

• After joining, you may want to drop duplicate columns using .drop() or rename them with .withColumnRenamed().

👉 Do you want me to show you a practical example with sample data (like two small DataFrames and their join results), or a real-world use case (e.g., joining customer and order tables)?

Interview Questions

Q1: What are the exact conditions required for Spark to execute a Broadcast Hash Join (BHJ)? Can it fail?

Answer: For Spark to execute a Broadcast Hash Join, the following conditions must be met:

• At least one of the DataFrames must be smaller than the threshold defined by spark.sql.autoBroadcastJoinThreshold (default is 10MB), or explicitly wrapped in the broadcast() hint.
• The join type must support broadcasting the specific table (e.g., you cannot broadcast the left table in a Left Outer Join).
• The executor memory on the driver and worker nodes must be large enough to hold the broadcasted dataset.

Can it fail? Yes. If you force a broadcast hint on a DataFrame that turns out to be larger than the available driver memory or executor container memory, it will throw an Out Of Memory (OOM) error (java.lang.OutOfMemoryError: Java heap space). This happens because the driver must collect the data first before broadcasting it to the workers.

Q2: How does Adaptive Query Execution (AQE) dynamically optimize joins during runtime?

Answer: AQE introduces a feedback loop that optimizes query plans based on runtime statistics gathered during the shuffle stages. For joins, it does three key things:

• Dynamically Coalesces Shuffle Partitions: It merges small, post-shuffle partitions to prevent CPU overhead from running thousands of tiny tasks.
• Dynamically Switches Join Strategies: If Spark initially planned a Sort-Merge Join but discovers after filtering that one table's actual size is well below the broadcast threshold, AQE converts the plan to a Broadcast Hash Join on the fly.
• Dynamically Handles Skew Joins: If it detects a highly skewed partition (a partition significantly larger than the median size), it automatically splits that partition into smaller sub-partitions and matches them against the corresponding data from the other side.

Q3: What is a Broadcast Nested Loop Join (BNLJ), and when does Spark fallback to it?

Answer: A Broadcast Nested Loop Join is one of Spark's least efficient join strategies. It acts like a nested for loop across distributed data: for every row in one dataset, it scans the entire other dataset.

• When it occurs: Spark falls back to a BNLJ when you attempt to join two DataFrames using a non-equi join condition (e.g., using operators like <, >, or != instead of ==) and one of the tables is small enough to be broadcasted. If neither table fits in memory, it results in a highly expensive Cartesian Product (crossJoin).

---

Programmatic & Real-World Scenarios

Q4: How do you perform a join using multiple conditions and non-equi operators in PySpark?

Answer: You can combine multiple join columns and conditional logic by using the logical operators & (AND) and | (OR) enclosed in parenthesis within the join expression:

python

from pyspark.sql.functions import col

# Joining on an exact key AND a date range constraint (Non-equi join)
joined_df = df_orders.join(
    df_promotions,
    (col("df_orders.store_id") == col("df_promotions.store_id")) & 
    (col("df_orders.order_date").between(col("df_promotions.start_date"), col("df_promotions.end_date"))),
    "inner"
)

Use code with caution.

Q5: If you are joining two massive tables that are already Bucket-Sorted, how does Spark behave?

Answer: If both tables are bucketed on the exact same join columns, with the same number of buckets, and pre-sorted, Spark can completely skip the Shuffle and Sort phases of a Sort-Merge Join. It will execute a Bucket-Aware Join, reading the matching files directly into memory and merging them locally. This yields massive performance improvements for highly repetitive enterprise workloads.

---

Execution Plan Diagnosis

Q6: How do you identify what type of join Spark is executing under the hood?

Answer: You can inspect the physical execution plan by calling the .explain() method on the resulting DataFrame:

python

joined_df.explain()

Use code with caution.

Look at the text output from top to bottom. You will see specific physical operators denoting the strategy:

• BroadcastHashJoin or BHJ
• SortMergeJoin or SMJ
• BroadcastNestedLoopJoin or BNLJ
• ShuffleHashJoin

---

Join Strategy Selection Matrix

Join Strategy	Shuffle Required?	Memory Requirements	Best Used For
Broadcast Hash Join (BHJ)	No ❌	High on Executor (fits small table)	Big table 🤝 Small table
Sort-Merge Join (SMJ)	Yes 🔄	Low (spills to disk if needed)	Big table 🤝 Big table
Shuffle Hash Join	Yes 🔄	Medium (fits partition hash table)	Big table 🤝 Medium table
Broadcast Nested Loop (BNLJ)	No ❌	High on Executor	Non-equi joins (<, >, !=)