Skip to main content

04.3 Select, Filter & Column Expressions

PySpark select and filter architecture

Architect goal: Express transformations as Spark-native column expressions and understand projection pruning, predicate pushdown, NULL semantics, and plan optimization.

1. The mental model​

A statement such as:

df.select("customer_id", "revenue").filter(F.col("revenue") > 1000)

does not mean Python loops through every row.

PySpark expressions ↓ Logical plan ↓ Analysis + optimization ↓ Physical plan ↓ Distributed execution

Current PySpark supports DataFrame.filter with either a Column condition or SQL-expression string; where() is an alias for filter(). citeturn350977search1turn350977search3

2. Select — projection​

from pyspark.sql import functions as F

result = df.select( "customer_id", "revenue", (F.col("revenue") * 1.18).alias("revenue_with_tax") )

A projection defines which columns and expressions should survive to the next stage.

If a source has 150 columns but the pipeline needs 6, carrying all 150 can increase scan requirements, memory pressure, serialization, network transfer, and downstream shuffle payload. Spark's optimizer can eliminate unnecessary columns when the plan permits it. citeturn350977search0turn350977search2

3. Explicit projection vs select all​

Exploration:

df.select("*")

Production contract:

df.select( "order_id", "customer_id", "order_date", "amount" )

Architect rule: use star projection intentionally for pass-through use cases; prefer explicit columns for stable Silver, Gold, and semantic interfaces.

4. Column expressions​

A Spark Column represents an expression evaluated by the Spark engine.

Examples:

F.col("amount") F.col("amount") * 1.18 F.upper(F.col("country")) F.coalesce(F.col("amount"), F.lit(0)) F.when(F.col("amount") > 1000, "HIGH").otherwise("LOW")

Use F.lit() for literal values when building expressions.

5. Derived columns and aliases​

result = df.select( F.col("customer_id").alias("customer"), F.round("revenue", 2).alias("revenue_rounded"), F.when(F.col("revenue") >= 5000, "VIP") .when(F.col("revenue") >= 1000, "Premium") .otherwise("Standard") .alias("segment") )

For several derived columns, one clear select can be easier to review than a long sequence of withColumn calls.

6. Filter — row selection​

df.filter(F.col("country") == "IN")

Equivalent:

df.where(F.col("country") == "IN")

SQL-expression form:

df.filter("country = 'IN'")

The current API accepts either a Column or a SQL expression string. citeturn350977search3

7. Boolean logic​

Use Spark's operators:

filtered = df.filter( (F.col("country") == "IN") & (F.col("revenue") > 1000) )

OR:

filtered = df.filter( (F.col("country") == "IN") | (F.col("country") == "SG") )

NOT:

filtered = df.filter(~F.col("is_cancelled"))

Do not use Python and/or/not with Spark Column expressions. Parenthesize individual comparisons.

8. Common predicates​

Equality:

F.col("status") == "OPEN"

Inequality:

F.col("amount") != 0

Range:

F.col("amount").between(100, 1000)

Membership:

F.col("country").isin("IN", "SG", "AE")

Current PySpark supports predicates including isin, between, isNull, isNotNull, like, and contains. citeturn350977search1

9. NULL-safe filtering​

Use explicit NULL predicates:

df.filter(F.col("customer_id").isNull()) df.filter(F.col("customer_id").isNotNull())

For equality where two NULL values should be treated as equal, use null-safe equality:

F.col("a").eqNullSafe(F.col("b"))

Three-valued logic​

SQL filtering has:

TRUE FALSE UNKNOWN

For example, NULL > 1000 evaluates to UNKNOWN. A filter retains rows whose predicate evaluates to TRUE.

This is why NULL handling is a correctness concern, not merely a syntax detail.

10. String filtering​

df.filter(F.col("name").startswith("A")) df.filter(F.col("name").contains("tech")) df.filter(F.col("name").like("Data%")) df.filter(F.col("email").rlike(r"^[^@]+@[^@]+$"))

Regex can be CPU-intensive at scale. Prefer simpler predicates when they express the same requirement.

11. Date and timestamp filtering​

Use semantic temporal types instead of comparing arbitrary strings.

