Introduction

Spark DataFrame supports all basic SQL Join Types like INNER, LEFT OUTER, RIGHT OUTER, LEFT ANTI, LEFT SEMI, CROSS, SELF JOIN. Spark SQL Joins are wider transformations that result in data shuffling over the network, hence they have huge performance issues when not designed with care.

PySpark SQL Joins come with more optimization by default (thanks to DataFrames & Datasets) however, there are still some performance issues to consider while using them.

1. SQL Join Types & Syntax

Below are different Join Syntaxes for Java Spark.

1) Dataset<Row> join(Dataset<?> right);
2) Dataset<Row> join(Dataset<?> right, String usingColumn);
3) Dataset<Row> join(Dataset<?> right, Seq<String> usingColumns);
4) Dataset<Row> join(Dataset<?> right, Seq<String> usingColumns, String joinType);
5) Dataset<Row> join(Dataset<?> right, Column joinExprs);
6) Dataset<Row> join(Dataset<?> right, Column joinExprs, String joinType);

The rest of the tutorial explains Join Types using syntax 6, which takes arguments for the right join DataFrame, join expression, and the type of join in String.

For Syntax 4 & 5, you can use either “JoinType” or “Join String” defined in the table below for the “joinType” string argument. When you use “JoinType”, you should import org.apache.spark.sql.catalyst.plans._ as this package defines JoinType objects.

| JOINTYPE | JOIN STRING | EQUIVALENT SQL JOIN | | -------------- | ----------------------------------- | ------------------- | | Inner.sql | inner | INNER JOIN | | FullOuter.sql | outer, full, fullouter, full_outer | FULL OUTER JOIN | | LeftOuter.sql | left, leftouter, left_outer | LEFT JOIN | | RightOuter.sql | right, rightouter, right_outer | RIGHT JOIN | | Cross.sql | cross | | | LeftAnti.sql | anti, leftanti, left_anti | | | LeftSemi.sql | semi, leftsemi, left_semi | |

All Join objects are defined in the joinTypes class. To use these, import org.apache.spark.sql.catalyst.plans.{LeftOuter, Inner, ...}.

Before we jump into Spark SQL Join examples, first, let’s create emp and dept DataFrames. Here, column emp_id is unique in emp, and dept_id is unique in the dept dataset. emp_dept_id from emp has a reference to dept_id in the dept dataset.

List<Row> empData = Arrays.asList(
  RowFactory.create(1, "Smith", -1, "2018", "10", "M", 3000),
  RowFactory.create(2, "Rose", 1, "2010", "20", "M", 4000),
  RowFactory.create(3, "Williams", 1, "2010", "10", "M", 1000),
  RowFactory.create(4, "Jones", 2, "2005", "10", "F", 2000),
  RowFactory.create(5, "Brown", 2, "2010", "40", "", -1),
  RowFactory.create(6, "Brown", 2, "2010", "50", "", -1)
);
StructType empSchema = new StructType(new StructField[]{
  new StructField("emp_id", DataTypes.IntegerType, false, Metadata.empty()),
  new StructField("name", DataTypes.StringType, false, Metadata.empty()),
  new StructField("superior_emp_id", DataTypes.IntegerType, true, Metadata.empty()),
  new StructField("year_joined", DataTypes.StringType, false, Metadata.empty()),
  new StructField("emp_dept_id", DataTypes.StringType, false, Metadata.empty()),
  new StructField("gender", DataTypes.StringType, false, Metadata.empty()),
  new StructField("salary", DataTypes.IntegerType, false, Metadata.empty())
});
Dataset<Row> empDF = spark.createDataFrame(empData, empSchema);
empDF.show(false);

List<Row> deptData = Arrays.asList(
  RowFactory.create("Finance", 10),
  RowFactory.create("Marketing", 20),
  RowFactory.create("Sales", 30),
  RowFactory.create("IT", 40)
);
StructType deptSchema = new StructType(new StructField[]{
  new StructField("dept_name", DataTypes.StringType, false, Metadata.empty()),
  new StructField("dept_id", DataTypes.IntegerType, false, Metadata.empty())
});
Dataset<Row> deptDF = spark.createDataFrame(deptData, deptSchema);
deptDF.show(false);

