Data Pipelines

---
name: data-pipelines
description: Use when building or debugging data pipelines with Airflow or Prefect, writing dbt models or tests, designing incremental loads, implementing idempotent ETL/ELT jobs, validating data quality, or orchestrating multi-step data workflows.
---

# 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

```python
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()
```

### Operators & Sensors

```python
from airflow.operators.bash import BashOperator
from airflow.operators.python import BranchPythonOperator
from airflow.sensors.filesystem import FileSensor
from airflow.sensors.sql import SqlSensor
from airflow.providers.http.sensors.http import HttpSensor

# Wait for a file to appear (S3, GCS, local)
wait_for_export = FileSensor(
    task_id="wait_for_export",
    filepath="/data/exports/{{ ds }}/orders.csv",
    poke_interval=60,    # check every 60s
    timeout=3600,        # fail after 1 hour
    mode="reschedule",   # release worker slot while waiting
)

# Wait for upstream table to be populated
wait_for_source = SqlSensor(
    task_id="wait_for_orders",
    conn_id="source_db",
    sql="SELECT COUNT(*) FROM orders WHERE date = '{{ ds }}' HAVING COUNT(*) > 0",
    poke_interval=120,
    mode="reschedule",
)

# Branch: skip load on weekends
def should_load(**context):
    if context["execution_date"].weekday() >= 5:
        return "skip_load"
    return "load"

branch = BranchPythonOperator(task_id="check_day", python_callable=should_load)
```

### XComs — Task Communication

```python
# Push value
@task
def extract() -> dict:
    return {"row_count": 1042, "checksum": "abc123"}  # return value auto-pushes XCom

# Pull value
@task
def validate(stats: dict):   # passed as argument from task dependency
    assert stats["row_count"] > 0, "Empty extract"

# Manual XCom pull (classic operators)
def load(**context):
    stats = context["task_instance"].xcom_pull(task_ids="extract")
    print(stats["row_count"])
```

**XCom limits:** XComs are stored in the Airflow metadata DB — not suited for large data. Pass row counts, checksums, and file paths through XComs; never entire datasets.

### Dynamic Task Mapping

```python
@task
def get_regions() -> list[str]:
    return ["us-east", "eu-west", "ap-south"]

@task
def process_region(region: str):
    extract_and_load(region)

# Creates one task instance per region — parallelized automatically
process_region.expand(region=get_regions())
```

### Connections & Variables

```python
from airflow.hooks.base import BaseHook
from airflow.models import Variable

# Never hardcode credentials — use Connections
conn = BaseHook.get_connection("my_postgres")
dsn = f"postgresql://{conn.login}:{conn.password}@{conn.host}/{conn.schema}"

# Runtime config — use Variables (or better: Airflow Params)
batch_size = int(Variable.get("etl_batch_size", default_var=1000))
```

---

## dbt

### Project Structure

```
dbt_project/
├── models/
│   ├── staging/          # stg_* — raw → typed, renamed, deduplicated
│   │   └── stg_orders.sql
│   ├── intermediate/     # int_* — business logic joins
│   │   └── int_order_items.sql
│   └── marts/            # final — wide tables for BI/downstream
│       └── fct_revenue.sql
├── tests/                # custom SQL tests
├── macros/               # Jinja macros
├── seeds/                # static CSV reference data
└── dbt_project.yml
```

### Model Types & Materializations

```sql
-- staging/stg_orders.sql
-- Materialization: view (cheap, always fresh)
{{ config(materialized='view') }}

SELECT
    order_id::VARCHAR      AS order_id,
    user_id::VARCHAR       AS user_id,
    created_at::TIMESTAMP  AS created_at,
    amount_cents / 100.0   AS amount_usd,
    status
FROM {{ source('raw', 'orders') }}
WHERE status != 'test'
```

```sql
-- marts/fct_revenue.sql
-- Materialization: table (fast reads, rebuilt on each run)
{{ config(materialized='table') }}

SELECT
    DATE_TRUNC('day', o.created_at) AS date,
    p.name                          AS product_name,
    SUM(oi.quantity)                AS units_sold,
    SUM(oi.quantity * oi.unit_price) AS revenue_usd
FROM {{ ref('stg_orders') }}      o    -- ref() creates dependency
JOIN {{ ref('int_order_items') }} oi ON o.order_id = oi.order_id
JOIN {{ ref('stg_products') }}    p  ON oi.product_id = p.product_id
WHERE o.status = 'completed'
GROUP BY 1, 2
```

### Incremental Models

