1. The Problem
PySpark pipeline joined raw tasks with tenant dimensions. Narrow operations (.filter, .withColumn) ran fast, but .join() stalled on Stage 3 with massive Shuffle Read bytes and disk spill.
2. What I Initially Thought
3. What I Learned
Narrow transformations (filter, select) execute locally on worker nodes with zero network transfer. Pushing .filter() and .select() before .join() reduces the shuffle payload by up to 90%.
4. What I Built
Refactored PySpark transformation jobs applying early filtering and projection selection before wide join operations (pipelines/silver/clean_tasks.py).
# Narrow Transformations executed locally ON WORKER NODE before shuffle
filtered_tasks = (
spark.read.schema(TASK_SCHEMA).parquet("abfss://bronze/tasks/")
.filter(col("created_at") >= "2026-01-01")
.select("task_id", "tenant_id", "title", "updated_at")
)
# Wide Transformation with 90% smaller shuffle payload!
final_df = filtered_tasks.join(filtered_tenants, "tenant_id", "inner")5. The Experiment
Joined full un-filtered DataFrames containing 45 columns before applying filter logic.
Pushed .filter() and .select() projections before .join() wide transformation.
Spark Shuffle Read bytes reduced by order of magnitude; disk spill eliminated completely.
6. What Went Wrong
Attempted to eliminate shuffle by adding coalesce(1) before a groupBy, which collapsed all data onto a single executor core and caused an OOM crash.
7. Engineering Decision & Trade-offs
Mandated early predicate pushdown and explicit column pruning before wide operations across all PySpark pipelines to minimize network serialization payloads.
8. What I Would Do Differently in Production
In a production Lakehouse, ensure underlying Parquet tables are written with Z-Ordering or Bucketing on frequently joined keys to enable Bucket Sort Merge Joins with zero shuffle.
Questions I Can Now Answer Confidently in an Interview:
- What is the difference between narrow and wide transformations in Apache Spark?
- Why does a Spark Shuffle cause disk spill and network bottlenecks?
- How do early filtering and column pruning reduce PySpark job execution times?