Lab Type

Hands-On Debugging

Difficulty

Advanced Hands-On

Reward

700 XP · 🐞 Pipeline Debugger

Core Stack

PySpark · SQL · Data Reconciliation

Target Deliverable

Repaired Script & Root Cause Report

⚡ Hands-On Debugging Lab

DEV-012

Production Pipeline Debugging Challenge

A production revenue pipeline ran without throwing runtime errors, but the CFO dashboard was halted by the automated reconciliation audit: revenue is inflated by +34%, customer counts are wrong, and orders are mysteriously missing. Your job is to investigate the code, find the 5 defects, fix them, and prove zero variance.

Reproduce Failure

Investigate Discrepancy

Root Cause Analysis

Code Repair & Rerun

Reconcile & Report

Mission & Incident Context

Understand why this silent failure occurred and what the business impact is.

The Silent Failure Nightmare

The most dangerous bugs in data engineering are not crashes or syntax exceptions—they are silent logic errors where a pipeline finishes with exit code 0, but produces corrupt numbers.

Last night, the automated daily sales job ran in Databricks. When the downstream financial reconciliation check triggered at 04:15 UTC, it raised an immediate critical alert: the Gold table revenue did not match the audited source transactions. The CFO board dashboard is currently blank because the release gate halted publication.

Silent Duplication

Cartesian or 1-to-many joins silently inflate financial records without triggering errors.

Null Comparison Blindspots

In SQL and Spark, col == 0 evaluates to NULL (not TRUE) when col is null, silently dropping rows.

Type Coercion Pitfalls

CSV columns without explicit casting default to StringType, turning '+' into string concatenation.

Aggregation Grain Errors

Confusing count() with countDistinct() creates impossible customer retention statistics.

Starter Materials & Broken Code

Inspect the broken PySpark script, input CSVs, and the automated audit failure log.

📄 broken_pipeline.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count, countDistinct, to_date

spark = SparkSession.builder.appName("CorruptedDailyRevenuePipeline").getOrCreate()

# -----------------------------------------------------------------
# BUGGY PIPELINE SCRIPT
# This script executes without throwing syntax errors, but produces
# severe data corruption, doubled revenue numbers, and dropped records.
# -----------------------------------------------------------------

# Step 1: Ingest orders and promotions
orders_df = spark.read.csv("/data/orders_sample.csv", header=True)
promos_df = spark.read.csv("/data/promotions_sample.csv", header=True)

# Step 2: Filter recent orders
# [DEFECT A]: Notice the hard-coded date filter! What happens to backfills?
recent_orders = orders_df.filter(col("order_date") >= "2024-01-01")

# Step 3: Handle discount filtering
# [DEFECT B]: A business rule states "exclude orders with active discounts".
# What happens to orders where discount is NULL in the CSV?
discount_filtered = recent_orders.filter(col("discount") == 0)

# Step 4: Join promotions to enrich marketing channel
# [DEFECT C]: Look at the join condition below. Customer C101 has multiple promos!
# What happens to the row count of orders after this join?
enriched_orders = discount_filtered.join(
    promos_df,
    discount_filtered.customer_id == promos_df.customer_id,
    "inner"
)

# Step 5: Compute final total amount with shipping fee
# [DEFECT D]: Check the schema/data types of order_amount and shipping_fee!
# What does string concatenation do if not cast to numeric?
final_orders = enriched_orders.withColumn(
    "total_billed",
    col("order_amount") + col("shipping_fee")
)

# Step 6: Daily Executive Summary Aggregation
# [DEFECT E]: Check customer counting and revenue aggregation grain
daily_summary = final_orders.groupBy("order_date", "store_id").agg(
    sum("total_billed").alias("daily_revenue"),
    count("customer_id").alias("unique_customers"), # BUG: count vs countDistinct
    count("order_id").alias("order_count")
)

daily_summary.write.format("delta").mode("overwrite").save("/lakehouse/gold/daily_sales")

🔍 Investigation Clue: Contains 5 critical logic defects: hardcoded date filter, null drop bug, 1-to-many promo join duplication, string concatenation type error, and non-distinct customer aggregation.

Hands-On Debugging Workflow

Follow the professional Data Engineering incident troubleshooting lifecycle.

Reproduce the Error Locally

Run the broken pipeline with the provided orders_sample.csv and promotions_sample.csv.

▸Execute the script and print the intermediate DataFrame row counts after each transformation.

▸Verify the final Gold output revenue ($1,375.00) matches the discrepancy reported in the log.

Diagnose Defect 1 — Join Multiplication

Investigate why the row count increases after joining promotions_df.

▸Examine promotions_sample.csv for customer C101. How many active promo codes does C101 have?

▸An inner join on customer_id duplicates orders for C101, adding phantom revenue.

▸Fix: Either deduplicate promotions to 1 record per customer or join only on active date window.

Diagnose Defect 2 — Silent Null Dropping

