PySpark Joins

Spark tutorial · PySpark.in

PySpark Joins

PySpark offers a variety of join operations to combine DataFrames similar to SQL joins. In this tutorial, we present six chapters, each focusing on a type of join, with clear explanations, code examples, output illustrations, use cases, pitfalls, and best practices. These chapters are aimed at beginner to intermediate Spark developers and use a consistent example scenario for clarity.

1. Inner Join

An Inner Join returns only the rows that have matching keys in both DataFrames. If a key value is present in one DataFrame but not the other, that row will be omitted from the result. This is the default join type in PySpark (if no how is specified). Inner joins are useful when you need to combine data from two sources and only care about the records that exist in both.

Example and Code

Let's consider two DataFrames: Employees and Departments. The Employees table has a foreign key dept_id indicating the department each employee belongs to, which links to the Departments table. Some employees might not have a department, and some departments might have no employees, which will help illustrate join behavior (including how nulls are handled).

First, we create the example DataFrames and show their contents:

```

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("JoinsTutorial").getOrCreate()

# Create Employees DataFrame

employees = [

(1, "Alice", 101),

(2, "Bob", 102),

(3, "Charlie", None), # Charlie has no dept (null)

(4, "David", 104), # David's dept_id 104 doesn't exist in Departments

(5, "Eve", 101)

]

df_employees = spark.createDataFrame(employees, ["emp_id", "name", "dept_id"])

df_employees.show()

# Create Departments DataFrame

departments = [

(101, "HR"),

(102, "IT"),

(103, "Marketing")

]

df_departments = spark.createDataFrame(departments, ["dept_id", "dept_name"])

df_departments.show()

```

Now, perform an inner join on the dept_id key to combine employees with their department names:

```

# Inner join employees with departments on dept_id

inner_join_df = df_employees.join(df_departments, on="dept_id", how="inner")

inner_join_df.show()

```

The result includes only the rows where the dept_id appeared in both DataFrames. Specifically:

Notice that department HR (101) had two employees (Alice and Eve), and both rows appear in the joined output. In general, if keys are not unique, an inner join will produce multiple combined rows for each pair of matches (e.g. if one department has many employees, or an employee key appears multiple times in the other table). This can lead to duplicate or multiplied rows, which is expected behavior but something to be aware of.

Use Cases

Potential Pitfalls

Best Practices

2. Left, Right, and Full Joins

Outer joins include rows that do not have matches in the other DataFrame, padding with nulls where data is missing. PySpark supports left outer, right outer, and full outer joins (also called full join). These let you preserve data from one or both sides that would be lost in an inner join, at the cost of introducing nulls for missing fields. We discuss each:

Left Outer Join

A Left Join (left outer join) returns all rows from the left DataFrame, and the matching rows from the right DataFrame. If there are rows in the left DataFrame that have no match in the right, they still appear in the result, with nulls for the right side's columns. Essentially, a left join keeps everything from the left and adds data from the right when available.

Using our example data, suppose we want to list all employees and add their department names when available. We can do:

```

#Please execute Create dataframe code to execute this code

left_join_df = df_employees.join(df_departments, on="dept_id", how="left")

left_join_df.show()

```

Key observations for the left join result:

In summary, df_employees.left_outer_join(df_departments) yields a DataFrame of all employees, annotated with department info when a match exists, or null for department fields if not.

Use cases: Left joins are common when one dataset is primary and you want to enrich it with info from a secondary dataset if available. For example, retrieving all customers and their latest order (if they have one), or as in our case, all employees with their department if assigned.

