Image by Author | Canva

Joining various datasets into one dataset is something you’ll often have to do in data analysis and engineering workflows. In PySpark, this joining takes the form of joining DataFrames.

In the following 1,000 words or so, I will cover all the information you need to join DataFrames efficiently in PySpark. This will include explanations of what PySpark and DataFrames are before I explain all the possible join types, their syntax, and examples.

What is PySpark?

PySpark is the Python API for Apache Spark. Apache Spark is, thanks for asking, a big data analytics engine for both batch data processing and live data streaming.

With PySpark, you can perform big data processing, (real-time) analytics, machine learning tasks, and data integration.

This is very useful if you’re a data engineer or a data scientist working with big data.

What are DataFrames in PySpark?

As per the official Apache Spark documentation definition, DataFrames are datasets “organized into named columns…conceptually equivalent to a table in a relational database or a data frame in R/Python”. Compared to relational databases and R/Python/pandas, DataFrames in PySpark have much more powerful optimization capabilities using the Catalyst optimizer.

They are designed for handling large-scale datasets, so there’s a high level of abstraction for structured data, thus benefiting from the capabilities of Spark’s distributed execution engine. In layperson’s terms, they’re damn fast!

Joins in PySpark

Joining means you’re combining data from two or more DataFrames based on a related column or index.

In PySpark, you can use these joins.

Before showing examples of each join type, let’s first set up PySpark and create sample DataFrames that we’ll join.

Initializing a PySpark Session And Creating a Sample Dataset

I won’t talk about installing PySpark here. If you don’t have PySpark installed already, follow one of the installation instructions from the official PySpark documentation.

Before joining DataFrames, you first need to initialize a PySpark session. Do it using this code.

from pyspark.sql import SparkSession

# Initialize Spark session
spark = SparkSession.builder \
    .appName("Custom DataFrame Creation") \
    .getOrCreate()

Next, we’ll create sample DataFrames that we’ll join in the following sections. The DataFrames we’ll use are recreated data from IBM’s Interaction Summary interview question.

# Data for 'customer_interactions'
customer_interactions_data = [
    (1, 7, "click", "2024-02-13"),
    (2, 4, "view", "2024-01-25"),
    (3, 8, "like", "2024-02-18"),
    (4, 5, "like", "2024-01-27"),
    (5, 7, "view", "2024-02-28"),
    (6, 10, "view", "2024-02-11"),
    (7, 3, "view", "2024-01-28"),
    (8, 7, "like", "2024-02-29"),
    (9, 8, "like", "2024-01-16"),
    (10, 5, "click", "2024-01-15"),
    (11, 4, "click", "2024-02-16"),
    (12, 8, "like", "2024-02-20"),
    (13, 8, "view", "2024-02-13"),
    (14, 3, "view", "2024-02-24"),
    (15, 6, "click", "2024-02-21")
]
customer_interactions_columns = ["interaction_id", "customer_id", "interaction_type", "interaction_date"]
customer_interactions_df = spark.createDataFrame(data=customer_interactions_data, schema=customer_interactions_columns)

# Data for 'user_content'
user_content_data = [
    (1, 2, "comment", "hello world! this is a TEST."),
    (2, 8, "comment", "what a great day"),
    (3, 4, "comment", "WELCOME to the event."),
    (4, 2, "comment", "e-commerce is booming."),
    (5, 6, "comment", "Python is fun!!"),
    (6, 6, "review", "123 numbers in text."),
    (7, 10, "review", "special chars: @#$$%^&*()"),
    (8, 4, "comment", "multiple CAPITALS here."),
    (9, 6, "review", "sentence. and ANOTHER sentence!"),
    (10, 2, "post", "goodBYE!")
]
user_content_columns = ["content_id", "customer_id", "content_type", "content_text"]
user_content_df = spark.createDataFrame(data=user_content_data, schema=user_content_columns)

# Show the DataFrames
customer_interactions_df.show()
user_content_df.show()

The above code returns the customer_interactions_df:

and user_content_df in the output:

Now that we have DataFrames ready, I can show you how to join them.

Joining DataFrames

In this section, I will show you codes for different join types and explain the outputs.

Inner Join

The DataFrames are inner-joined like this. The syntax is relatively simple. Create a DataFrame (inner_join_df) that will contain the joined data. Next, use the .join() operation to join the customer_interactions_df DataFrame with another DataFrame. We do it by defining parameters in .join(): first, we name the second DataFrame to be joined (user_content_df), then we define the column on which the DataFrames will be joined (on=”customer_id”), and, finally, we define the join type (how=”inner”).

inner_join_df = customer_interactions_df.join(user_content_df, on="customer_id", how="inner")
inner_join_df.show(truncate=False)

The second code line just displays the joined data (in the non-truncated format, which is important to fully see data in the content_text column). The output shows only the matching rows from both DataFrames.

Left Outer Join

The syntax for left outer join (and all other joins, except cross join, for that matter) is the same as in the previous example: the only thing you change is the join type parameter in the .join() operation.

left_join_df = customer_interactions_df.join(user_content_df, on="customer_id", how="left")
left_join_df.show(truncate=False)

Where you see NULLs in the output, those are non-matching values from user_content_df.

Right Outer Join

Here’s the code for right-joining the DataFrames.

right_join_df = customer_interactions_df.join(user_content_df, on="customer_id", how="right")
right_join_df.show(truncate=False)

As the right join is the mirror image of the left join, the NULLs in the output below are the non-matching rows from the left (customer_interactions_df) DataFrame.

Full Outer Join

Use this code to perform a full outer join on DataFrames in PySpark.

full_join_df = customer_interactions_df.join(user_content_df, on="customer_id", how="outer")
full_join_df.show(truncate=False)

NULLs appear in the output where there are non-matching rows in either DataFrames.

Cross Join

Use the following code to cross-join DataFrames. Here, the syntax is a bit different. Since cross join results in a Cartesian product (all combinations of all data rows), there’s no need to specify the column on which DataFrames will be joined.

Instead of .join(), we use .crossJoin() to join the DataFrames, with the second DataFrame written in parentheses.

cross_join_df = customer_interactions_df.crossJoin(user_content_df)
cross_join_df.show(truncate=False)

Here’s the output.

Semi Join

We’re now back to the syntax we’re used to; we again use the .join() operation. To create a semi join, use left_semi in the how argument.

semi_join_df = customer_interactions_df.join(user_content_df, on="customer_id", how="left_semi")
semi_join_df.show(truncate=False)

The output shows the rows from customer_interactions_df where there are matching rows in user_content_df.

Anti Join

To anti join DataFrames, simply use the left_anti keyword.

anti_join_df = customer_interactions_df.join(user_content_df, on="customer_id", how="left_anti")
anti_join_df.show(truncate=False)

The output shows the rows from customer_interactions_df where no matches exist in user_content_df.

Resources for Practicing PySpark

As with any coding, practicing is crucial for all the concepts to really sink in and become second nature. As for PySpark, here are several resources for practicing:

1. Databricks Community Edition
2. pyspark_exercises repository on GitHub
3. Analytical questions with a PySpark code editor on StrataScratch
4. PySpark projects and tutorials on Kaggle
5. DataCamp’s interactive PySpark Courses

Conclusion

There it is: seven different ways of joining DataFrames in PySpark. Understanding these joins and their use cases is crucial for building efficient data pipelines or performing almost any operation on large data sets.