Investigate why orders ORD-902 and ORD-906 vanished from the final output.

▸In orders_sample.csv, discount is empty (NULL). In Spark SQL, NULL == 0 evaluates to NULL, not True!

▸Orders with NULL discount were silently discarded, losing legitimate sales revenue.

▸Fix: Use coalesce(col('discount'), lit(0)) == 0 or (col('discount') == 0) | col('discount').isNull().

Diagnose Defect 3 — Data Types & String Concatenation

Check the schema of total_billed = order_amount + shipping_fee.

▸Because inferSchema wasn't used and no StructType was provided, CSV columns are strings.

▸In Python/Spark, '120.00' + '10.00' evaluates to '120.0010.00' instead of 130.00!

▸Fix: Explicitly cast columns using col('order_amount').cast('decimal(10,2)').

Diagnose Defect 4 & 5 — Aggregation & Hardcoded Dates

Correct customer metric counting and date windowing.

▸count('customer_id') counts total non-null rows, NOT unique customers. Replace with countDistinct().

▸Hard-coded date filter '2024-01-01' drops historical backfills. Parameterize with execution_date.

Expected Output & Reconciled Metrics

Your repaired pipeline must produce this exact financial baseline.

Target Reconciled Metrics

True Total Orders

7 Orders

ORD-901 through ORD-907

True Total Revenue

$1,025.50

Zero variance against source

True Unique Customers

5 Customers

C101, C102, C103, C104, C105

Variance

$0.00

Audit status: CLEARED

Expected Reconciled Store Output (2024-03-10)

order_date	store_id	daily_revenue	unique_customers	order_count
2024-03-10	STORE_01	$365.50	4	4 (ORD-901, 902, 904, 906)
2024-03-10	STORE_02	$660.00	3	3 (ORD-903, 905, 907)

Validation Checks & Test Script

Run these verification assertions to prove that all 5 defects have been eradicated.

Python Assertion Suite — Verify Repaired Pipeline

# 1. Assert Total Revenue matches source sum
total_rev = repaired_df.agg(sum("daily_revenue")).collect()[0][0]
assert float(total_rev) == 1025.50, f"Expected 1025.50, got {total_rev}"

# 2. Assert Order Count matches source clean orders
total_orders = repaired_df.agg(sum("order_count")).collect()[0][0]
assert int(total_orders) == 7, f"Expected 7 orders, got {total_orders}"

# 3. Assert No Duplicate Orders were created by Join
raw_ids = orders_df.select("order_id").distinct().count()
assert repaired_orders_df.count() <= raw_ids, "Join generated duplicate rows!"

print("✓ All validation assertions passed! Pipeline is certified clean.")

Evidence To Submit & Root Cause Report

Document your investigation findings using the standard Data Engineering Incident Review format.

Your submission must include the repaired Python code and a structured Root Cause Analysis (RCA) report:

☑

1. Repaired Source Code (repaired_pipeline.py)

Clean script with explicit schemas, safe join logic, null handling, and distinct customer aggregations.

☑

2. Execution Log Showing Zero Discrepancy

Terminal screenshot or text log showing the financial reconciliation check passing with $0.00 variance.

☑

3. Root Cause 1: Promotion Join Duplication

Explanation of why the promo join produced duplicates and how you restructured the relationship.

☑

4. Root Cause 2: Null Discount Drop

Explanation of why discount == 0 dropped nulls and your chosen coalesce/null-safe solution.

☑

5. Root Cause 3: String Concatenation Bug

Explanation of how untyped CSV inputs converted arithmetic addition into string concatenation.

☑

6. Preventive Engineering Safeguards

1 paragraph recommending automated pre-write reconciliation checks to prevent future silent corruption.

Common Mistakes To Avoid

Learn from these common debugging errors.

⚠️ Fixing the Symptom with dropDuplicates() at the End

Slapping .dropDuplicates() at the end of a broken join hides the root cause and can mask other join bugs. Fix the join condition itself.

⚠️ Replacing NULL with 0 Globally Without Context

Blindly filling all nulls across all columns with 0 can corrupt non-numeric fields or misrepresent optional business attributes.

⚠️ Assuming PySpark Auto-Casts Strings in withColumn

Unlike some SQL engines that coerce strings to numbers on '+', PySpark will happily concatenate strings if both columns are StringType.

Stretch / Bonus Objectives

Automate reconciliation and add regression tests.

⭐ Automated Pre-Commit Circuit Breaker

Write a PySpark assertion helper that automatically aborts the Delta table write if the output row count exceeds the input row count.

⭐ Pytest Unit Test Suite

Write isolated pytest functions testing each transformation step with mocked DataFrames to prevent regression.

Debugging Certification

Complete DEV-012 to Earn 700 XP & Pipeline Debugger Badge

Diagnose all 5 defects, write the repaired pipeline, pass all validation tests, and submit your Root Cause Analysis report.

🐞

700 XP

Pipeline Debugger