Eric Campbell 5 min read Python

PySpark - SQL IN Operator

• The `isin()` method in PySpark provides cleaner syntax than multiple OR conditions, but performance degrades significantly when filtering against lists with more than a few hundred values—use...

Key Insights

• The isin() method in PySpark provides cleaner syntax than multiple OR conditions, but performance degrades significantly when filtering against lists with more than a few hundred values—use broadcast joins instead for large value sets.

• PySpark’s IN operator handles NULL values differently than you might expect: NULL values are never matched by isin(), and you must explicitly handle them with separate conditions if needed.

• Dynamic list generation from Python variables works seamlessly with isin(), but always validate for empty lists to avoid runtime errors and unexpected behavior in production pipelines.

Introduction to the IN Operator in PySpark

The IN operator is a fundamental SQL construct that filters rows based on whether a column value matches any value in a specified list. In PySpark, this functionality is primarily implemented through the isin() method, though you can also use SQL-style syntax when working with Spark SQL queries.

The IN operator shines when you need to filter data against a specific set of known values. Instead of chaining multiple OR conditions—which becomes unwieldy and error-prone—you can express the same logic concisely. For example, filtering for three specific product categories using OR conditions requires verbose syntax, while the IN operator handles it elegantly in a single expression.

Here’s the difference in approach:

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

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

# Sample data
data = [
    ("laptop", "electronics", 1200),
    ("shirt", "clothing", 45),
    ("phone", "electronics", 800),
    ("jeans", "clothing", 60),
    ("book", "media", 15)
]
df = spark.createDataFrame(data, ["product", "category", "price"])

# Traditional OR approach - verbose
filtered_or = df.filter(
    (col("category") == "electronics") | 
    (col("category") == "clothing")
)

# IN operator approach - clean
filtered_in = df.filter(col("category").isin("electronics", "clothing"))

Both produce identical results, but isin() is more maintainable and readable, especially as your value list grows.

Basic IN Operator Syntax

PySpark offers two primary ways to use the IN operator: the DataFrame API with isin() and SQL expressions with the traditional IN clause.

The isin() method accepts values as individual arguments or can be unpacked from a Python list:

# Method 1: Direct values as arguments
result1 = df.filter(col("category").isin("electronics", "clothing", "media"))

# Method 2: Unpacking from a list
categories = ["electronics", "clothing", "media"]
result2 = df.filter(col("category").isin(*categories))

# Method 3: Using string column name
result3 = df.filter(df.category.isin("electronics", "clothing"))

For SQL-style queries, use the traditional IN syntax:

# Register DataFrame as temporary view
df.createOrReplaceTempView("products")

# SQL-style IN operator
result_sql = spark.sql("""
    SELECT product, category, price 
    FROM products 
    WHERE category IN ('electronics', 'clothing')
""")

result_sql.show()

Both approaches are equally valid. Choose the DataFrame API for programmatic pipelines and SQL syntax when working with analysts familiar with traditional SQL or when migrating existing queries.

Working with Different Data Types

The IN operator works seamlessly across different data types, but you need to understand type handling to avoid subtle bugs.

String Filtering:

# String values - case sensitive by default
string_data = [("Alice", "NY"), ("Bob", "CA"), ("Charlie", "TX")]
df_strings = spark.createDataFrame(string_data, ["name", "state"])

# Standard string filtering
result = df_strings.filter(col("state").isin("NY", "CA", "FL"))
result.show()
# Returns: Alice (NY), Bob (CA)

Numeric Value Filtering:

# Integer and float values
numeric_data = [(1, 100.5), (2, 200.75), (3, 150.25)]
df_numeric = spark.createDataFrame(numeric_data, ["id", "amount"])

# Filter by integer IDs
result = df_numeric.filter(col("id").isin(1, 3, 5))
result.show()

Date and Timestamp Filtering:

from pyspark.sql.functions import to_date
from datetime import date

date_data = [
    ("2024-01-15",), ("2024-02-20",), ("2024-03-10",)
]
df_dates = spark.createDataFrame(date_data, ["date_string"])
df_dates = df_dates.withColumn("date", to_date("date_string"))

# Filter using date objects
target_dates = [date(2024, 1, 15), date(2024, 3, 10)]
result = df_dates.filter(col("date").isin(*target_dates))
result.show()

NULL Value Handling:

This is critical—isin() does NOT match NULL values:

null_data = [("A", 1), ("B", None), ("C", 2), ("D", None)]
df_nulls = spark.createDataFrame(null_data, ["letter", "number"])

# This will NOT return rows with NULL numbers
result = df_nulls.filter(col("number").isin(1, 2, None))
result.show()
# Returns only: A (1), C (2)

