The Data Foundry

Built by Data with Pranjal

Public Beta

The Data Foundry is improving every week. If something feels unclear, broken, or missing, tell us and we will use it to improve the platform.

MixedIntermediateMixed LabFree

Yesterday's Sales Missing After Late Source Arrival

You are supporting the daily sales reporting pipeline for a retail company. Business users open the 9 AM revenue dashboard and see zero sales for yesterday, even though the source system clearly processed orders. The pipeline is orchestrated by Airflow, transformed with PySpark, and loaded into a cloud warehouse table used by the dashboard.

Production incident room

INC-2026-05-08-DAILY-SALES

You are the data engineer on-call. Business is waiting for a clear diagnosis, a safe fix, and a prevention plan.

Live incident

Severity

P1 - Executive sales dashboard incorrect

Reported by

Sales Operations

Reported at

2026-05-08 09:12 IST

Pipeline

daily_sales_to_bigquery

Owner

Data Platform On-call

Impact

Morning sales review is blocked and Finance cannot reconcile yesterday's daily revenue.

Expected business state

Yesterday's sales should show 2 paid orders worth 420.00

Current broken state

Dashboard shows 0 orders and 0.00 gross sales

Scenario context

The pipeline expected yesterday's sales export to be available before the 2 AM Airflow run. The source file arrived late, so the first run failed. A later retry picked up data, but the PySpark transformation filters by ingestion date instead of sales business date, so late-arriving rows for yesterday still do not land in the correct dashboard partition.

Business requirement

Diagnose why yesterday's sales are late, fix the PySpark transformation so sales are assigned to the correct business_date, and propose an Airflow scheduling change that waits for the required source partition before loading the warehouse. For executable validation, create a Spark DataFrame named daily_sales or fixed_daily_sales with business_date, order_count, and gross_sales. Do not write to the warehouse from the practice runner.

Production evidence board

Inspect the artifacts before touching code

This is how real incidents feel: business ticket, orchestration logs, source freshness, code review, and warehouse validation all point to the answer.

Business ticket

Incident

Sales Ops reports yesterday's sales are missing from the 9 AM dashboard.

Expected 2 paid orders totaling 420.00 for business_date=2026-05-07. Dashboard currently shows 0 orders and 0.00 gross sales.

Airflow run history

Orchestration

The 2 AM run failed waiting for the upstream source partition.

The source success marker for dt=2026-05-07 was not available before the sensor timeout. A retry started after the file arrived, but the transform still loaded zero rows.

Source landing check

Storage

The source file did arrive, but after the scheduled run window.

The sales file for source_file_date=2026-05-07 landed at 03:21. Valid records have sale_ts_utc on 2026-05-07 but ingested_at on 2026-05-08.

PySpark transform review

Code

The job derives business_date from ingested_at instead of the sale timestamp.

Late-arriving records are filtered out because ingest_date becomes 2026-05-08 while the dashboard partition being rebuilt is 2026-05-07.

Warehouse validation

Data Quality

The warehouse partition was overwritten with an empty result.

The load step completed, but records_loaded=0 should have failed the data quality gate for a paid-sales business date.

Step 1: diagnose like on-call

What is the most likely root cause?

Choose your diagnosis before writing the fix. This builds the production judgment interviewers are actually testing.

Sample production data

Use these small tables to reason about the bug before writing the fix. This data is used for the visible sample check when you click Run.

raw_sales_events

order_idcustomer_idsale_ts_utcamountstatussource_file_dateingested_at
70011012026-05-07 10:15:00250PAID2026-05-072026-05-08 03:22:10
70021022026-05-07 18:40:00170PAID2026-05-072026-05-08 03:25:44
70031032026-05-08 01:05:0090PAID2026-05-082026-05-08 03:26:01
70041042026-05-07 12:30:0075CANCELLED2026-05-072026-05-08 03:23:00

airflow_runs

dag_idlogical_datescheduled_atstatusrecords_loaded
daily_sales_to_bigquery2026-05-072026-05-08 02:00:00failed0
daily_sales_to_bigquery2026-05-072026-05-08 03:45:00success0

warehouse_daily_sales_expected

business_dateorder_countgross_sales
2026-05-072420

Schema

raw_sales_events(order_id, customer_id, sale_ts_utc, amount, status, source_file_date, ingested_at)
warehouse_daily_sales(business_date, order_count, gross_sales, loaded_at)

Sample input

The Airflow logical date is 2026-05-08 and the dashboard needs sales for business_date=2026-05-07. Two valid sales records arrived after the scheduled 2 AM run, so they have ingested_at on 2026-05-08 but sale_ts_utc on 2026-05-07.

Broken logic / code

from pyspark.sql import functions as F

run_date = "2026-05-07"

raw_sales = spark.read.parquet(raw_sales_path)

# Broken: this assumes records for the sales day are ingested on the same date.
# Late files arrive on 2026-05-08, so yesterday's valid sales are filtered out.
daily_sales = (
    raw_sales
    .withColumn("ingest_date", F.to_date("ingested_at"))
    .filter(F.col("ingest_date") == F.lit(run_date))
    .filter(F.col("status") == F.lit("PAID"))
    .groupBy(F.col("ingest_date").alias("business_date"))
    .agg(
        F.countDistinct("order_id").alias("order_count"),
        F.sum("amount").alias("gross_sales")
    )
)

daily_sales.write.mode("overwrite").insertInto("warehouse.daily_sales")

Logs / error

[2026-05-08 02:00:14] {sales_sensor.py:88} INFO - Waiting for gs://retail-landing/sales/dt=2026-05-07/_SUCCESS
[2026-05-08 02:15:14] {sales_sensor.py:104} ERROR - Source partition dt=2026-05-07 not available before timeout
[2026-05-08 02:15:15] {taskinstance.py:1345} ERROR - Task failed: source data delay
[2026-05-08 03:21:58] {gcs.py:61} INFO - File arrived: gs://retail-landing/sales/dt=2026-05-07/part-0001.json
[2026-05-08 03:45:08] {spark_submit.py:512} INFO - Retry started for logical_date=2026-05-07
[2026-05-08 03:47:31] {spark_submit.py:540} INFO - PySpark output rows=0 for business_date=2026-05-07
[2026-05-08 03:48:02] {bigquery.py:220} WARNING - Loaded 0 rows into warehouse.daily_sales partition 2026-05-07

Actual output

Dashboard for 2026-05-07 shows order_count=0 and gross_sales=0 because the transformation filters on ingest_date instead of sale business date.

Expected output / expected logic

business_date | order_count | gross_sales
2026-05-07 | 2 | 420.00

Your attempt

Write the corrected PySpark transformation

Create a DataFrame named daily_sales or fixed_daily_sales. Run checks the visible sample; Submit fix runs hidden late-data edge cases.

Saved

Interview-style explanation

Now explain your solution as if you are in an interview: symptom, root cause, fix, edge cases, trade-offs, monitoring, and prevention.