Case 3 — Pipeline Reliability: Idempotent Airflow DAG Refactor

Executive Context & Business Prompt

You have joined RetailPulse, an omnichannel retail platform processing 10 million transactions daily. The daily merchant payout pipeline runs on Apache Airflow.

Last Tuesday, a transient network timeout occurred while writing merchant payouts to Snowflake. Airflow retried the task 3 times according to its retry policy. Because the task executed a raw INSERT INTO merchant_payouts SELECT ..., the payout batch was inserted 4 times. Merchant bank transfers were calculated at $4 imes$ their actual balance, causing a $2.4M financial incident before the treasury team caught it.

VP of Data Platform Prompt:

"Every single pipeline we run must be completely idempotent. Re-running a task 1 time or 100 times for the same execution date must produce the exact same outcome without duplicates or corrupted aggregates. Refactor this broken merchant payout DAG using modern TaskFlow API patterns, atomic partition swaps, and automated validation."

Explicit Portfolio Deliverables

1. Refactored Airflow DAG (merchant_payout_dag.py):
   • Rewritten with TaskFlow API (@dag, @task).
   • Replaces datetime.utcnow() with Airflow execution partition context (ds, data_interval_start).
   • Implements atomic partition writes (e.g. DELETE ... WHERE payout_date = '{{ ds }}' followed by INSERT, or atomic MERGE INTO).
2. Task Memory Safety Audit:
   • Offloads heavy aggregation to the warehouse rather than pulling large PySpark/pandas DataFrames into worker RAM.
3. Automated Idempotency Verification Test (test_idempotency.py):
   • Author a test script that executes the DAG task 3 consecutive times for execution date 2026-09-15 and asserts that SELECT COUNT(*) on the target partition remains strictly identical.

Starter Code Scaffold (Broken Legacy Pipeline)

# BROKEN LEGACY DAG — DO NOT USE IN PRODUCTION
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import pandas as pd

def compute_payouts():
    # BUG 1: Uses non-deterministic system clock instead of execution date!
    today = datetime.now().strftime('%Y-%m-%d')
    
    # BUG 2: Pulls millions of rows into worker memory!
    df = pd.read_sql(f"SELECT merchant_id, SUM(amount) as total FROM raw_sales WHERE sale_date = '{today}' GROUP BY 1", con=db_engine)
    
    # BUG 3: Plain append causes duplicate rows on any retry!
    df.to_sql('merchant_payouts', con=warehouse_engine, if_exists='append', index=False)

dag = DAG('broken_payouts', start_date=datetime(2023, 1, 1), schedule_interval='@daily')
task = PythonOperator(task_id='payout_task', python_callable=compute_payouts, dag=dag)

100-Point Grading Rubric