etl-pipeline

name	etl-pipeline
description	Design and implement production-ready ETL/ELT pipelines for data warehousing and analytics. Covers extraction strategies, transformation logic, incremental loading, data quality, orchestration, schema evolution, and monitoring.
argument-hint	["data sources","transformation requirements","destination system","schedule"]
allowed-tools	Read, Write, Bash

ETL Pipeline Design

Design production-ready ETL (Extract-Transform-Load) and ELT pipelines that reliably move data from heterogeneous sources to analytical destinations. Goes beyond basic SQL — covers orchestration, incremental loading, data quality gates, schema evolution, SCD handling, and operational monitoring.

Process

1. Inventory sources. Document each source: type (OLTP DB, API, file, stream), auth method, schema, volume, change frequency, and SLA constraints.
2. Choose ETL vs ELT. Transform before load (ETL) for sensitive data masking or when destination compute is expensive. Transform after load (ELT) when destination (Snowflake, BigQuery, Redshift) has cheap compute and raw storage is acceptable.
3. Design extraction strategy. Full refresh, incremental (watermark), CDC (change data capture), or log-based replication per source type.
4. Define transformation logic. Cleaning, deduplication, enrichment, type casting, business rule application, aggregation, join resolution.
5. Model destination schema. Fact/dimension star schema, SCD Type 1/2/3, surrogate keys, audit columns.
6. Implement data quality gates. Schema validation, null checks, range checks, referential integrity, row count reconciliation, freshness checks.
7. Set up orchestration. DAG definition, dependency management, retries, SLA alerts, backfill capability.
8. Design monitoring. Job duration, row counts in/out, error rates, data freshness, downstream impact alerts.
9. Handle schema evolution. Column additions, type changes, renames — forward/backward compatibility strategy.
10. Plan for failure modes. Partial load recovery, idempotent reruns, dead-letter queues for bad records.

Output Format

ETL Pipeline: [Pipeline Name]

Sources: [count and types] Destination: [data warehouse / lake] Schedule: [cron + timezone] Load Strategy: [Full Refresh | Incremental | CDC] Orchestrator: [Airflow / Prefect / dbt / Dagster] SLA: [max acceptable latency]

---

Airflow DAG — Incremental Load Pattern

# dags/etl_orders_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from datetime import datetime, timedelta
import pandas as pd
import logging

logger = logging.getLogger(__name__)

default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=60),
    "email_on_failure": True,
    "email": ["[e-mail]"],
    "sla": timedelta(hours=2),
}

with DAG(
    dag_id="etl_orders_incremental",
    default_args=default_args,
    schedule_interval="0 2 * * *",  # 2 AM UTC daily
    start_date=datetime(2024, 1, 1),
    catchup=True,
    max_active_runs=1,
    tags=["etl", "orders", "production"],
) as dag:

    def extract_orders():
        execution_date = context[]
        next_execution_date = context[]
        pg_hook = PostgresHook(postgres_conn_id=)
        query = 
        df = pg_hook.get_pandas_df(query, parameters={
            : execution_date.isoformat(),
            : next_execution_date.isoformat(),
        })
        logger.info()
         df.empty:
             
        output_path = 
        df.to_parquet(output_path, index=)
         {: (df), : output_path}

     ():
        ti = context[]
        result = ti.xcom_pull(task_ids=)
         result  :
            
        df = pd.read_parquet(result[])
        errors = []
         col  [, , ]:
             df[col].isna().() > :
                errors.append()
         df[].duplicated().():
            errors.append()
         (df[] < ).():
            errors.append()
        valid_statuses = {, , , , }
        invalid = (df[]..lower().unique()) - valid_statuses
         invalid:
            errors.append()
         errors:
             ValueError( + .join(errors))
        logger.info()

     ():
        ti = context[]
        result = ti.xcom_pull(task_ids=)
         result  :
            
        df = pd.read_parquet(result[])
        df[] = pd.to_datetime(df[])
        df[] = df[].astype()
        df[] = df[].dt.year
        df[] = df[].dt.month
        df[] = df[].dt.quarter
        df[] = df[] > 
        df[] = df[]..lower()..strip()
         datetime  timezone
        df[] = datetime.now(timezone.utc)
        df[] = context[].date()
        output_path = result[].replace(, )
        df.to_parquet(output_path, index=)
         {: (df), : output_path}

     ():
        ti = context[]
        result = ti.xcom_pull(task_ids=)
         result  :
            logger.info()
            
        df = pd.read_parquet(result[])
        sf_hook = SnowflakeHook(snowflake_conn_id=)
        temp_table = 
        sf_hook.run()
        sf_hook.insert_rows(
            table=temp_table,
            rows=df.values.tolist(),
            target_fields=df.columns.tolist(),
        )
        sf_hook.run()
        logger.info()

    t_extract  = PythonOperator(task_id=,   python_callable=extract_orders)
    t_validate = PythonOperator(task_id=, python_callable=validate_extract)
    t_transform= PythonOperator(task_id=, python_callable=transform_orders)
    t_load     = PythonOperator(task_id=,python_callable=load_to_snowflake)

    t_extract >> t_validate >> t_transform >> t_load

