Definition
- Predicate pushdown – an optimization where Catalyst pushes filter conditions to the data source so rows are eliminated before entering executor memory.
- Columnar formats – Parquet and ORC use per-row-group min/max statistics to skip entire chunks without decompressing.
- JDBC – the predicate becomes a SQL WHERE clause executed by the database with its own indexes.
- Row-based formats – CSV and JSON have no statistics, so the reader must parse every row before filtering.
- Verification – call
explain()and read the PushedFilters line on the scan node to confirm pushdown occurred.
Imagine the National Bank of India's archive clerk. Without pushdown, the clerk hauls every box of records to your desk and you throw away the irrelevant ones yourself. With predicate pushdown, you tell the clerk up front "only boxes for transactions above fifty thousand rupees," and the archive's index lets the clerk leave the other boxes on the shelf. But if the archive is loose unlabelled papers in a heap (CSV), there is no index, so the clerk still reads every sheet.
Understanding the Concepts
What is predicate pushdown?
When you write a filter in a Spark query, Catalyst does not blindly read all data and then filter. It analyzes the filter and, if the data source supports it, hands the predicate to the source reader. The source returns fewer rows, cutting I/O, network, and deserialization cost.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("PushdownDemo").getOrCreate()
# Filter on amount > 50000 is pushed to the Parquet reader
result = spark.read.parquet("/data/transactions") \
.filter(col("amount") > 50000) \
.select("account_id", "amount")
# Catalyst pushes the filter to the source
# The Parquet reader skips row groups where max(amount) <= 50000
result.explain(True)
== Physical Plan ==
*(1) Project [account_id, amount]
+- *(1) Filter (amount > 50000)
+- *(1) ColumnarToRow
+- FileScan parquet [account_id,amount]
PushedFilters: [GreaterThan(amount,50000)]
ReadSchema: struct<account_id:string,amount:decimal(12,2)>
How does pushdown work with Parquet and ORC?
Parquet and ORC divide files into row groups (Parquet) or stripes (ORC). Each column chunk stores min/max statistics. The reader checks these statistics against the filter and skips chunks whose range cannot match.
# Parquet row group statistics enable file-level skipping
# File has 10 row groups, each with min/max for amount:
# RG 0: min=100, max=5000 -> skip (max < 50000)
# RG 1: min=2000, max=15000 -> skip (max < 50000)
# RG 2: min=10000, max=200000 -> READ (50000 in range)
# RG 3: min=500, max=3000 -> skip
# ... and so on
# Only row groups where the filter could match are read!
# ORC works similarly with stripe-level statistics
orc_result = spark.read.orc("/data/transactions_orc") \
.filter(col("amount") > 50000)
orc_result.explain(True)
How does pushdown work with JDBC sources?
For JDBC, Catalyst translates the Spark filter into a SQL WHERE clause sent to the database. The database uses its own indexes, making this extremely effective.
# JDBC pushdown: filter becomes a SQL WHERE clause
result = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://db.nbi.com:5432/banking") \
.option("dbtable", "transactions") \
.option("user", "spark_reader") \
.option("password", "***") \
.load() \
.filter(col("account_id") == "SAV2001") \
.filter(col("amount") > 50000)
result.explain(True)
# PushedFilters: [EqualTo(account_id,SAV2001), GreaterThan(amount,50000)]
# The database executes: SELECT * FROM transactions
# WHERE account_id = 'SAV2001' AND amount > 50000
# using its B-tree index on account_id
Why is pushdown less effective with CSV and JSON?
Row-based text formats have no per-column statistics and no block-level min/max. The reader must read, decompress, and parse every row before evaluating the filter. The filter still reduces downstream rows, but saves almost nothing on the expensive read-and-parse step.
# CSV: no pushdown benefit on the read
csv_result = spark.read \
.option("header", "true") \
.schema("account_id STRING, amount DECIMAL(12,2), txn_date DATE") \
.csv("/data/transactions.csv") \
.filter(col("amount") > 50000)
csv_result.explain(True)
# Note: PushedFilters is empty or shows the filter as a Spark-side Filter
# The CSV reader reads EVERY row and Spark filters afterward
Performance comparison on the same dataset:
Parquet with pushdown: reads 2 of 10 row groups -> 1.5 seconds
CSV without pushdown: reads all rows, filters -> 45 seconds
30x difference because Parquet skips 80% of the data at the source
What predicates can and cannot be pushed down?
Not all filters are pushable. The source must understand the predicate type.
| Pushable | Not Pushable | |---|---| | `=`, `!=`, `<`, `>`, `<=`, `>=` | User-defined functions (UDFs) | | `IN (list)` | Non-deterministic functions (rand) | | `IS NULL`, `IS NOT NULL` | Complex expressions (substring) | | `BETWEEN` | Functions wrapping the column | | `AND`, `OR` combinations | Cross-column predicates |
from pyspark.sql.functions import udf, rand
from pyspark.sql.types import BooleanType
# This filter IS pushed down
pushable = spark.read.parquet("/data/transactions") \
.filter(col("amount") > 50000)
# This UDF filter is NOT pushed down (opaque to Catalyst)
@udf(BooleanType())
def is_high_value(amount):
return amount > 50000
not_pushable = spark.read.parquet("/data/transactions") \
.filter(is_high_value(col("amount")))
# UDF runs in Spark AFTER reading all data from source
# This non-deterministic filter is NOT pushed down
not_pushable_2 = spark.read.parquet("/data/transactions") \
.filter(rand() < 0.01)
How do you verify pushdown happened?
Call explain() and look for the PushedFilters line on the scan node. Filters listed there were pushed to the source.
result = spark.read.parquet("/data/transactions") \
.filter(col("account_id") == "SAV2001") \
.filter(col("amount") > 50000)
result.explain(True)
# Look for:
# PushedFilters: [EqualTo(account_id,SAV2001), GreaterThan(amount,50000)]
# If a filter appears ONLY as a Filter node above the scan,
# it was NOT pushed down and runs in Spark after reading
What about column pruning alongside pushdown?
Catalyst also pushes column selection to the source, so only the columns used in the query are read. Combined with predicate pushdown, this dramatically reduces I/O.
# Only reads account_id and amount columns from Parquet
# AND skips row groups where amount <= 50000
result = spark.read.parquet("/data/transactions") \
.filter(col("amount") > 50000) \
.select("account_id", "amount")
result.explain(True)
# ReadSchema: struct<account_id:string,amount:decimal(12,2)>
# Other columns (txn_date, merchant, etc.) are not read at all!
What should a senior engineer know?
- Predicate pushdown is one of the most impactful optimizations in Spark – it can reduce I/O by orders of magnitude.
- Store analytical data in Parquet or ORC rather than CSV or JSON to enable effective pushdown and column pruning.
- At the National Bank of India, a high-value transfer query on Parquet skipped most row groups using statistics, while the same query on CSV read every row.
- ZORDER and partition pruning amplify pushdown by tightening the min/max ranges per file.
- Always verify pushdown with explain() before assuming it happened; UDFs and complex expressions silently prevent it.
Visual Overview
Without Predicate Pushdown (CSV) ┌─────────────────────────────────┐ │ Source: 10 GB CSV │ │ Read ALL rows into Spark │──> 10 GB into memory │ Then filter in Spark │──> 500 MB survives └─────────────────────────────────┘ With Predicate Pushdown (Parquet) ┌─────────────────────────────────┐ │ Source: 10 GB Parquet │ │ Check row group min/max stats │ │ Skip row groups outside range │──> 2 GB read (8 GB skipped) │ Filter remaining in Spark │──> 500 MB survives └─────────────────────────────────┘
Parquet Row Group Skipping ┌───────────────┐ amount max=5000 -> SKIP │ Row Group 0 │ ├───────────────┤ amount max=15000 -> SKIP │ Row Group 1 │ ├───────────────┤ amount max=200000 -> READ │ Row Group 2 │ ├───────────────┤ amount max=3000 -> SKIP │ Row Group 3 │ └───────────────┘ Filter: amount > 50000 Only Row Group 2 is decompressed and read!
Key Takeaway
Predicate pushdown is the optimization where Catalyst pushes filter conditions to the data source so rows are eliminated before entering Spark executor memory. Because reading and deserializing data is usually the most expensive part of a scan, filtering at the source saves I/O, network, and CPU, and it is one of the most impactful performance behaviors to understand.
How well it works depends on the source. For Parquet and ORC, each file is split into row groups with per-column min/max statistics. The reader checks the filter against these statistics and skips entire row groups whose range cannot match. For JDBC, the predicate becomes a SQL WHERE clause the database executes with its own indexes. Both are extremely effective. Row-based CSV and JSON are the weak case: no statistics, so the reader parses every row before filtering.
At the National Bank of India, storing transaction data in Parquet rather than CSV exports meant a high-value transfer query skipped most row groups using statistics, while the same query on CSV read every single row. The performance difference was thirty-fold. The rule is: store analytical data columnar so pushdown works, verify with explain() that PushedFilters shows your predicates, and avoid wrapping filter columns in UDFs or functions that prevent pushdown.
The senior perspective is that predicate pushdown, column pruning, and partition pruning together form the foundation of efficient Spark reads, and all three require columnar formats and careful schema design to work well.
Remember one sentence: predicate pushdown filters at the source using column statistics, and it only works well with columnar formats like Parquet and ORC that have those statistics.
---