Handling Missing Data in PySpark

Handling missing data is a crucial aspect of data cleaning and preprocessing. PySpark offers multiple methods to address missing or null values, depending on the specific requirements of your workflow. This guide provides a structured approach, detailed examples, and additional use cases.

Methods to Handle Missing Data in PySpark

Drop Rows with Missing Values (dropna)

You can drop rows containing null or missing values using the dropna() method. This can be applied globally or to specific columns.

Drop Rows with Any Missing Values

# Remove rows with any missing values
df_clean = df.dropna()

Drop Rows Based on Specific Columns

# Remove rows where 'column1' or 'column2' contains null values
df_clean = df.dropna(subset=["column1", "column2"])

Drop Rows with a Threshold

Retain rows with at least a specified number of non-null values.

# Remove rows with fewer than 2 non-null values
df_clean = df.dropna(thresh=2)

---

Replace Missing Values (fillna)

Replace missing values with specified constants or default values.

Replace All Null Values with a Single Value

# Replace all null values with 0
df_filled = df.fillna(0)

Replace Nulls in Specific Columns

# Replace nulls in 'column1' with 0 and in 'column2' with 'unknown'
df_filled = df.fillna({"column1": 0, "column2": "unknown"})

---

Impute Missing Values with Mean, Median, or Mode

PySpark allows filling missing values with statistical measures such as mean, median, or mode.

Fill Missing Values with Mean

from pyspark.sql.functions import mean

# Calculate the mean of 'column1'
mean_value = df.select(mean(df['column1'])).collect()[0][0]

# Replace missing values in 'column1' with the mean
df_filled = df.fillna({"column1": mean_value})

Use pyspark.ml.feature.Imputer

from pyspark.ml.feature import Imputer

# Create and configure an Imputer
imputer = Imputer(inputCols=["column1", "column2"], 
                  outputCols=["column1_imputed", "column2_imputed"])
df_imputed = imputer.fit(df).transform(df)

Change the strategy using imputer.setStrategy("median").

---

Identifying Rows with Missing Data

Before handling missing data, it’s helpful to identify where and how much data is missing.

Filter Rows with Null Values

# Filter rows where 'column1' has a null value
df_nulls = df.filter(df['column1'].isNull())

Count Missing Values in Each Column

from pyspark.sql.functions import col, count, when

# Count null values in each column
df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show()

---

Handling Complex Data Types

PySpark supports complex types like arrays, structs, and maps. Missing values in such types can also be handled effectively.

Fill Missing Values in Struct Columns

from pyspark.sql.functions import when, struct, lit

# Fill null 'city' field in a StructType column
df_filled = df.withColumn(
    "address",
    when(col("address.city").isNull(), 
         struct(lit("Unknown").alias("city"), col("address.state"))
    ).otherwise(col("address"))
)

---

Dropping or Filling Columns with High Missing Rates

Sometimes, dropping columns with too many missing values is more practical than handling each missing entry.

Drop Columns with a High Percentage of Missing Values

# Set threshold for missing value percentage
threshold = 0.5
total_rows = df.count()

# Identify columns with more than 50% missing values
cols_to_drop = [c for c in df.columns if df.filter(col(c).isNull()).count() / total_rows > threshold]

# Drop these columns
df_clean = df.drop(*cols_to_drop)

---

Handling Nulls and NaNs in Joins

When performing joins, nulls in key columns can affect the results.

Fill Nulls Before Join

# Fill null values in the join key column
df = df.fillna({"join_key": "default_value"})
other_df = other_df.fillna({"join_key": "default_value"})

# Perform the join
joined_df = df.join(other_df, on="join_key")

---

Managing None, null, and NaN

Key Differences

• None: Python’s equivalent of null. Treated as null in PySpark.
• null: Spark’s representation of missing values.
• NaN: Numeric representation of invalid or undefined values (specific to FloatType and DoubleType).

Examples

Check for None or null

# Filter rows where 'name' is null
df.filter(df["name"].isNull()).show()

Check for NaN

from pyspark.sql.functions import isnan

# Filter rows where 'score' is NaN
df.filter(isnan(df["score"])).show()

Handle None, null, and NaN Together

# Replace NaN and null values
df_filled = df.fillna({"name": "Unknown", "score": 0})
df_filled.show()

---

.na.drop() vs dropna()

Both .na.drop() and dropna() drop rows with null values. The difference lies in how they’re called:

• .na.drop(): Accessed via the .na namespace.
• dropna(): Directly called on the DataFrame.

Example

# Using .na.drop()
df_clean = df.na.drop()

# Using dropna()
df_clean = df.dropna()

Both methods support arguments like subset, how, and thresh.

---

Summary

Key Methods

• Drop rows: dropna()
• Replace nulls: fillna()
• Impute values: pyspark.ml.feature.Imputer
• Identify nulls: isNull(), isnan()

Use Cases

• Drop rows with any/all nulls.
• Replace or fill missing values with constants or computed values.
• Impute values using statistical measures.
• Handle complex data types and nested fields.

PySpark provides a robust toolkit to manage missing data, enabling you to maintain data integrity and make informed decisions based on your specific dataset requirements.

---