df = df.withColumn( "order_date", F.to_date("order_date") )

result = df.filter( F.col("order_date").between( F.lit("2026-01-01").cast("date"), F.lit("2026-01-31").cast("date") ) )

For timestamps, make timezone assumptions explicit.

12. Predicate pushdown​

Conceptually:

Without pushdown: Storage → read many rows → Spark Filter → fewer rows

With pushdown: Storage → supported predicate → fewer rows read → Spark

Predicate pushdown means a filter can be evaluated closer to the source when the source supports it.

Important: writing filter() early does not guarantee physical pushdown. The optimizer and source connector determine what can actually be pushed down.

Verify with explain() and runtime metrics.

13. Projection pruning​

Projection pruning is the column equivalent:

Source: 100 columns ↓ Query needs 5 ↓ Plan requests only required columns

For columnar formats such as Parquet, reading fewer columns can materially reduce I/O.

Spark's performance documentation describes optimizer and columnar execution techniques for DataFrame and SQL workloads. citeturn350977search0

14. Filter ordering — the nuanced answer​

The beginner rule is “always put filters first.”

The production rule is:

Express selective, semantically correct filters early, then verify the optimized plan.

Spark can reorder and combine compatible operations. Therefore, do not micro-optimize source-code ordering without checking the optimized plan.

15. Select + filter + expressions​

A realistic pipeline:

result = ( df .select( "order_id", "customer_id", "country", "amount", "status" ) .filter( (F.col("country") == "IN") & (F.col("status") == "COMPLETED") & (F.col("amount") > 1000) ) .select( "order_id", "customer_id", F.round("amount", 2).alias("amount") ) )

The intent is:

read → restrict → project → derive.

16. Conditional expressions with when​

result = df.withColumn( "risk_band", F.when(F.col("amount") >= 100000, "HIGH") .when(F.col("amount") >= 25000, "MEDIUM") .otherwise("LOW") )

Keep complex business rules readable and testable.

17. Reusable expressions​

Avoid duplicating complex business logic:

high_value = F.col("amount") >= 10000 india = F.col("country") == "IN"

result = df.filter(high_value & india)

Reusable expressions improve readability and make unit testing easier.

18. Nested-field projection​

For nested data:

df.select( "customer_id", "address.city", "address.country" )

Or:

df.select( F.col("address.city").alias("city") )

Project only the nested fields required by the consumer.

19. selectExpr​

SQL-style projection is also available:

result = df.selectExpr( "customer_id", "amount * 1.18 AS amount_with_tax", "upper(country) AS country_upper" )

Choose select() or selectExpr() according to team readability and conventions.

20. Explain the plan​

Use:

result.explain() result.explain("formatted") result.explain("cost")

Look for:

  • Required columns
  • Filter placement
  • Scan operators
  • Exchanges
  • Join operators
  • Estimated statistics
  • Unexpected projections

Spark documentation recommends inspecting cost estimates and runtime statistics when diagnosing plans. citeturn350977search0

21. Production example — Bronze to Silver​

from pyspark.sql import functions as F

orders = spark.table("main.bronze.orders")

silver = ( orders .select( "order_id", "customer_id", "country", "order_date", "amount", "status" ) .filter(F.col("order_id").isNotNull()) .filter(F.col("status").isin("COMPLETED", "SHIPPED")) .filter(F.col("amount") > 0) .withColumn("order_date", F.to_date("order_date")) .withColumn( "amount", F.col("amount").cast("decimal(18,2)") ) .select( "order_id", "customer_id", "country", "order_date", "amount", "status" ) )

silver.explain("formatted")

Review whether unused columns disappeared, filters are applied effectively, and unexpected exchanges or casts appear.

22. Anti-patterns​

Python row loops​

for row in df.collect(): ...

This moves data to Python and defeats distributed processing for large datasets.

Excessive select("*")​

Wide intermediate datasets increase data movement and weaken explicit contracts.

Python functions when Spark expressions exist​

Prefer Spark-native expressions so Spark can optimize the computation.

Regex everywhere​

Use regex only when necessary.

Assuming pushdown​

A filter in source code is not proof of source-side pushdown.

Assuming source-code order equals physical order​

The optimizer can rewrite the logical plan.

