How to Perform Data Cleaning in PySpark

Data cleaning is not a super fun task! But it is necessary if you want to work with real-world datasets at all. This tutorial will help you learn how to perform data cleaning tasks such as handling missing values, removing duplicates, and transforming columns when analyzing large datasets with PySpark.

Before You Begin

Ensure you have PySpark installed:

! pip3 install pyspark

You can either install PySpark on your local working environment. Or you can follow along using a cloud notebook environment. Here’s the Google Colab notebook for this tutorial.

1. Start a PySpark Session

First, start a PySpark session.

from pyspark.sql import SparkSession

# Initialize a Spark session
spark = SparkSession.builder \
	.appName("DataCleaning") \
	.getOrCreate()

2. Generate a Sample Dataset

You can work with any dataset of your choice. We’ll create a dataset with some typical data quality issues such as missing values, duplicate records, and the like.

import random
import pandas as pd

# Function to generate random data with some missing values and duplicates
def generate_data(n):
    customer_ids = [f'C{str(i).zfill(5)}' for i in range(1, 101)]
    product_categories = ['Electronics', 'Books', 'Clothing', 'Groceries', 'Furniture']

    data = []
    for i in range(n):
        customer_id = random.choice(customer_ids) if i % 10 != 0 else None  # Introduce some missing values
        transaction_id = f'T{str(random.randint(10000, 99999))}'
        transaction_date = pd.Timestamp('2023-01-01') + pd.to_timedelta(random.randint(0, 180), unit='d')
        amount = round(random.uniform(5, 500), 2)
        product_category = random.choice(product_categories)
        data.append((customer_id, transaction_id, transaction_date, amount, product_category))

        # Introduce duplicates
        data.extend(data[:10])

     return data

We’ll generate a dataframe of 10000 records.

# Generate 10,000 rows of data
data = generate_data(10_000)

columns = ['CustomerID', 'TransactionID', 'TransactionDate', 'Amount', 'ProductCategory']
df = pd.DataFrame(data, columns=columns)
spark_df = spark.createDataFrame(df)

spark_df.show(5)

Output:

+----------+-------------+-------------------+------+---------------+
|CustomerID|TransactionID|    TransactionDate|Amount|ProductCategory|
+----------+-------------+-------------------+------+---------------+
|      NULL|       T17203|2023-03-20 00:00:00|221.92|          Books|
|      NULL|       T17203|2023-03-20 00:00:00|221.92|          Books|
|    C00058|       T63296|2023-02-11 00:00:00|157.92|      Groceries|
|      NULL|       T17203|2023-03-20 00:00:00|221.92|          Books|
|      NULL|       T17203|2023-03-20 00:00:00|221.92|          Books|
+----------+-------------+-------------------+------+---------------+
only showing top 5 rows

3. Handle Missing Values

Missing values are a common problem in most datasets. You can handle missing values by dropping records with missing values (this isn’t normally recommended!) or filling them with default values.

If you want to drop rows with missing Customer ID, you can do it like so:

# Drop rows with missing CustomerID
spark_df = spark_df.dropna(subset=["CustomerID"])

But you can also choose to fill the missing ID field with a placeholder like ‘Unknown’:

# Fill missing CustomerID with a default value
spark_df = spark_df.fillna({"CustomerID": "Unknown"})

4. Remove Duplicates

Removing duplicates is another common data cleaning task. In this example, we’ll drop duplicate rows based on a specific column, say the ‘TransactionID’.

# Drop duplicate rows based on 'TransactionID'
spark_df = spark_df.dropDuplicates(subset=["TransactionID"])

5. Transform Columns

Suppose we want to normalize the ‘Amount’ column by scaling it between 0 and 1. Here’s how we can do it.

from pyspark.sql.functions import col, min, max

# Normalize the 'Amount' column
min_amount = spark_df.agg(min(col("Amount"))).collect()[0][0]
max_amount = spark_df.agg(max(col("Amount"))).collect()[0][0]

spark_df = spark_df.withColumn("Amount", (col("Amount") - min_amount) / (max_amount - min_amount))

6. Handle Outliers

Outliers can significantly skew your data analysis. A common method for detecting outliers is to use the Interquartile Range (IQR)—the difference between the third and first quartiles—Q3 – Q1.

And the points outside the interval [Q1 + 1.5*IQR, Q3 – 1.5*IQR] are generally considered outliers.

from pyspark.sql.functions import col, expr

# Calculate Q1, Q3, and IQR
quantiles = spark_df.approxQuantile("Amount", [0.25, 0.75], 0.05)
Q1 = quantiles[0]
Q3 = quantiles[1]
IQR = Q3 - Q1

# Define the upper and lower bounds
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# Filter out the outliers
spark_df = spark_df.filter((col("Amount") >= lower_bound) & (col("Amount") <= upper_bound))

7. Convert Data Types

Converting data types is another common task in data cleaning. The fields in the sample dataset have the expected data type.

spark_df.dtypes

Output:

[('CustomerID', 'string'),
 ('TransactionID', 'string'),
 ('TransactionDate', 'date'),
 ('Amount', 'double'),
 ('ProductCategory', 'string')]

But if there are one or more fields with incorrect data types, you can convert them to the expected data type.

For example, you can convert ‘TransactionDate’ to a date type like so:

from pyspark.sql.functions import to_date

# Convert 'TransactionDate' to date format
spark_df = spark_df.withColumn("TransactionDate", to_date(col("TransactionDate"))

In this tutorial, we’ve performed data cleaning in PySpark on a sample dataset—one step at a time. Now try cleaning another dataset of your choice. Happy data cleaning!

Additional Resources

Here are a few resources for additional reference: