Airflow Real-World Data Pipeline Project

This topic builds a complete end-to-end data pipeline from scratch. You will apply every concept from this course — DAGs, operators, XComs, connections, error handling, monitoring, and the TaskFlow API — in one working project.

Project Brief: Daily E-Commerce Sales Pipeline

A fictional online store, ShopEasy, needs an automated daily pipeline that:

1. Extracts yesterday's sales records from a PostgreSQL database
2. Calls an external currency conversion API to normalize amounts to USD
3. Transforms the data — cleans nulls, computes totals, flags outliers
4. Loads the clean data into a reporting table
5. Generates a summary report and uploads it to S3
6. Sends a Slack notification with the key numbers

Pipeline Architecture Diagram

                        [start]
                           |
             ┌─────────────┴─────────────┐
             ▼                           ▼
    [extract_sales]             [fetch_fx_rates]
    (PostgreSQL → Python)       (HTTP API → Python)
             \                           /
              └──────────┬──────────────┘
                         ▼
                  [transform_data]
                  (clean + enrich)
                         |
                         ▼
                   [load_to_db]
                 (write to reporting table)
                         |
                         ▼
                [upload_report_to_s3]
                         |
                         ▼
                  [notify_slack]
                         |
                        [end]

Project Folder Structure

~/airflow/
├── dags/
│   └── shopeasy_daily_pipeline.py   ← main DAG file
├── plugins/
│   └── hooks/
│       └── slack_alert_hook.py      ← custom Slack hook
└── config/
    └── pipeline_config.json         ← pipeline settings

Step 1: Create the Config File

Keep environment-specific settings out of code:

// config/pipeline_config.json
{
  "source_db_conn_id": "shopeasy_postgres",
  "reporting_db_conn_id": "shopeasy_reporting",
  "s3_conn_id": "aws_default",
  "s3_bucket": "shopeasy-reports",
  "slack_conn_id": "slack_webhook",
  "currency_api_conn_id": "exchange_rate_api",
  "base_currency": "USD",
  "outlier_threshold_usd": 10000
}

Step 2: Set Up Connections in Airflow UI

Go to Admin → Connections and create these five connections:

Step 3: The Full DAG File

# dags/shopeasy_daily_pipeline.py

import json
import logging
from datetime import datetime, timedelta
from pathlib import Path

from airflow.decorators import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.http.hooks.http import HttpHook
from airflow.models import Variable

logger = logging.getLogger(__name__)

# ── Load pipeline config ───────────────────────────────────────
CONFIG_PATH = Path(__file__).parent.parent / "config" / "pipeline_config.json"
CFG = json.loads(CONFIG_PATH.read_text())

# ── Default task settings ──────────────────────────────────────
DEFAULT_ARGS = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "email_on_failure": True,
    "email": Variable.get("alert_email", default_var="data@shopeasy.com"),
}