This prints the “emp” and “dept” DataFrame to the console.

//Emp Dataset
+------+--------+---------------+-----------+-----------+------+------+
|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    |
|6     |Brown   |2              |2010       |50         |      |-1    |
+------+--------+---------------+-----------+-----------+------+------+

//Dept Dataset
+---------+-------+
|dept_name|dept_id|
+---------+-------+
|Finance  |10     |
|Marketing|20     |
|Sales    |30     |
|IT       |40     |
+---------+-------+

2. Inner Join

Spark Inner join is the default join and it’s mostly used. It is used to join two DataFrames/Datasets on key columns; where keys don’t match, the rows get dropped from both datasets (emp & dept).

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "inner")
  .show(false);

When we apply Inner join on our datasets, it drops “emp_dept_id” 50 from “emp” and “dept_id” 30 from “dept” datasets. Below is the result of the above Join expression.

+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|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 a.k.a full, fullouter join returns all rows from both Spark DataFrame/Datasets; where join expression doesn’t match, it returns null on respective record columns.

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "outer")
  .show(false);

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "full")
  .show(false);

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "fullouter")
  .show(false);

From our emp dataset, the emp_dept_id with a value of 50 doesn’t have a record in dept; hence, the dept columns have null. Similarly, the dept_id 30 doesn’t have a record in emp; hence, you see nulls on emp columns. Below is the result of the above Join expression.

+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|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 Left a.k.a Left Outer join returns all rows from the left DataFrame/Dataset regardless of whether a match is found on the right dataset. When join expression doesn’t match, it assigns null for that record and drops records from the right where the match is not found.

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "left")
  .show(false);

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "leftouter")
  .show(false);

From our dataset, emp_dept_id 50 doesn’t have a record in the dept dataset, hence, this record contains null on the dept columns (dept_name & dept_id). And dept_id 30 from the dept dataset is dropped from the results. Below is the result of the above Join expression.

+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|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 Right a.k.a Right Outer join is the opposite of left join. Here, it returns all rows from the right DataFrame/Dataset regardless of whether a match is found on the left dataset. When the join expression doesn’t match, it assigns null for that record and drops records from the left where the match is not found.

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "right")
  .show(false);

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "rightouter")
  .show(false);

From our example, the right dataset dept_id 30 doesn't have it on the left dataset emp; hence, this record contains null on emp columns. And emp_dept_id 50 is dropped as a match is not found on the left. Below is the result of the above Join expression.

+------+--------+---------------+-----------+-----------+------+------+---------+-------+
|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

Spark Left Semi join is similar to inner join, with the difference being leftsemi join returns all columns from the left DataFrame/Dataset and ignores all columns from the right dataset. In other words, this join returns columns from the only left dataset for the records matched in the right dataset on join expression, ignoring records not matched on join expression from both left and right datasets.

The same result can be achieved using select on the result of the inner join; however, using this join would be efficient.

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "leftsemi")
  .show(false);

Below is the result of the above join expression.

//leftsemi join
+------+--------+---------------+-----------+-----------+------+------+
|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 does the exact opposite of the Spark leftsemi join. leftanti join returns only columns from the left DataFrame/Dataset for non-matched records.

empDF.join(deptDF, empDF.col("emp_dept_id").equalTo(deptDF.col("dept_id")), "leftanti")
  .show(false);

Yields below output

+------+-----+---------------+-----------+-----------+------+------+
|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 complete without a self join. Though there is no self-join type available, we can use any of the above-explained join types to join DataFrame to itself. Below example uses inner self join

empDF.as("emp1").join(empDF.as("emp2"),
  col("emp1.superior_emp_id").equalTo(col("emp2.emp_id")), "inner")
  .select(
    col("emp1.emp_id"), 
    col("emp1.name"),
    col("emp2.emp_id").as("superior_emp_id"),
    col("emp2.name").as("superior_emp_name"))
  .show(false);

Here, we are joining emp dataset with itself to find out 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 explained here are available at the GitHub project for reference.