```sql
-- Only process new/updated rows — essential for large tables
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',    -- or 'delete+insert', 'insert_overwrite'
    on_schema_change='append_new_columns',
) }}

SELECT
    order_id,
    user_id,
    amount_usd,
    created_at,
    updated_at
FROM {{ source('raw', 'orders') }}

{% if is_incremental() %}
    -- Only load rows newer than the last run
    WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}
```

### Sources & Freshness

```yaml
# models/staging/sources.yml
version: 2

sources:
  - name: raw
    database: analytics
    schema: raw_data
    freshness:
      warn_after: {count: 6, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _loaded_at       # column that holds ingestion timestamp
    tables:
      - name: orders
        description: Raw orders from the transactional database
      - name: products
```

```bash
# Check source freshness in CI
dbt source freshness
```

### dbt Tests

```yaml
# models/staging/stg_orders.yml
version: 2

models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - not_null
          - unique
      - name: status
        tests:
          - accepted_values:
              values: ["pending", "completed", "cancelled", "refunded"]
      - name: user_id
        tests:
          - not_null
          - relationships:
              to: ref('stg_users')
              field: user_id
      - name: amount_usd
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 100000
```

```sql
-- tests/assert_revenue_non_negative.sql — custom SQL test (fails if rows returned)
SELECT date, revenue_usd
FROM {{ ref('fct_revenue') }}
WHERE revenue_usd < 0
```

### Macros

```sql
-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)::NUMERIC(10, 2)
{% endmacro %}

-- Usage in a model
SELECT {{ cents_to_dollars('amount_cents') }} AS amount_usd
```

```sql
-- macros/generate_surrogate_key.sql (or use dbt_utils)
{% macro surrogate_key(fields) %}
    MD5(CONCAT_WS('|', {% for f in fields %}COALESCE(CAST({{ f }} AS VARCHAR), ''){% if not loop.last %}, {% endif %}{% endfor %}))
{% endmacro %}
```

### dbt Commands

```bash
dbt run                              # run all models
dbt run --select staging             # run a directory
dbt run --select stg_orders+         # run model and all downstream
dbt run --select +fct_revenue        # run model and all upstream
dbt test                             # run all tests
dbt test --select stg_orders         # test one model
dbt build                            # run + test in dependency order
dbt source freshness                 # check source data freshness
dbt docs generate && dbt docs serve  # generate + serve lineage docs
dbt compile                          # render SQL without running
```

---

## Idempotency Patterns

A pipeline run is **idempotent** if running it twice produces the same result as running it once.

```python
# GOOD: delete-then-insert for a known partition
def load_partition(date: str, records: list[dict]):
    with engine.begin() as conn:
        conn.execute(
            text("DELETE FROM daily_stats WHERE date = :date"),
            {"date": date}
        )
        conn.execute(insert(DailyStats), records)

# GOOD: UPSERT (merge) on unique key
def upsert_orders(records: list[dict]):
    stmt = pg_insert(orders_table).values(records)
    stmt = stmt.on_conflict_do_update(
        index_elements=["order_id"],
        set_={"status": stmt.excluded.status, "updated_at": stmt.excluded.updated_at}
    )
    with engine.begin() as conn:
        conn.execute(stmt)

# BAD: append-only — reruns duplicate data
def load_orders(records):
    engine.execute(insert(orders_table).values(records))  # duplicates on rerun
```

**Airflow idempotency:** Use `{{ ds }}` (execution date, not run date) in all queries. Two runs for the same `ds` must produce the same output.

---

## Incremental Load Strategies

| Strategy | How | Use When |
|---|---|---|
| **Full refresh** | Truncate + reload entire table | Small tables (<1M rows), no CDC |
| **Incremental by timestamp** | `WHERE updated_at > last_run_max` | Source has reliable `updated_at` |
| **Incremental by partition** | Process one date partition per run | Append-only event data |
| **CDC (change data capture)** | Debezium → Kafka → warehouse | High-volume, low-latency, soft deletes |
| **Snapshot** | dbt snapshot (`strategy: timestamp`) | Track slowly-changing dimensions |

```python
# Watermark-based incremental (Python)
def get_watermark(conn, table: str) -> datetime:
    row = conn.execute(
        text("SELECT COALESCE(MAX(updated_at), '1970-01-01') FROM :table", bindparams=[bindparam("table")])
    ).fetchone()
    return row[0]

def extract_incremental(source_conn, watermark: datetime) -> list[dict]:
    return source_conn.execute(
        text("SELECT * FROM orders WHERE updated_at > :wm ORDER BY updated_at"),
        {"wm": watermark},
    ).fetchall()
```