Pitfalls: If many left-side rows have no match, the result will contain many nulls on the right side. You may need to handle or filter out those nulls after the join if they are not desired (e.g., using .filter or filling nulls with default values). Also, be mindful that left join can still multiply rows if the right side has multiple matches for a key (though in well-designed schemas that shouldn't happen for one-to-many relationships).

Best Practices: For left joins on large data, ensure the left DataFrame is the one you want to preserve entirely. If the right side is smaller, consider broadcasting it. Also, consider using left semi join (discussed later) if you only need to check existence of a match rather than retrieving columns from the right side.

Right Outer Join

A Right Join is the mirror image of a left join: it returns all rows from the right DataFrame, and adds matching rows from the left. Non-matching rows from the left are filled with nulls. In other words, it keeps everything on the right.

If we reverse our perspective and want to list all departments and any employees in them (if present), we use a right join:

```

right_join_df = df_employees.join(df_departments, on="dept_id", how="right")

right_join_df.show()

```

Observations:

Use cases: Right joins are less common, but useful when the right-hand dataset is the primary one. For example, you have a master list of all possible categories (right side) and want to show all categories with data from the left side if present. In scenarios where you want "all items from B with any matches from A," a right join is appropriate (or you can simply swap the order of DataFrames and do a left join).

Pitfalls: Similar to left join, expect nulls for unmatched entries and possible duplicate rows if the left side has multiple matches for a right-side key. If the left dataset is very large and the right is small, a right join may be inefficient — it's usually better to swap and use left join (since Spark doesn't support a direct "right semi/anti join", you typically swap datasets to achieve the equivalent).

Best Practices: Ensure that a right join is actually needed; often you can rewrite a right join as a left join by swapping the DataFrames. Use right join when it semantically makes sense (for readability of your code). Like left join, consider broadcasting the larger side appropriately, and handle nulls resulting from no matches.

Full Outer Join

A Full Outer Join (or just Full Join) returns all rows from both DataFrames, whether or not there's a match. Where a match is missing on either side, PySpark will fill in nulls for that side's columns. Full joins are useful for combining two datasets completely while identifying which data came from only one side.

Continuing our example, a full outer join on dept_id would include every employee and every department:

```

full_join_df = df_employees.join(df_departments, on="dept_id", how="outer")

full_join_df.show()

```

The full join result has 6 rows, representing the union of left and right sets:

Use cases: Full outer joins are useful in data reconciliation, merging two sources to find all records from both. For example, combining two lists of users to see all unique users and attributes from either source, or merging yearly datasets where you want all entries from both years, marking those that are new or dropped. It’s also used when you need to identify differences between datasets (the unmatched rows on each side).

Pitfalls: Full joins can produce a lot of nulls and potentially a very large result (size can approach the sum of both inputs). They also incur the cost of shuffling both datasets. If either side is huge, a full join can be expensive. Also, handling the null values post-join (e.g., deciding how to treat rows that exist only on one side) adds complexity. Be mindful of keys that appear in one side only – after a full join you might want to coalesce fields or tag records as coming from one source.

Best Practices: Only use full join when you truly need all data from both sides. Often, an analysis can be done with left or right joins or separate processing of unmatched records. If you do use a full join, be prepared to handle nulls (for example, using coalesce(colA, colB) if the same information might come from either side). Also consider splitting the problem: sometimes performing an inner join plus separate anti joins (to collect non-matches from each side) and then unioning results can give more control. Spark's optimizer handles full join in one step, but it might be less efficient than treating matches and mismatches separately if you need to process them differently.

3. Cross Join

A Cross Join returns the Cartesian product of two DataFrames. This means every row from DataFrame A is paired with every row from DataFrame B, without any matching condition. The number of output rows will be M × N (where M and N are the number of rows in A and B respectively), which can grow extremely large. Cross joins are equivalent to an SQL cartesian product; they are rarely needed in typical analytics but can be useful in specific scenarios (like generating combinations or test data).

PySpark requires an explicit call or option to perform a cross join, as it can be computationally expensive. You can use df.crossJoin(other_df) or specify how="cross" in the join function. If you attempt to join without a condition (and without how="cross"), Spark may throw an analysis error unless cross joins are enabled in the configuration (to prevent accidental cartesian products).

Example and Code

For a simple illustration, let's generate all combinations of employee names and department names (ignoring the actual dept_id relationships). We can do this by selecting just the name columns and performing a cross join:

```

# Select only name columns and perform a cross join

names_df = df_employees.select("name")

deptnames_df = df_departments.select("dept_name")

cross_df = names_df.crossJoin(deptnames_df)

cross_df.show(6) # show first 6 results for brevity

print("Total rows in cross join result:", cross_df.count())

```