---

Destination Schema Design

-- Fact table with audit columns
CREATE TABLE analytics.fact_orders (
    order_sk        VARCHAR(32)    NOT NULL,   -- surrogate key (MD5 hash)
    order_id        BIGINT         NOT NULL,   -- natural key
    customer_id     BIGINT         NOT NULL,
    order_date      DATE           NOT NULL,
    status          VARCHAR(20)    NOT NULL,
    total_amount    DECIMAL(12,2)  NOT NULL,
    order_year      SMALLINT       NOT NULL,
    order_month     SMALLINT       NOT NULL,
    order_quarter   SMALLINT       NOT NULL,
    is_high_value   BOOLEAN        NOT NULL DEFAULT FALSE,
    etl_loaded_at   TIMESTAMP_NTZ  NOT NULL,
    etl_batch_date  DATE           NOT NULL,
    PRIMARY KEY (order_sk),
    UNIQUE (order_id)
)
CLUSTER BY (order_year, order_month);

-- Pipeline run metadata
CREATE TABLE meta.pipeline_runs (
    run_id          BIGINT AUTOINCREMENT PRIMARY KEY,
    pipeline_name   VARCHAR(100)   NOT NULL,
    batch_date      DATE           NOT NULL,
    rows_loaded     BIGINT           ,
    status          ()    ,
    run_at          TIMESTAMP_NTZ  ,
    error_message   TEXT
);

---

dbt ELT Pattern

-- models/staging/stg_orders.sql
{{ config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='append_new_columns',
    incremental_strategy='merge',
    cluster_by=['order_date']
) }}

WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
    {% if is_incremental() %}
        WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
    {% endif %}
),
cleaned AS (
    SELECT
        order_id::BIGINT                AS order_id,
        customer_id::BIGINT             AS customer_id,
        order_date::DATE                AS order_date,
        LOWER(TRIM(status))             AS status,
        total_amount::DECIMAL(12,2)     AS total_amount,
        updated_at::TIMESTAMP_NTZ       AS updated_at,
        CURRENT_TIMESTAMP()             AS dbt_loaded_at
    FROM source
    WHERE order_id IS NOT NULL
)
   cleaned

# models/staging/schema.yml
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: [pending, processing, shipped, delivered, cancelled]
      - name: total_amount
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1000000

---

SCD Type 2 — Slowly Changing Dimensions

-- Dimension table with full history
CREATE TABLE analytics.dim_customers (
    customer_sk     BIGINT AUTOINCREMENT PRIMARY KEY,
    customer_id     BIGINT         NOT NULL,
    email           VARCHAR(255)   NOT NULL,
    name            VARCHAR(200)   NOT NULL,
    tier            VARCHAR(20),
    region          VARCHAR(50),
    valid_from      DATE           NOT NULL,
    valid_to        DATE,           -- NULL = current record
    is_current      BOOLEAN        NOT NULL DEFAULT TRUE,
    created_at      TIMESTAMP_NTZ  NOT NULL DEFAULT CURRENT_TIMESTAMP()
);

-- SCD2 merge procedure
CREATE OR REPLACE PROCEDURE analytics.upsert_dim_customers(batch_date DATE)
RETURNS VARCHAR LANGUAGE SQL AS $$
BEGIN
    -- Expire changed records
    UPDATE analytics.dim_customers AS t
    SET valid_to = :batch_date - 1, is_current = FALSE
    FROM staging.customers_stage AS s
     t.customer_id  s.customer_id
       t.is_current  
       (t.email  s.email  t.name  s.name  t.tier  s.tier);

    
     analytics.dim_customers
        (customer_id, email, name, tier, region, valid_from, valid_to, is_current)
     s.customer_id, s.email, s.name, s.tier, s.region,
           :batch_date, , 
     staging.customers_stage s
      analytics.dim_customers t
         t.customer_id  s.customer_id  t.is_current  
     t.customer_id  
        t.email  s.email  t.name  s.name  t.tier  s.tier;

       :batch_date;
;
$$;

---

Data Quality Framework

# data_quality/checks.py
from dataclasses import dataclass
from typing import Callable
import pandas as pd
import logging

logger = logging.getLogger(__name__)

