How to Filter Rows Based on Multiple Conditions in a PySpark DataFrame: The Ultimate Guide

Published on April 17, 2025

Diving Straight into Filtering Rows with Multiple Conditions in a PySpark DataFrame

Filtering rows in a PySpark DataFrame based on multiple conditions is a powerful technique for data engineers using Apache Spark, enabling precise data extraction for complex queries in ETL pipelines. Whether you're selecting employees meeting specific salary and age criteria, identifying transactions within a date range and category, or refining datasets for analysis, this skill ensures targeted data selection. This comprehensive guide explores the syntax and steps for filtering rows using multiple conditions, with examples covering basic multi-condition filtering, nested data, handling nulls, and SQL-based approaches. Each section addresses a specific aspect of filtering, supported by practical code, error handling, and performance optimization strategies to build robust pipelines. The primary method, filter() or where(), is explained with all relevant considerations. Let’s refine those datasets! For more on PySpark, see PySpark Fundamentals.

Filtering Rows with Multiple Conditions

The primary method for filtering rows in a PySpark DataFrame is the filter() method (or its alias where()), which selects rows meeting specified conditions. To filter based on multiple conditions, combine boolean expressions using logical operators (& for AND, | for OR, ~ for NOT). This approach is ideal for ETL pipelines requiring complex data selection criteria, such as combining numerical thresholds, string matches, or other comparisons.

Understanding filter() and where() Parameters

  • filter(condition) or where(condition):
    • condition (Column or str, required): A boolean expression defining the filtering criteria, such as (col("column1") > value1) & (col("column2") == value2) or a SQL-like string (e.g., "column1 > value1 AND column2 = value2").
    • Returns: A new DataFrame containing only the rows where the condition evaluates to True.
    • Note: filter() and where() are interchangeable, with where() offering a SQL-like syntax for readability. Parentheses are crucial when combining conditions to ensure correct operator precedence.

Here’s an example filtering employees with a salary greater than 80,000 and age less than 30:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Initialize SparkSession
spark = SparkSession.builder.appName("MultiConditionFilter").getOrCreate()

# Create DataFrame
data = [
    ("E001", "Alice", 25, 75000.0, "HR"),
    ("E002", "Bob", 30, 82000.5, "IT"),
    ("E003", "Cathy", 28, 90000.75, "HR"),
    ("E004", "David", 35, 100000.25, "IT"),
    ("E005", "Eve", 28, 78000.0, "Finance")
]
df = spark.createDataFrame(data, ["employee_id", "name", "age", "salary", "department"])

# Filter rows where salary > 80000 and age < 30
filtered_df = df.filter((col("salary") > 80000) & (col("age") < 30))
filtered_df.show(truncate=False)

Output:

+-----------+-----+---+---------+----------+
|employee_id|name |age|salary   |department|
+-----------+-----+---+---------+----------+
|E003       |Cathy|28 |90000.75 |HR        |
+-----------+-----+---+---------+----------+

This filters rows where salary exceeds 80,000 and age is less than 30, returning one row (E003). The & operator combines conditions, requiring both to be true. Parentheses ensure correct precedence. Validate:

assert filtered_df.count() == 1, "Incorrect row count"
assert "Cathy" in [row["name"] for row in filtered_df.select("name").collect()], "Expected name missing"

Error to Watch: Missing parentheses causes precedence errors:

try:
    filtered_df = df.filter(col("salary") > 80000 & col("age") < 30)  # Incorrect precedence
    filtered_df.show()
except Exception as e:
    print(f"Error: {e}")

Output:

Error: Operator precedence error

Fix: Use parentheses:

assert isinstance((col("salary") > 80000) & (col("age") < 30), Column), "Invalid condition syntax"

Filtering with Complex Multi-Condition Logic

For more intricate filtering, combine multiple conditions using AND (&), OR (|), and NOT (~) operators to create complex logic. This is useful for scenarios requiring nuanced criteria, such as filtering employees with specific salary ranges, ages, or departments.

from pyspark.sql.functions import col

# Filter rows where (salary > 80000 AND age < 30) OR department is Finance
filtered_df = df.filter(((col("salary") > 80000) & (col("age") < 30)) | (col("department") == "Finance"))
filtered_df.show(truncate=False)

Output:

+-----------+-----+---+---------+----------+
|employee_id|name |age|salary   |department|
+-----------+-----+---+---------+----------+
|E003       |Cathy|28 |90000.75 |HR        |
|E005       |Eve  |28 |78000.0  |Finance   |
+-----------+-----+---+---------+----------+

This filters rows where either salary exceeds 80,000 and age is less than 30, or department is "Finance", returning two rows (E003, E005). The | operator allows rows meeting either condition. Validate:

assert filtered_df.count() == 2, "Incorrect row count"
assert filtered_df.filter(col("department") == "Finance").count() == 1, "Incorrect Finance count"

Error to Watch: Null values in conditions can skew results:

# Example with nulls
data_with_nulls = data + [("E006", None, 32, 85000.0, "IT")]
df_nulls = spark.createDataFrame(data_with_nulls, ["employee_id", "name", "age", "salary", "department"])
filtered_df = df_nulls.filter((col("salary") > 80000) & (col("name") != "Bob"))
# Includes E006 despite null name

Fix: Handle nulls explicitly:

filtered_df = df_nulls.filter((col("salary") > 80000) & (col("name") != "Bob") & (col("name").isNotNull()))
assert filtered_df.count() == 2, "Nulls not handled correctly"

Filtering Nested Data with Multiple Conditions

