# Three Hours for Yesterday's Numbers > Three terabytes scanned. One day changed. Canonical URL: Domain: PySpark · Difficulty: medium · Seniority: L5 ## Problem A nightly job refreshes `daily_category_sales`, a summary of one day's sales broken down by product category, but it rebuilds from the entire multi-year `transactions` history on every run even though only the most recent day is ever new. For the latest day present in `transactions`, total each product category's revenue and units sold, and make the job read only that day's data rather than the whole table. ## Worked solution and explanation ### What's really being tested Strip the retail dressing away and this is a partition-pruning problem. The job's real defect is one empty bracket in the physical plan: PartitionFilters: []. The transactions fact table is split into 1,460 date directories, and every night this job opens all of them, 2.94 TB and 4.2 billion rows, to rebuild a summary where only the latest day actually changed. Anyone can write the join and the group-by. What separates candidates is making Catalyst push a filter on transaction_date down to the FileScan so it touches one directory instead of all 1,460. Miss it and you burn three hours and 48 GB of shuffle to produce a handful of category totals you could have read in a minute. > **The whole fix is one Column filter, placed early** > > A filter prunes partitions only if Catalyst can evaluate it against the partition column at plan time. Put F.col('transaction_date') == the target day before the join and the aggregation, and the plan flips from PartitionFilters: [] to PartitionFilters: [(transaction_date = 2026-12-28)]. Everything downstream shrinks with the scan. ### Read the plan, not the code **Before: full scan** PartitionFilters: []. FileScan reads all 1,460 directories: 2.94 TB, 4.2B rows. The join and group-by shuffle write 48.2 GB. Executors sit in GC for roughly 95s each. Wall clock: 183 minutes, SLA blown by 2h 18m. **After: pruned** PartitionFilters: [(transaction_date = 2026-12-28)]. FileScan reads 1 directory: 1.6 GB, 2.3M rows. products broadcasts, so shuffle drops to about 12 MB. GC falls to about 8s. Wall clock: about 13 minutes, comfortably inside the 45-minute SLA. #### Step 1: Prune to the one partition that changed The source is partitioned by transaction_date, so a filter on that column lets Spark skip directories entirely, before it opens a single Parquet file. Filter with F.col('transaction_date') == the latest day. Prove it worked by reading the plan: you want PartitionFilters to name transaction_date, not an empty list. #### Step 2: Attach category, then aggregate category lives in products, not in transactions, so join to bring it in. products is tiny next to the fact table, so let Spark broadcast it and avoid shuffling the fact side. Then group by category and sum total_amount for revenue and quantity for units. Because the scan already returned one day, this runs over 2.3M rows, not 4.2B. > **Finding the latest date can undo the pruning** > > Deriving the target day with a full max('transaction_date') aggregation reads the whole table, the exact scan you were trying to avoid. It is fine as a one-off, but a nightly job already knows its date: pass yesterday's partition as a literal so the FileScan prunes and never touches the history. Also watch the join: an inner join silently drops transactions with a null or unmatched product_id, so decide whether those belong in the totals. **Pruned incremental daily aggregation** ```python from pyspark.sql import SparkSession, functions as F spark = SparkSession.builder.getOrCreate() # Only the latest day is new. The source is partitioned by transaction_date, # so a filter on that column lets Catalyst push a PartitionFilter to the # FileScan and read one date directory instead of all 1,460. latest_day = ( spark.table("transactions") .agg(F.max("transaction_date").alias("d")) .first()["d"] ) sales = spark.table("transactions").filter(F.col("transaction_date") == latest_day) daily = ( sales.join(spark.table("products"), "product_id") .groupBy("category") .agg( F.sum("total_amount").alias("total_revenue"), F.sum("quantity").alias("total_units"), ) .orderBy(F.col("total_revenue").desc(), F.col("category")) ) daily.write.mode("overwrite").saveAsTable("daily_category_sales") ``` *One partition read, broadcast join for category, sum the 2.3M rows that changed.* > **The scan is the whole bill** > > 2.94 TB down to 1.6 GB is a roughly 1,800x cut in bytes read, and everything downstream shrinks with it: shuffle 48.2 GB to about 12 MB, GC about 95s to 8s per executor, runtime 183 min to about 13. No extra executors, no tuning of shuffle partitions. You did not make the cluster faster, you stopped asking it to read four years of history to refresh one day. > **The tell they watch for** > > A senior engineer opens the physical plan before touching the code. They point at PartitionFilters: [] and say the scan is unfiltered, that is the entire problem, rather than guessing at executor memory or shuffle partitions. Naming the plan node, and knowing that a max() lookup can quietly reintroduce the scan, is the signal that they understand where the time actually goes. ## Common follow-up questions - The source is not partitioned by date, only clustered on it. How does your fix change? _(Tests that pruning depends on physical layout, not just a WHERE clause.)_ - You need to backfill 90 days after a schema change. How do you run this without scanning all 1,460 partitions? _(Incremental thinking: a bounded date range still prunes to 90 directories.)_ - Two late-arriving transactions land for a date you already processed. How do you correct that day without a full rebuild? _(Idempotent overwrite of a single partition.)_ ## Related - [All practice problems](https://datadriven.io/problems) - [Mock interview mode](https://datadriven.io/interview/three_hours_for_yesterday_s_numbers) - [Data Engineering Interview Prep Guide](https://datadriven.io/data-engineer-interview-prep) - [Daily Challenge](https://datadriven.io/daily) --- Source: DataDriven (https://datadriven.io). DataDriven is the data engineering interview community. Live code execution in SQL, Python, and Spark sandboxes. Every feature is open to every member.