# ── DAG Definition ─────────────────────────────────────────────
@dag(
    dag_id="shopeasy_daily_pipeline",
    description="Extract, transform, and load daily sales to reporting",
    start_date=datetime(2024, 1, 1),
    schedule="0 6 * * *",     # every day at 6:00 AM
    catchup=False,
    default_args=DEFAULT_ARGS,
    tags=["shopeasy", "sales", "daily"],
)
def shopeasy_daily_pipeline():

    # ── TASK 1: Extract sales from PostgreSQL ──────────────────
    @task(task_id="extract_sales")
    def extract_sales(**context):
        execution_date = context["ds"]   # "YYYY-MM-DD" of the logical date
        logger.info(f"Extracting sales for date: {execution_date}")

        hook = PostgresHook(postgres_conn_id=CFG["source_db_conn_id"])
        sql = """
            SELECT
                order_id,
                customer_id,
                currency,
                amount,
                product_category,
                created_at
            FROM orders
            WHERE DATE(created_at) = %s
              AND status = 'completed';
        """
        records = hook.get_records(sql, parameters=[execution_date])
        logger.info(f"Extracted {len(records)} completed orders")

        # Convert to list of dicts for easy downstream use
        columns = ["order_id","customer_id","currency","amount","product_category","created_at"]
        sales = [dict(zip(columns, row)) for row in records]
        return sales   # auto-pushed to XCom


    # ── TASK 2: Fetch currency exchange rates ──────────────────
    @task(task_id="fetch_fx_rates")
    def fetch_fx_rates(**context):
        logger.info("Fetching exchange rates from API")
        hook = HttpHook(http_conn_id=CFG["currency_api_conn_id"], method="GET")
        response = hook.run(endpoint=f"/latest?base={CFG['base_currency']}")
        rates = response.json().get("rates", {})
        logger.info(f"Fetched rates for {len(rates)} currencies")
        return rates   # e.g. {"GBP": 0.79, "EUR": 0.92, "INR": 83.1}


    # ── TASK 3: Transform — clean, convert, flag outliers ──────
    @task(task_id="transform_data")
    def transform_data(sales: list, fx_rates: dict):
        logger.info(f"Transforming {len(sales)} records")
        threshold = CFG["outlier_threshold_usd"]
        clean_rows = []
        skipped = 0

        for row in sales:
            # Skip rows with missing critical fields
            if not row.get("order_id") or row.get("amount") is None:
                skipped += 1
                continue

            currency = row["currency"].upper()
            amount   = float(row["amount"])

            # Convert to USD
            if currency == "USD":
                amount_usd = amount
            elif currency in fx_rates:
                amount_usd = round(amount / fx_rates[currency], 2)
            else:
                logger.warning(f"Unknown currency {currency} for order {row['order_id']} — skipping")
                skipped += 1
                continue

            clean_rows.append({
                "order_id":         row["order_id"],
                "customer_id":      row["customer_id"],
                "product_category": row["product_category"] or "Unknown",
                "original_amount":  amount,
                "original_currency":currency,
                "amount_usd":       amount_usd,
                "is_outlier":       amount_usd > threshold,
                "created_at":       str(row["created_at"]),
            })

        total_usd     = sum(r["amount_usd"] for r in clean_rows)
        outlier_count = sum(1 for r in clean_rows if r["is_outlier"])
        logger.info(f"Transformed {len(clean_rows)} rows | Skipped {skipped} | Total USD: {total_usd:.2f} | Outliers: {outlier_count}")

        return {
            "rows":          clean_rows,
            "total_usd":     total_usd,
            "order_count":   len(clean_rows),
            "outlier_count": outlier_count,
            "skipped":       skipped,
        }


    # ── TASK 4: Load clean data into reporting database ────────
    @task(task_id="load_to_db")
    def load_to_db(transformed: dict, **context):
        rows = transformed["rows"]
        execution_date = context["ds"]
        logger.info(f"Loading {len(rows)} rows for {execution_date}")

        hook = PostgresHook(postgres_conn_id=CFG["reporting_db_conn_id"])

        # Create table if it does not exist
        hook.run("""
            CREATE TABLE IF NOT EXISTS daily_sales_usd (
                order_id          VARCHAR PRIMARY KEY,
                customer_id       VARCHAR,
                product_category  VARCHAR,
                original_amount   NUMERIC,
                original_currency VARCHAR,
                amount_usd        NUMERIC,
                is_outlier        BOOLEAN,
                created_at        TIMESTAMP,
                loaded_at         TIMESTAMP DEFAULT NOW()
            );
        """)

        # Delete existing rows for this date (idempotent load)
        hook.run(
            "DELETE FROM daily_sales_usd WHERE DATE(created_at) = %s;",
            parameters=[execution_date]
        )

        # Bulk insert
        hook.insert_rows(
            table="daily_sales_usd",
            rows=[
                (
                    r["order_id"], r["customer_id"], r["product_category"],
                    r["original_amount"], r["original_currency"],
                    r["amount_usd"], r["is_outlier"], r["created_at"]
                )
                for r in rows
            ],
            target_fields=[
                "order_id","customer_id","product_category",
                "original_amount","original_currency",
                "amount_usd","is_outlier","created_at"
            ],
        )
        logger.info("Load complete")
        return True


    # ── TASK 5: Generate report CSV and upload to S3 ──────────
    @task(task_id="upload_report_to_s3")
    def upload_report_to_s3(transformed: dict, **context):
        import csv, io
        execution_date = context["ds"]
        rows = transformed["rows"]

        # Build CSV in memory
        buffer = io.StringIO()
        writer = csv.DictWriter(buffer, fieldnames=rows[0].keys())
        writer.writeheader()
        writer.writerows(rows)
        csv_bytes = buffer.getvalue().encode("utf-8")

        # Upload to S3
        s3_key = f"daily-reports/{execution_date}/sales_report.csv"
        s3_hook = S3Hook(aws_conn_id=CFG["s3_conn_id"])
        s3_hook.load_bytes(
            bytes_data=csv_bytes,
            key=s3_key,
            bucket_name=CFG["s3_bucket"],
            replace=True,
        )
        s3_path = f"s3://{CFG['s3_bucket']}/{s3_key}"
        logger.info(f"Report uploaded to {s3_path}")
        return s3_path


    # ── TASK 6: Send Slack notification ───────────────────────
    @task(task_id="notify_slack")
    def notify_slack(transformed: dict, s3_path: str, **context):
        execution_date = context["ds"]
        hook = HttpHook(http_conn_id=CFG["slack_conn_id"], method="POST")

        message = (
            f":white_check_mark: *ShopEasy Daily Pipeline — {execution_date}*\n"
            f"• Orders processed: {transformed['order_count']}\n"
            f"• Total revenue (USD): ${transformed['total_usd']:,.2f}\n"
            f"• Outlier orders: {transformed['outlier_count']}\n"
            f"• Skipped rows: {transformed['skipped']}\n"
            f"• Report: {s3_path}"
        )

        hook.run(
            endpoint="",
            data=json.dumps({"text": message}),
            headers={"Content-Type": "application/json"},
        )
        logger.info("Slack notification sent")


    # ── Wire up the task dependencies ─────────────────────────
    sales      = extract_sales()
    fx_rates   = fetch_fx_rates()
    transformed = transform_data(sales, fx_rates)   # depends on both extract tasks
    loaded     = load_to_db(transformed)
    s3_path    = upload_report_to_s3(transformed)
    loaded >> s3_path                                # S3 upload waits for DB load
    notify_slack(transformed, s3_path)


