Case 3 — Pipeline Reliability: Idempotent Airflow DAG Refactor

# 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) ```python # 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 | Criteria | Points | Passing Standard | |---|:---:|---| | **1. Execution Date Partition Scoping** | 25 pts | Replaces dynamic system clocks (`datetime.now()`) with deterministic Airflow execution context (`{{ ds }}`) enabling historical backfills. | | **2. Atomic Write / Idempotent Swap** | 25 pts | Implements atomic `DELETE-before-INSERT` or `MERGE INTO` scoped strictly to target date partition, guaranteeing zero duplicate records. | | **3. Worker Compute Offloading** | 25 pts | Eliminates worker memory bloat by delegating transformation compute to the warehouse database rather than pandas in worker RAM. | | **4. Automated Retry & Backfill Testing** | 25 pts | Provides automated test verifying that running multiple backfills for the same date yields identical row counts and checksums. |