---

## Data Validation

### dbt-native (preferred)

```yaml
# Generic tests: not_null, unique, accepted_values, relationships
# Package tests: dbt_utils, dbt_expectations (Great Expectations style)
- name: amount_usd
  tests:
    - dbt_expectations.expect_column_values_to_be_between:
        min_value: 0
        max_value: 50000
        row_condition: "status = 'completed'"
```

### Python validation (Great Expectations)

```python
import great_expectations as gx

context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")

validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="orders_suite",
)
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_be_unique("order_id")
validator.expect_column_values_to_be_between("amount_usd", min_value=0)
validator.expect_column_pair_values_A_to_be_greater_than_B(
    "completed_at", "created_at"
)

results = validator.validate()
if not results.success:
    raise ValueError(f"Data quality check failed: {results}")
```

### Row-count reconciliation

```python
@task
def reconcile(source_count: int, target_count: int, tolerance: float = 0.001):
    delta = abs(source_count - target_count) / max(source_count, 1)
    if delta > tolerance:
        raise ValueError(
            f"Row count mismatch: source={source_count}, target={target_count}, "
            f"delta={delta:.2%} > {tolerance:.2%} tolerance"
        )
```

---

## Monitoring & Alerting

```python
# Airflow: SLA miss callback
def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    send_slack_alert(f"SLA missed for DAG {dag.dag_id}: {task_list}")

@dag(sla_miss_callback=sla_miss_callback)
def my_dag():
    ...

# Airflow: task-level SLA (fail if task exceeds duration)
load = PythonOperator(
    task_id="load",
    python_callable=load_fn,
    sla=timedelta(minutes=30),   # alert if this task takes >30 min
)
```

```python
# Emit pipeline metrics to Prometheus/StatsD
from airflow.stats import Stats

Stats.incr("pipeline.rows_processed", count=row_count, tags={"dag": dag_id})
Stats.timing("pipeline.duration_ms", value=duration_ms, tags={"dag": dag_id})
```

---

## Red Flags

- **`catchup=True` on a new DAG** — Airflow will try to backfill all missed runs since `start_date`; set `catchup=False` on new DAGs and trigger backfills manually with `airflow dags backfill`
- **Passing datasets through XComs** — XComs are stored in the Airflow metadata DB (SQLite or Postgres); passing DataFrames or large lists corrupts the DB and kills performance; pass file paths, row counts, or checksums only
- **Non-idempotent pipeline** — if a run fails halfway and must be retried, appending duplicates corrupts the target; always upsert or delete-then-insert on a partition key
- **No `updated_at` index on source tables** — incremental loads do `WHERE updated_at > watermark`; without an index this is a full-table scan on every run; ensure the source has an index on the watermark column
- **Hard-coded credentials in DAG code** — DAGs are stored in version control and Airflow logs; always use Airflow Connections or environment variables, never string literals
- **`mode="poke"` on long-waiting sensors** — poke mode holds a worker slot while waiting; use `mode="reschedule"` so the slot is released between checks
- **Unbounded full-refresh on large tables** — a full refresh of a 500M-row table is slow and expensive; use incremental models with `unique_key` + merge strategy once the table exceeds 10M rows
- **No data quality tests before downstream loads** — failing silently and loading bad data is worse than failing loudly; add `dbt test` or row-count reconciliation as a gate before final loads

## Checklist

- [ ] DAG has `catchup=False` and `max_active_runs=1` unless backfill is intended
- [ ] All tasks are idempotent — reruns produce the same result
- [ ] Execution date (`{{ ds }}`) used in queries, not wall-clock time
- [ ] XComs carry only metadata (counts, paths, checksums) — not datasets
- [ ] Airflow Connections used for all credentials — no hardcoded secrets
- [ ] Sensors use `mode="reschedule"` not `mode="poke"`
- [ ] dbt staging models rename, cast, and deduplicate raw source data
- [ ] `ref()` used for all cross-model dependencies — never hardcoded table names
- [ ] Incremental models have `unique_key` and handle late-arriving data
- [ ] Source freshness checks configured and run in CI (`dbt source freshness`)
- [ ] dbt tests cover: `not_null`, `unique`, `accepted_values`, `relationships` on key columns
- [ ] Row-count reconciliation between source and target after each load
- [ ] SLA alerts configured for critical DAGs
- [ ] Backfill procedure documented and tested

> See also: `database-design` (index design, query optimization, migration patterns)
> See also: `observability` (structured logging, metrics, SLO alerting for pipeline health)