shopeasy_daily_pipeline()

Data Flow Walkthrough

6:00 AM — Scheduler triggers the DAG
   │
   ├── [extract_sales] runs
   │     Queries PostgreSQL for yesterday's completed orders
   │     Pushes list of dicts to XCom
   │
   ├── [fetch_fx_rates] runs (parallel to extract_sales)
   │     Calls currency API
   │     Pushes rate dict to XCom
   │
   ▼
[transform_data] runs (after both above complete)
   │   Pulls sales list + fx_rates from XCom automatically
   │   Converts currencies → USD
   │   Flags outliers above $10,000
   │   Pushes summary + clean rows to XCom
   │
   ▼
[load_to_db] runs
   │   Reads clean rows from XCom
   │   Deletes today's existing rows (idempotent)
   │   Bulk-inserts new rows into daily_sales_usd table
   │
   ▼
[upload_report_to_s3] runs (after load completes)
   │   Builds CSV from clean rows
   │   Uploads to s3://shopeasy-reports/daily-reports/YYYY-MM-DD/sales_report.csv
   │   Pushes S3 path to XCom
   │
   ▼
[notify_slack] runs
   │   Composes summary message with all key numbers
   │   Posts to Slack via webhook
   │
   ▼
Pipeline complete ✅ — typically finishes in under 3 minutes

Making the Load Idempotent

The pipeline deletes existing rows for the same date before inserting. This means re-running the DAG for the same day never creates duplicates. Re-running after a fix is always safe.

Run 1 (normal): inserts 320 rows for Jan 15
Run 2 (re-run after a bug fix):
  → deletes Jan 15 rows
  → inserts fresh 320 rows
  → result: still 320 rows, no duplicates

Testing Individual Tasks From the CLI

Run a single task in isolation without triggering the full DAG:

# Test the extract task for a specific date
airflow tasks test shopeasy_daily_pipeline extract_sales 2024-01-15

# Test the transform task
airflow tasks test shopeasy_daily_pipeline transform_data 2024-01-15

The test command runs the task immediately, prints its output, but does not save any state to the database. Use it during development to verify each task works before running the full pipeline.

Handling the Currency API Being Down

The fetch_fx_rates task has retries=3 inherited from DEFAULT_ARGS. If the currency API is temporarily unavailable:

Attempt 1: API timeout → task marked "up_for_retry"
   Wait 5 minutes
Attempt 2: API timeout → task marked "up_for_retry"
   Wait 10 minutes (exponential backoff)
Attempt 3: API responds → task succeeds
   Pipeline continues normally

The pipeline recovers automatically without any manual action from the team.

Monitoring This Pipeline in Production

Extending the Pipeline

Once this pipeline runs reliably, these are natural next steps:

• Add a data_quality_check task after load_to_db that queries the reporting table and raises AirflowFailException if row counts or totals look wrong
• Swap the static config file for Airflow Variables so the ops team can tune thresholds from the UI without touching code
• Add a @task.branch that sends a high-priority page to PagerDuty when outlier_count exceeds the threshold, instead of a routine Slack message
• Use Dynamic Task Mapping (task.expand) to process multiple regional databases in parallel by mapping extract_sales over a list of connection IDs
• Deploy the DAG using Git-sync on Kubernetes so any merged pull request automatically updates the running pipeline within 60 seconds