Introduction
PySpark DataFrame supports all basic SQL Join Types like INNER, LEFT OUTER, RIGHT OUTER, LEFT ANTI, LEFT SEMI, CROSS, and SELF JOIN. Spark SQL Joins are wider transformations that involve data shuffling over the network, which can lead to significant performance issues if not handled with care.
However, PySpark SQL Joins come with optimizations by default, especially due to the use of DataFrames and Datasets, although some performance issues may still need attention during their use.
1. SQL Join Types & Syntax
Below are different Join Syntaxes in PySpark:
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "inner")
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "outer")
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "left")
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "right")
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "leftsemi")
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "leftanti")
empDF.join(empDF.alias("emp2"), empDF.superior_emp_id == empDF.alias("emp2").emp_id, "inner")
Let's kick off by creating the emp and dept DataFrames. Here, the emp_id is unique in the emp dataset, dept_id is unique in the dept dataset, and emp_dept_id from emp references dept_id in dept.
emp_data = [(1, "Smith", -1, "2018", "10", "M", 3000),
(2, "Rose", 1, "2010", "20", "M", 4000),
(3, "Williams", 1, "2010", "10", "M", 1000),
(4, "Jones", 2, "2005", "10", "F", 2000),
(5, "Brown", 2, "2010", "40", "", -1),
(6, "Brown", 2, "2010", "50", "", -1)]
emp_columns = ["emp_id", "name", "superior_emp_id", "year_joined",
"emp_dept_id", "gender", "salary"]
empDF = spark.createDataFrame(emp_data, emp_columns)
empDF.show(truncate=False)
dept_data = [("Finance", 10),
("Marketing", 20),
("Sales", 30),
("IT", 40)]
dept_columns = ["dept_name", "dept_id"]
deptDF = spark.createDataFrame(dept_data, dept_columns)
deptDF.show(truncate=False)
2. Inner Join
Spark Inner join, the default join type, is primarily used to join two DataFrames/Datasets on key columns while dropping unmatched rows from both datasets (emp & dept).
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "inner").show(truncate=False)
Applying an inner join on our datasets excludes emp_dept_id 50 from emp and dept_id 30 from dept datasets. Here's the result:
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|1 |Smith |-1 |2018 |10 |M |3000 |Finance |10 |
|2 |Rose |1 |2010 |20 |M |4000 |Marketing|20 |
|3 |Williams|1 |2010 |10 |M |1000 |Finance |10 |
|4 |Jones |2 |2005 |10 |F |2000 |Finance |10 |
|5 |Brown |2 |2010 |40 | |-1 |IT |40 |
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
3. Full Outer Join
Outer, also known as full, or fullouter join, returns all rows from both DataFrames/Datasets, where unmatched expressions result in null on respective records.
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "outer").show(truncate=False)
In our emp dataset, emp_dept_id with value 50 has no record in dept, resulting in null for dept columns. Likewise, dept_id 30 has no match in emp, resulting in nulls in emp columns. Here's the output:
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|2 |Rose |1 |2010 |20 |M |4000 |Marketing|20 |
|5 |Brown |2 |2010 |40 | |-1 |IT |40 |
|1 |Smith |-1 |2018 |10 |M |3000 |Finance |10 |
|3 |Williams|1 |2010 |10 |M |1000 |Finance |10 |
|4 |Jones |2 |2005 |10 |F |2000 |Finance |10 |
|6 |Brown |2 |2010 |50 | |-1 |null |null |
|null |null |null |null |null |null |null |Sales |30 |
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
4. Left Outer Join
Spark's Left, or Left Outer join, returns all rows from the left DataFrame/Dataset. If no match is found on the right dataset, it sets null for that record and drops unmatched records from the right.
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "left").show(truncate=False)
In this dataset, emp_dept_id 50 has no match in dept, resulting in null values for dept columns (dept_name & dept_id). Also, dept_id 30 from dept is not included. Here's the data:
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|1 |Smith |-1 |2018 |10 |M |3000 |Finance |10 |
|2 |Rose |1 |2010 |20 |M |4000 |Marketing|20 |
|3 |Williams|1 |2010 |10 |M |1000 |Finance |10 |
|4 |Jones |2 |2005 |10 |F |2000 |Finance |10 |
|5 |Brown |2 |2010 |40 | |-1 |IT |40 |
|6 |Brown |2 |2010 |50 | |-1 |null |null |
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
5. Right Outer Join
Spark's Right, or Right Outer join, is the reverse of the left join. Here, all rows from the right DataFrame/Dataset are included, regardless of matches in the left dataset. Records with no matches receive null values, while those without matches in the left dataset are excluded.
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "right").show(truncate=False)
For example, dept_id 30 on the right does not match any records on the left (emp), resulting in null values for emp columns, and emp_dept_id 50 is omitted due to no left match. See below:
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|4 |Jones |2 |2005 |10 |F |2000 |Finance |10 |
|3 |Williams|1 |2010 |10 |M |1000 |Finance |10 |
|1 |Smith |-1 |2018 |10 |M |3000 |Finance |10 |
|2 |Rose |1 |2010 |20 |M |4000 |Marketing|20 |
|null |null |null |null |null |null |null |Sales |30 |
|5 |Brown |2 |2010 |40 | |-1 |IT |40 |
+------+--------+---------------+-----------+-----------+------+------+---------+-------+
6. Left Semi Join
The Left Semi join in Spark resembles the inner join, with the distinction that leftsemi joins return all columns from the left DataFrame/Dataset, ignoring all columns from the right. This join selects columns from the left-only for matching records in the right. Unmatched records are ignored from both sides.
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "leftsemi").show(truncate=False)
Resulting data is shown below:
+------+--------+---------------+-----------+-----------+------+------+
|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|
+------+--------+---------------+-----------+-----------+------+------+
|1 |Smith |-1 |2018 |10 |M |3000 |
|2 |Rose |1 |2010 |20 |M |4000 |
|3 |Williams|1 |2010 |10 |M |1000 |
|4 |Jones |2 |2005 |10 |F |2000 |
|5 |Brown |2 |2010 |40 | |-1 |
+------+--------+---------------+-----------+-----------+------+------+
7. Left Anti Join
Left Anti join is the mirror opposite of the leftsemi join. It returns only columns from the left DataFrame/Dataset for unmatched records.
empDF.join(deptDF, empDF.emp_dept_id == deptDF.dept_id, "leftanti").show(truncate=False)
Here's the result:
+------+-----+---------------+-----------+-----------+------+------+
|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|
+------+-----+---------------+-----------+-----------+------+------+
|6 |Brown|2 |2010 |50 | |-1 |
+------+-----+---------------+-----------+-----------+------+------+
8. Self Join
Spark Joins are not considered comprehensive without incorporating a self join. While there's no specific self-join type, any previously detailed join types can be employed to join a DataFrame with itself. Let's use an inner self join as an example.
from pyspark.sql import functions as F
empDF.alias("emp1").join(
empDF.alias("emp2"), F.col("emp1.superior_emp_id") == F.col("emp2.emp_id"), "inner"
).select(
F.col("emp1.emp_id"),
F.col("emp1.name"),
F.col("emp2.emp_id").alias("superior_emp_id"),
F.col("emp2.name").alias("superior_emp_name")
).show(truncate=False)
This demonstrates a self join using emp DataFrame to identify the superior emp_id and name for all employees.
+------+--------+---------------+-----------------+
|emp_id|name |superior_emp_id|superior_emp_name|
+------+--------+---------------+-----------------+
|2 |Rose |1 |Smith |
|3 |Williams|1 |Smith |
|4 |Jones |2 |Rose |
|5 |Brown |2 |Rose |
|6 |Brown |2 |Rose |
+------+--------+---------------+-----------------+
Examples detailed here are available at the GitHub Repository for reference.