23. Hands-on lab — expression engineering​

Create an orders DataFrame with:

  • order_id
  • customer_id
  • country
  • amount
  • status
  • order_date

Build a result that:

  1. Keeps only required columns.
  2. Selects Indian orders.
  3. Keeps COMPLETED or SHIPPED.
  4. Removes NULL customer_id.
  5. Keeps amounts between 1,000 and 100,000.
  6. Creates amount_with_tax.
  7. Creates a value_band.
  8. Produces a clean analytical contract.

Then run:

result.explain("formatted") result.show()

Create the same result once with Column expressions and once with SQL expressions. Compare readability and physical plans.

24. Hands-on lab — NULL semantics​

Use this conceptual dataset:

id | amount | country ---+--------+-------- 1 | 100 | IN 2 | NULL | IN 3 | 200 | NULL 4 | NULL | NULL

Predict the output of:

df.filter(F.col("amount") > 100) df.filter(F.col("amount").isNull()) df.filter(F.col("country").isNotNull())

Then execute and explain why NULL produces UNKNOWN rather than TRUE/FALSE.

25. Performance lab — projection and filtering​

Compare:

wide_result = wide_df.select("*").filter( F.col("country") == "IN" )

with:

narrow_result = wide_df.select( "id", "country", "amount" ).filter( F.col("country") == "IN" )

Inspect both plans and, on a meaningful dataset, compare scan and runtime metrics.

The objective is not to assume a speedup; it is to learn how to validate the effect.

26. Architect decision framework​

QuestionWhy
Which columns are actually required?Projection pruning
Which rows are required?Reduce work
Can the predicate be pushed down?Reduce I/O
Are NULL semantics correct?Correctness
Are expressions Spark-native?Distributed execution
Is regex really required?CPU efficiency
Does the plan contain unexpected operators?Performance
Is the output a stable contract?Governance

27. Interview questions​

Fundamentals​

  1. What is a Spark Column expression?
  2. select() vs selectExpr()?
  3. filter() vs where()?
  4. How do you combine conditions?
  5. Why not Python and/or/not with Spark Columns?
  6. How do you filter NULLs?

Performance​

  1. What is predicate pushdown?
  2. What is projection pruning?
  3. Does putting a filter first guarantee pushdown?
  4. How do you verify pushdown?
  5. Why avoid unused columns?
  6. How can filtering reduce downstream shuffle volume?

Architect​

  1. How would you filter a 20 TB Silver pipeline?
  2. Logical optimization vs physical pushdown?
  3. How would you diagnose a selective filter that does not improve runtime?
  4. How would you design reusable business-rule expressions?
  5. How would you protect consumers from accidental select("*") changes?

28. Cheat sheet​

from pyspark.sql import functions as F

df.select("id", "amount") df.select(F.col("amount").alias("revenue"))

df.filter(F.col("amount") > 100) df.where(F.col("country") == "IN")

(F.col("amount") > 100) & (F.col("country") == "IN") (F.col("country") == "IN") | (F.col("country") == "SG") ~F.col("is_cancelled")

F.col("country").isin("IN", "SG") F.col("amount").between(100, 1000)

F.col("amount").isNull() F.col("amount").isNotNull() F.col("a").eqNullSafe(F.col("b"))

F.lit(100) F.when(F.col("amount") > 1000, "HIGH").otherwise("LOW")

F.col("address.city")

df.selectExpr( "id", "amount * 1.18 AS amount_with_tax" )

df.explain("formatted")

29. Key takeaways​

  1. Select and filter are Spark expressions, not Python row loops.
  2. Projection pruning reduces unnecessary data movement.
  3. Predicate pushdown can reduce source-side work when supported.
  4. NULL requires three-valued SQL semantics.
  5. Use &, |, and ~ with Spark Columns.
  6. Do not assume pushdown — verify the plan.
  7. Use explicit projections for stable production contracts.
  8. Performance tuning starts with understanding the optimized plan.

Next​

04.4 PySpark Aggregations

We will cover groupBy, aggregation functions, partial aggregation, shuffle behavior, skew, distinct counts, approximate aggregations, pivoting, and production aggregation patterns.

Official references​