data-pipelines

Data Pipelines

Orchestration, transformation, and validation patterns for production data pipelines.

When to Activate

• Writing Airflow DAGs, operators, sensors, or XComs
• Building dbt models, sources, tests, or macros
• Designing incremental vs full-load strategies
• Implementing idempotent pipeline runs
• Validating data quality with dbt tests or Great Expectations
• Orchestrating multi-step ELT/ETL workflows
• Debugging failed runs, backfills, or data freshness issues

ETL vs ELT Decision

Approach	Transform where	Use when
ETL	Before loading (in pipeline code)	Target warehouse has limited compute; PII must be masked before storage
ELT	After loading (in warehouse SQL)	Modern warehouse (BigQuery, Snowflake, Redshift); raw data must be preserved
Streaming	Continuously (Kafka + Flink/Spark)	Sub-minute latency required; event sourcing

Default for modern stacks: ELT — land raw data, transform with dbt, version-control SQL.

Airflow

DAG Structure

from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook

@dag(
    schedule="0 6 * * *",          # 6 AM daily
    start_date=datetime(2026, 1, 1),
    catchup=False,                  # don't backfill missed runs on deploy
    max_active_runs=1,              # prevent overlapping runs
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "email_on_failure": True,
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task
    def extract_orders(execution_date=None) -> list[dict]:
        hook = PostgresHook(postgres_conn_id="source_db")
        # Use execution_date for idempotent extraction
        rows = hook.get_records(
            "SELECT * FROM orders WHERE date = %s",
            parameters=[execution_date.date()],
        )
        return [dict(r) for r in rows]

    @task
    def transform(orders: list[dict]) -> list[dict]:
        return [
            {**o, "revenue_usd": o["amount"] * o["fx_rate"]}
            for o in orders
            if o["status"] == "completed"
        ]

    @task
    def load(records: list[dict], execution_date=None):
        hook = PostgresHook(postgres_conn_id="warehouse")
        # Idempotent: delete-then-insert for the partition date
        hook.run("DELETE FROM daily_revenue WHERE date = %s", parameters=[execution_date.date()])
        hook.insert_rows("daily_revenue", [[r["date"], r["revenue_usd"]] for r in records])

    orders = extract_orders()
    transformed = transform(orders)
    load(transformed)

dag = daily_revenue_pipeline()