Nested DataFrames, with structs or arrays, are common in complex datasets like employee contact details. Filtering rows based on multiple conditions involving nested fields, such as contact.phone and contact.email, requires dot notation (e.g., contact.phone) within filter(). This is essential for hierarchical data in ETL pipelines.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType
from pyspark.sql.functions import col

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

# Define schema with nested structs
schema = StructType([
    StructField("employee_id", StringType(), False),
    StructField("name", StringType(), True),
    StructField("contact", StructType([
        StructField("phone", LongType(), True),
        StructField("email", StringType(), True)
    ]), True),
    StructField("department", StringType(), True)
])

# Create DataFrame
data = [
    ("E001", "Alice", (1234567890, "alice@company.com"), "HR"),
    ("E002", "Bob", (None, "bob@company.com"), "IT"),
    ("E003", "Cathy", (5555555555, "cathy@company.com"), "HR"),
    ("E004", "David", (9876543210, "david@company.com"), "IT")
]
df = spark.createDataFrame(data, schema)

# Filter rows where contact.phone > 5000000000 and department is IT
filtered_df = df.filter((col("contact.phone") > 5000000000) & (col("department") == "IT"))
filtered_df.show(truncate=False)

Output:

+-----------+-----+--------------------------------+----------+
|employee_id|name |contact                         |department|
+-----------+-----+--------------------------------+----------+
|E004       |David|[9876543210, david@company.com] |IT        |
+-----------+-----+--------------------------------+----------+

This filters rows where contact.phone exceeds 5,000,000,000 and department is "IT", returning one row (E004). Validate:

assert filtered_df.count() == 1, "Incorrect row count"
assert "David" in [row["name"] for row in filtered_df.select("name").collect()], "Expected name missing"

Error to Watch: Invalid nested field fails:

try:
    filtered_df = df.filter((col("contact.invalid_field") > 5000000000) & (col("department") == "IT"))
    filtered_df.show()
except Exception as e:
    print(f"Error: {e}")

Output:

Error: StructField 'contact' does not contain field 'invalid_field'

Fix: Validate nested field:

assert "phone" in [f.name for f in df.schema["contact"].dataType.fields], "Nested field missing"

Filtering Using SQL Queries

For SQL-based ETL workflows or teams familiar with database querying, SQL queries via temporary views offer an intuitive way to filter rows with multiple conditions. The WHERE clause combines conditions using AND, OR, and NOT, mimicking the filter() logic in a SQL-like syntax.

# Create temporary view
df.createOrReplaceTempView("employees")

# Filter rows where salary > 80000 and age < 30 using SQL
filtered_df = spark.sql("SELECT * FROM employees WHERE salary > 80000 AND age < 30")
filtered_df.show(truncate=False)

Output:

+-----------+-----+---+---------+----------+
|employee_id|name |age|salary   |department|
+-----------+-----+---+---------+----------+
|E003       |Cathy|28 |90000.75 |HR        |
+-----------+-----+---+---------+----------+

This filters rows where salary exceeds 80,000 and age is less than 30 using SQL. Validate:

assert filtered_df.count() == 1, "Incorrect row count"
assert "Cathy" in [row["name"] for row in filtered_df.select("name").collect()], "Expected name missing"

Error to Watch: Unregistered view fails:

try:
    filtered_df = spark.sql("SELECT * FROM nonexistent WHERE salary > 80000 AND age < 30")
    filtered_df.show()
except Exception as e:
    print(f"Error: {e}")

Output:

Error: Table or view not found: nonexistent

Fix: Verify view:

assert "employees" in [v.name for v in spark.catalog.listTables()], "View missing"
df.createOrReplaceTempView("employees")

Optimizing Performance for Multi-Condition Filtering

Filtering rows with multiple conditions involves scanning the DataFrame and evaluating complex expressions, which can be computationally intensive for large datasets. Optimize performance to ensure efficient data extraction:

  1. Select Relevant Columns: Reduce data scanned:
df = df.select("employee_id", "name", "age", "salary", "department")
  1. Push Down Filters: Apply filters early to minimize data:
df = df.filter((col("salary") > 80000) & (col("age") < 30))
  1. Partition Data: Use partitionBy or repartition for large datasets:
df = df.repartition("department")
  1. Cache Intermediate Results: Cache filtered DataFrame if reused:
filtered_df.cache()

Example optimized filter:

optimized_df = df.select("employee_id", "name", "age", "salary", "department") \
                .filter((col("salary") > 80000) & (col("age") < 30)) \
                .repartition("department")
optimized_df.show(truncate=False)

Monitor performance via the Spark UI, focusing on scan and filter metrics.

Error to Watch: Large datasets with inefficient filtering slow performance:

# Example with large DataFrame
large_df = spark.range(10000000).join(df, "employee_id", "left")
filtered_df = large_df.filter((col("salary") > 80000) & (col("age") < 30))  # Inefficient

Fix: Optimize with early filtering and partitioning:

assert large_df.count() < 10000000, "Large dataset, optimize with early filters or partitioning"

Wrapping Up Your Multi-Condition Filtering Mastery

Filtering rows based on multiple conditions in a PySpark DataFrame is a vital skill for precise data extraction in ETL pipelines. Whether you’re using filter() or where() to combine conditions with logical operators, handling nested data with dot notation, addressing nulls, or leveraging SQL queries for intuitive filtering, Spark provides powerful tools to address complex data processing needs. By mastering these techniques, optimizing performance, and anticipating errors, you can efficiently refine datasets, enabling accurate analyses and robust applications. These methods will enhance your data engineering workflows, empowering you to manage data filtering with confidence.

Try these approaches in your next Spark job, and share your experiences, tips, or questions in the comments or on X. Keep exploring with DataFrame Operations to deepen your PySpark expertise!

More Spark Resources to Keep You Going