# To include NULLs, add explicit condition
result_with_nulls = df_nulls.filter(
    col("number").isin(1, 2) | col("number").isNull()
)
result_with_nulls.show()
# Returns: A (1), B (None), C (2), D (None)

Advanced Techniques

Dynamic List Generation:

Real-world applications often generate filter lists dynamically from configuration, database queries, or user input:

# Generate list from another DataFrame
categories_df = spark.createDataFrame([("electronics",), ("media",)], ["cat"])
category_list = [row.cat for row in categories_df.collect()]

# Use in filtering
filtered = df.filter(col("category").isin(*category_list))

Combining IN with Other Conditions:

# AND condition: IN operator + price range
premium_electronics = df.filter(
    (col("category").isin("electronics", "media")) & 
    (col("price") > 500)
)

# OR condition: Multiple IN operators
result = df.filter(
    col("category").isin("electronics") | 
    col("product").isin("shirt", "jeans")
)

NOT IN Operator:

The inverse operation uses ~ (tilde) or the isin() method negated:

# Exclude specific categories
excluded = df.filter(~col("category").isin("clothing", "media"))

# Alternative syntax
excluded_alt = df.filter(col("category").isin("clothing", "media") == False)

Subquery Results with IN:

# Get categories from one DataFrame to filter another
high_value_categories = df.filter(col("price") > 100) \
    .select("category").distinct()

# Collect to list and use in filter
categories_to_keep = [row.category for row in high_value_categories.collect()]
result = df.filter(col("category").isin(*categories_to_keep))

Performance Considerations

The IN operator’s performance characteristics change dramatically based on list size. For small lists (under 100 values), isin() performs excellently. Beyond that, consider alternatives.

Performance Comparison:

from pyspark.sql.functions import broadcast

# Large list scenario - 1000+ values
large_list = list(range(1000))

# Approach 1: IN operator (slower for large lists)
result_in = df.filter(col("id").isin(*large_list))

# Approach 2: Broadcast join (faster for large lists)
filter_df = spark.createDataFrame([(x,) for x in large_list], ["id"])
result_join = df.join(broadcast(filter_df), "id", "inner")

The broadcast join approach creates a small DataFrame from your filter list, broadcasts it to all executors, and performs an inner join. For lists exceeding 500-1000 values, this typically outperforms isin() because:

  1. The IN operator generates a large expression tree that Catalyst must optimize
  2. Broadcast joins leverage Spark’s distributed computing more efficiently
  3. Join operations are highly optimized in Spark’s execution engine

Optimization Tips:

  • Cache DataFrames when applying multiple IN filters to the same dataset
  • Use partition pruning when your IN values align with partition columns
  • Consider bloom filters for extremely large value sets (millions of values)
  • Monitor query plans with explain() to identify bottlenecks
# Check execution plan
df.filter(col("category").isin(*large_list)).explain()

Common Pitfalls and Best Practices

Empty List Handling:

Empty lists cause runtime failures. Always validate:

def safe_filter(df, column, values):
    if not values:
        return df.filter(col(column).isNull())  # or return empty df
    return df.filter(col(column).isin(*values))

# Usage
filter_values = []  # Empty list from some process
result = safe_filter(df, "category", filter_values)

Case Sensitivity:

String comparisons are case-sensitive by default:

# Won't match "ELECTRONICS" or "Electronics"
df.filter(col("category").isin("electronics"))

# Case-insensitive workaround
from pyspark.sql.functions import lower

df.filter(lower(col("category")).isin("electronics", "clothing"))

Maximum Values Limitation:

While PySpark doesn’t enforce a hard limit, extremely large IN lists (10,000+ values) cause:

  • Slow query planning
  • Memory pressure on the driver
  • Poor performance compared to joins

Best Practices:

  1. Validate input lists before passing to isin() to prevent empty list errors
  2. Use broadcast joins for lists exceeding 500 values
  3. Leverage partition pruning by filtering on partition columns when possible
  4. Handle NULLs explicitly rather than assuming they’ll be matched
  5. Consider set operations when filtering one DataFrame against another’s values
# Good practice: Defensive filtering
def robust_filter(df, column_name, filter_values):
    # Validate inputs
    if not filter_values:
        raise ValueError(f"Empty filter list for column {column_name}")
    
    # Remove duplicates and None values
    clean_values = [v for v in set(filter_values) if v is not None]
    
    # Choose strategy based on list size
    if len(clean_values) > 500:
        filter_df = spark.createDataFrame(
            [(v,) for v in clean_values], [column_name]
        )
        return df.join(broadcast(filter_df), column_name, "inner")
    else:
        return df.filter(col(column_name).isin(*clean_values))

The IN operator is a powerful tool in PySpark, but understanding its performance characteristics and edge cases ensures you use it effectively in production data pipelines.

Liked this? There's more.

Every week: one practical technique, explained simply, with code you can use immediately.