@dataclass
class QualityCheck:
    name: str
    severity: str  # "error" blocks pipeline, "warning" alerts only
    check_fn: Callable[[pd.DataFrame], bool]
    error_msg: str

def run_quality_checks(df: pd.DataFrame, checks: list) -> dict:
    results = {"passed": [], "warnings": [], "errors": []}
    for check in checks:
        try:
            passed = check.check_fn(df)
        except Exception as e:
            passed = False
        if passed:
            results["passed"].append(check.name)
        elif check.severity == "error":
            results["errors"].append(check.error_msg)
        else:
            results["warnings"].append(check.error_msg)
    if results["errors"]:
        raise ValueError("Quality checks failed:\n" + "\n".join(results["errors"]))
    for w  results[]:
        logger.warning()
     results


 ():
     QualityCheck(, ,
         df: df[col].notna().(), )

 ():
     QualityCheck(, ,
         df:  df[col].duplicated().(), )

 ():
     QualityCheck(, ,
         df: (df) >= n, )

 ():
     datetime  datetime, timezone, timedelta
     ():
        latest = pd.to_datetime(df[ts_col]).()
        cutoff = datetime.now(timezone.utc) - timedelta(hours=max_hours)
         latest.replace(tzinfo=timezone.utc) >= cutoff
     QualityCheck(, , check,
        )

---

Schema Evolution Handling

def detect_schema_changes(incoming_cols: list, current_cols: list) -> dict:
    incoming, current = set(incoming_cols), set(current_cols)
    return {
        "added":     list(incoming - current),   # safe — add columns
        "removed":   list(current - incoming),   # dangerous — requires review
        "unchanged": list(incoming & current),
    }

def apply_column_additions(sf_hook, table: str, added_cols: list, col_types: dict):
    """Safely add new columns without downtime."""
    for col in added_cols:
        col_type = col_types.get(col, "VARCHAR(255)")
        sf_hook.run(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {col} {col_type};")
        logger.info(f"Added column {col} ({col_type}) to {table}")

---

Extraction Strategies by Source Type

Source Type	Strategy	Tool	Notes
OLTP database (Postgres/MySQL)	Incremental watermark	SQLAlchemy + Airflow	Requires updated_at index
OLTP database (high volume)	CDC / log-based	Debezium + Kafka	Zero-latency, no source load
REST API	Cursor pagination	requests + retry	Rate limit aware
S3 / file drops	Partition scanning	boto3 + Spark	Track processed files
Event stream	Real-time consume	Kafka + Flink	Checkpoint offsets
SaaS platform	Fivetran / Airbyte	Managed connector	Fastest to implement

---

Rules

• Design for idempotency first. Rerunning a batch twice must yield the same result. Use MERGE/upsert — never blind INSERT.
• Incremental by default. Full refreshes don't scale past millions of rows. Use watermarks, sequence IDs, or CDC.
• Validate before loading. Quality gates catch upstream corruption before it contaminates analytics. Block on errors, warn on anomalies.
• Separate extract, transform, load. Each stage must be independently retryable and testable without re-running the others.
• Stage before merging. Write to a staging/temp table first, then MERGE into the target. A failed mid-load won't leave partial data.
• Track every run. Log pipeline name, batch window, row counts, status, and timestamps in a metadata table.
• Handle schema evolution explicitly. New columns are safe — automate them. Removed or renamed columns are breaking — gate with human review.
• Use SCD Type 2 for slowly changing dimensions. Preserve history for attributes that change (customer tier, address, pricing). Never silently overwrite.
• Set SLAs, not just schedules. Monitor both when the pipeline runs and when data is ready for consumers. Alert on both.
• Dead-letter bad records. Route malformed rows to a dead-letter table. Don't fail the whole pipeline over a few unparseable records.

Incremental Loading Pattern

def extract_incremental(table_name):
    # Get last successful watermark
    watermark = get_last_watermark(table_name)
    
    query = f"""
        SELECT * FROM {table_name}
        WHERE updated_at > %(watermark)s
        ORDER BY updated_at ASC
    """
    
    conn = psycopg2.connect(DATABASE_URL)
    df = pd.read_sql(query, conn, params={'watermark': watermark})
    
    # Save to staging
    df.to_parquet(f's3://staging/{table_name}/{date}.parquet')
    
    # Update watermark only after successful load
    if len(df) > 0:
        new_watermark = df['updated_at'].max()
        save_watermark(table_name, new_watermark)
    
    return len(df)

---

Pipeline Architecture

[PostgreSQL]  ─┐
               │
[S3 CSV]      ─┼─→ [Extract] → [Transform] → [Load] → [Snowflake]
               │      ↓            ↓            ↓
[REST API]    ─┘   [Staging]  [Quality]   [Final Tables]
                     Area       Checks

---