PySpark DataFrame Joins

1. The Anatomy of a PySpark Join

In PySpark, joins are performed using the .join() method on a DataFrame. The basic syntax is:

left_df.join(right_df, join_condition, how="join_type")

Let's break this down:

• right_df: The DataFrame you want to join with.
• join_condition: An expression that tells Spark how to match rows. This is typically a comparison of columns from the two DataFrames (e.g., left_df.id == right_df.id).
• how: A string specifying the type of join. If you omit this, PySpark performs an inner join by default.

To get a quick overview of the available join types and their syntax, let's consult a reference guide.

2. Visualizing the Core Join Types

Before we start coding, it's helpful to have a clear mental model of what each join does. The four most common types are inner, left outer, right outer, and full outer.

The Venn diagrams in the image below provide a classic and effective way to visualize how these joins combine two tables.

• Inner Join: Returns only the rows that have matching keys in both tables.
• Left Join: Returns all rows from the left table, and the matched rows from the right table. If there's no match, the columns from the right table will be null.
• Right Join: Returns all rows from the right table, and the matched rows from the left table. If there's no match, the columns from the left table will be null.
• Full Outer Join: Returns all rows from both tables. It places nulls where matches don't exist on either side.

3. Implementing Joins in PySpark

Now, let's see these concepts in action. The following video provides a practical, step-by-step walkthrough of performing different joins in a Databricks notebook.

Join Types Summary

Let's recap the syntax for the joins you just saw.

# Assume we have empDF and deptDF DataFrames




# Inner Join (the default)
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "inner").show()




# Left Outer Join
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "left_outer").show()




# Right Outer Join
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "right_outer").show()




# Full Outer Join
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "full_outer").show()




# Left Semi Join (returns columns from left DF only)
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "left_semi").show()




# Left Anti Join (returns columns from left DF only)
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "left_anti").show()

The Left Semi and Left Anti joins are particularly useful. They act as filters rather than combiners.

• Semi join is equivalent to a WHERE EXISTS subquery in SQL: "Give me all employees for whom a department exists."
• Anti join is equivalent to a WHERE NOT EXISTS subquery: "Give me all employees for whom a department does not exist."

These are excellent for data quality checks or for filtering one list based on another without needing to carry along any columns from the second list.

4. Hands-On Exercise

Time to apply what you've learned. In your Databricks notebook, create the following two DataFrames.

from pyspark.sql.functions import col




# Data for employees
employees_data = [
    (1, "Alice", 101),
    (2, "Bob", 102),
    (3, "Charlie", 101),
    (4, "David", 103),
    (5, "Eve", 104)  # Eve is in a department not listed in the departments table
]
employees_columns = ["emp_id", "emp_name", "dept_id"]
employees_df = spark.createDataFrame(data=employees_data, schema=employees_columns)

print("Employees DataFrame:")
employees_df.show()




# Data for departments
departments_data = [
    (101, "Engineering"),
    (102, "Marketing"),
    (105, "Finance")  # Finance department has no employees
]
departments_columns = ["dept_id", "dept_name"]
departments_df = spark.createDataFrame(data=departments_data, schema=departments_columns)

print("Departments DataFrame:")
departments_df.show()

Your Tasks:

1. Find all employees and their department names. The result should only include employees who belong to a department that exists in the departments_df.
2. Create a list of all employees, including their department name if available. If an employee's department is not in the departments_df, the department name should be null.
3. Find all departments that have no employees.
4. Identify any "orphan" employees. These are employees assigned to a dept_id that does not exist in the departments_df.

Solution Walkthrough:

Try to solve the tasks on your own first.

# Task 1: Inner join to find employees with a valid department
print("\nTask 1: Employees with their departments (Inner Join)")
employees_df.join(departments_df, employees_df.dept_id == departments_df.dept_id, "inner") \
    .select("emp_id", "emp_name", "dept_name") \
    .show()




# Task 2: Left join to list all employees
print("\nTask 2: All employees and their departments (Left Join)")
employees_df.join(departments_df, employees_df.dept_id == departments_df.dept_id, "left") \
    .select("emp_id", "emp_name", "dept_name") \
    .show()




# Task 3: Find departments with no employees. A right join is a good way to do this.
# We look for departments where the employee information is null.
print("\nTask 3: Departments with no employees (Right Join + Filter)")
departments_df.join(employees_df, employees_df.dept_id == departments_df.dept_id, "right") \
    .where(col("emp_id").isNull()) \
    .select("dept_name") \
    .show()




# An alternative for Task 3 using a left_anti join after swapping the dataframes.
print("\nTask 3 (Alternative): Departments with no employees (Left Anti Join)")
departments_df.join(employees_df, departments_df.dept_id == employees_df.dept_id, "left_anti").show()





# Task 4: Find orphan employees
print("\nTask 4: Employees with a non-existent department (Left Anti Join)")
employees_df.join(departments_df, employees_df.dept_id == departments_df.dept_id, "left_anti").show()

Conclusion

Excellent work! You can now effectively combine data from multiple sources using the full range of join operations available in PySpark. This is a fundamental skill for any data engineer, allowing you to build rich, consolidated datasets for analysis and processing.

Key Takeaways

• The .join() method is used to combine two DataFrames. The syntax is df1.join(df2, condition, how).
• The how parameter specifies the join type: inner, left, right, full, left_semi, and left_anti are the most common.
• Inner, Left, Right, and Full joins combine columns from both DataFrames based on the matching logic.
• Left Semi and Left Anti joins act as filters, returning only columns from the left DataFrame based on the presence or absence of a match in the right DataFrame.
• Your existing knowledge of SQL joins directly translates to PySpark, with the main difference being the method-based syntax.

Next Steps

Now that you can read, transform, aggregate, and join data to create a polished final dataset, the next logical step is to save it. In our next lesson, you will learn how to write DataFrames to Delta tables using various save modes, a crucial step for building reliable data pipelines on the Databricks Lakehouse Platform.