As expected, each of the 5 employee names is paired with each of the 3 department names, for a total of 5×3 = 15 combinations. The snippet above shows a few examples: Alice with HR, Alice with IT, Alice with Marketing, then Bob with each department, etc.

Use Cases

Pitfalls

Best Practices

4. Semi and Anti Join

PySpark supports semi and anti joins, which are handy for filtering based on existence of matches, without bringing in all columns from the other DataFrame. These correspond to the SQL concepts of semijoin (EXISTS) and antijoin (NOT EXISTS).

Spark only implements left semi/anti joins (not right semi/anti), but right-semi/anti logic can be achieved by swapping datasets.

Left Semi Join Example

Let's find all employees who have a valid department (i.e., their dept_id is found in the Departments table). We don't necessarily need the department details, just to know which employees are associated with any department.

```

semi_join_df = df_employees.join(df_departments, on="dept_id", how="left_semi")

semi_join_df.show()

```

We get three employees: Alice, Bob, and Eve. These are exactly the ones we saw in the inner join earlier, because they had matching departments. Charlie and David are excluded, since they did not have a matching dept_id in Departments. Notice that the output contains only the columns from df_employees (emp_id, name, dept_id); the department name is not included, even though the join condition checked against df_departments. This is the key feature of a semi join – it's like a filter on the left DataFrame by existence in the right.

Using a left semi join in this way is more efficient than doing a full inner join and then discarding the right side columns, because Spark can optimize it to just do an existence check instead of carrying all data from the right side.

Use cases: Left semi joins are great for filtering. For example, finding users that have made a purchase by semi joining the users table with the purchases table (on user_id). It yields the users who exist in purchases, without bringing in purchase details. It’s essentially doing df_left WHERE key IN (SELECT key FROM df_right) in one step.

Pitfalls: Since no columns from the right side are available, you can’t directly use this join if you actually need those columns in the output. In such cases, an inner or left join is necessary. Also, remember Spark only has left semi (no direct right semi), so if you need "all right rows that have a match in left", you would swap the tables or use a different approach.

Best Practices: Use semi joins for existence filtering because they are usually more efficient than an inner join followed by dropping columns. They communicate intent clearly (someone reading the code sees it's just checking membership). If you need to perform an existence check and then maybe retrieve the matching data from the right side, consider doing the semi join to filter the left side, then an inner join to get the details (this can reduce the data size going into the inner join).

Left Anti Join Example

Now, let's find employees who do not have any department in the Departments table. This will give us the inverse: which employees are not associated with a known department.

```

anti_join_df = df_employees.join(df_departments, on="dept_id", how="left_anti")

anti_join_df.show()

```

The result lists Charlie and David – exactly those employees we expected, since Charlie had no dept_id (null) and David had 104 which wasn't in Departments. These are the employees who have no matching dept_id in the Departments dataset. The output again includes only columns from the left side (employees).

Use cases: Left anti joins are very useful for finding discrepancies or missing entries. For example, finding records in one table that aren't present in another (customers who never made a purchase, items that are not in the catalog, etc.). It’s essentially df_left WHERE key NOT IN (SELECT key FROM df_right).

Pitfalls: As with semi join, no columns from the right side are available (since by definition these rows have no match on the right). If you actually need to see what they were not matched with, you'd have to do additional steps (but usually the point is they didn't match anything). There’s also no right anti join directly; to get “rows in right that have no match in left,” you can swap the inputs or use a full outer join and filter.

Best Practices: Use anti join for anti-pattern detection (finding missing keys) as it's clearer and often more efficient than doing a full outer join and filtering nulls. If performance is an issue and data is huge, sometimes collecting the small side's keys and broadcasting them to filter the large side can be an alternative (if semi/anti join performance isn't optimal by itself), but Spark's built-in semi/anti join is usually quite good.

Additional Tip: Spark does not support right semi or right anti join directly. If you ever need the equivalent of those, you can simply flip the DataFrames. For example, df_right.join(df_left, ..., "left_semi") would effectively give you the "right semi" (rows from df_right that have match in df_left).

More Spark tutorials

All tutorials · Try the free PySpark compiler · Practice challenges