Data Etl Pipeline

Data ETL Pipeline

Purpose

Design reliable ETL/ELT pipelines with Airflow orchestration, dbt transformations, incremental strategies, error handling, and data validation.

Agent Protocol

Trigger

Exact user phrases: "ETL", "ELT", "data pipeline", "Airflow", "dbt", "data transformation", "data ingestion", "batch processing", "pipeline orchestration", "incremental load", "data pipeline design", "DAG", "data workflow", "extract load transform".

Input Context

Before activating, verify:

• Source systems (databases, APIs, files, streams)
• Target warehouse (Snowflake, BigQuery, Redshift, DuckDB)
• Volume and frequency (daily/hourly batch, CDC, real-time)
• Orchestration preference (Airflow, Dagster, Prefect)
• Transformation tool (dbt, custom SQL, Spark)
• Data volume and growth rate
• SLAs for data freshness and availability
• Existing monitoring and alerting infrastructure

Output Artifact

ETL pipeline design with DAG structure, transformation config, error handling as YAML and SQL.

Response Format

# Airflow DAG skeleton
# Task definitions

# dbt model config
# Incremental strategy

# Transformation query template

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.

Completion Criteria

• Pipeline architecture diagram defined (sources → staging → warehouse)
• Airflow DAG structure with task dependencies and retries
• Incremental loading strategy selected and configured
• Error handling with retry, dead-letter, and notification
• Data validation checks on each stage
• Monitoring and alerting configured
• Data lineage tracking set up

Max Response Length

300 lines of code and configuration.

ETL vs ELT

ETL (Extract, Transform, Load)

Transform happens before loading. Best for: on-premises databases, structured data, complex transformations requiring significant compute, regulatory environments requiring data masking before storage. ETL requires a transformation engine (Spark, Python) between extraction and loading. Transformation reduces data volume before warehouse storage, saving on warehouse costs.

ELT (Extract, Load, Transform)

Transform happens in the warehouse. Best for: cloud warehouses (Snowflake, BigQuery, Redshift), raw data preservation, agile schema evolution, when the warehouse provides sufficient compute for transformations. ELT loads raw data into staging tables first, then transforms using SQL. Recommended for most cloud data warehouse pipelines.

Decision Guide

Airflow DAG Design

DAG Structure

One DAG per data domain. Structure: start → extract → validate_extract → load_staging → validate_staging → transform → validate_transform → load_mart → complete. Task dependencies use the bitshift operator: extract >> validate_extract >> load_staging.

DAG Template

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['pager@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(hours=1),
    'execution_timeout': timedelta(hours=2),
    'sla': timedelta(minutes=30),
}

with DAG(
    'etl_orders_daily',
    default_args=default_args,
    description='Daily orders ETL from PostgreSQL to warehouse',
    schedule='0 3 * * *',  # Daily 3 AM
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['etl', 'orders', 'daily'],
) as dag:
    start = DummyOperator(task_id='start')
    extract = PythonOperator(task_id='extract_orders', ...)
    complete = DummyOperator(task_id='complete')

    start >> extract >> complete

Task Parameters

Scheduling Patterns

Daily: 0 3 * * * (off-peak hours). Hourly: 0 * * * *. Weekly: 0 4 * * 1 (Mondays 4am). Event-driven: use TriggerDagRunOperator from another DAG. Data-aware scheduling: use Airflow 2.4+ datasets as dependencies between DAGs. Sensor pattern: ExternalTaskSensor or SqlSensor to wait for upstream completion.

dbt Transformation Patterns

Model Organization

models/
  staging/
    stg_orders.sql        # Source-close, minimal transforms
    stg_customers.sql
    stg_payments.sql
  intermediate/
    int_order_details.sql # Business logic, joins
    int_customer_metrics.sql
  marts/
    fct_orders.sql        # Aggregated, consumption-ready
    dim_customers.sql
    dim_products.sql

Staging Model Pattern

-- models/staging/stg_orders.sql
WITH source AS (
    SELECT * FROM {{ source('postgres', 'orders') }}
),
renamed AS (
    SELECT
        id AS order_id,
        customer_id,
        order_date,
        status,
        total_amount,
        created_at,
        updated_at
    FROM source
    WHERE _deleted = false  -- Exclude soft-deleted records
)
SELECT * FROM renamed

Intermediate Model Pattern

-- models/intermediate/int_order_details.sql
WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),
payments AS (
    SELECT * FROM {{ ref('stg_payments') }}
)
SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.status,
    o.total_amount,
    COALESCE(p.total_paid, 0) AS total_paid,
    o.total_amount - COALESCE(p.total_paid, 0) AS balance_due,
    CASE
        WHEN o.status = 'delivered' AND p.total_paid >= o.total_amount THEN 'paid'
        WHEN o.status = 'delivered' THEN 'payment_pending'
        WHEN o.status = 'cancelled' THEN 'cancelled'
        ELSE 'active'
    END AS order_health
FROM orders o
LEFT JOIN payments p ON o.order_id = p.order_id

Testing

Generic Tests

# schema.yml
version: 2
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id
      - name: status
        tests:
          - accepted_values:
              values: ['draft', 'submitted', 'confirmed', 'shipped', 'delivered', 'cancelled']
      - name: total_amount
        tests:
          - dbt_utils.expression_is_true:
              expression: '>= 0'

Test Maturity Model

Documentation

# schema.yml — model and column descriptions
version: 2
models:
  - name: fct_orders
    description: >
      Order fact table containing one row per completed order.
      Used for revenue reporting, sales analysis, and forecasting.
    columns:
      - name: order_id
        description: "Unique identifier for each order from source system"
        tests: [unique, not_null]
      - name: total_amount
        description: "Total order amount including tax and shipping, in USD"

Snapshots

SCD Type 2 snapshots track historical changes to dimension tables. Strategy: timestamp (based on updated_at column) or check (based on column value changes). Query current records with WHERE dbt_valid_to IS NULL.

-- snapshots/customers.sql
{% snapshot customers_snapshot %}
{{
    config(
        target_schema='snapshots',
        unique_key='customer_id',
        strategy='timestamp',
        updated_at='updated_at',
        invalidate_hard_deletes=True
    )
}}
SELECT * FROM {{ source('postgres', 'customers') }}
{% endsnapshot %}

Batch vs Incremental Processing

Batch Processing

Process all data from the source in each run. Simple to implement and debug. Best for: small datasets, reference data, initial loads, nightly reporting. Full refresh is the default materialization for small dimension tables.

Incremental Processing

Process only new or changed data since the last run. Complex to implement but efficient for large datasets. Best for: large fact tables, event data, append-only logs, daily/hourly loading.

{{ config(materialized='incremental', unique_key='order_id') }}

SELECT * FROM {{ source('postgres', 'orders') }}
{% if is_incremental() %}
    WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

Incremental Strategies

S3 Staging

Staging Area Design

Raw data lands in S3 (or equivalent cloud storage) partitioned by source, date, and load timestamp. Structure: s3://data-lake/landing/<source>/<date>/<load_id>/. Data is in columnar format (Parquet) for efficient querying. Glue Crawler or equivalent registers partitions in the metastore. Staging tables in the warehouse point to the S3 location.

staging_area:
  raw: s3://data-lake/raw/salesforce/
  structure: "{source}/{object}/{year}/{month}/{day}/{load_id}/"
  format: parquet
  compression: zstd
  retention: 30 days raw, 7 days staging
  validation:
    - row_count_match
    - schema_compatibility
    - checksum_verification

Data Validation

Stage Validations

Transform Validations

Referential integrity: FK columns match PK values in referenced tables. Aggregate comparison: totals match between source and target (SUM, COUNT). Unique key enforcement: no duplicates in PK columns. Distribution drift: value distributions compared to baseline (Kolmogorov-Smirnov test for numerical fields).

# Custom validation hook
def validate_row_count(df, expected_min=1000, expected_max=None):
    count = df.count()
    if count < expected_min:
        raise ValueError(f"Row count {count} below minimum {expected_min}")
    if expected_max and count > expected_max:
        raise ValueError(f"Row count {count} exceeds maximum {expected_max}")
    return True

Error Handling

Retry Strategy

Dead Letter Queue

-- DLQ table structure
CREATE TABLE pipeline_dlq (
    id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    dag_id VARCHAR(100) NOT NULL,
    task_id VARCHAR(100) NOT NULL,
    execution_date TIMESTAMPTZ NOT NULL,
    source_table VARCHAR(200),
    record_id VARCHAR(200),
    payload JSONB,
    error_message TEXT,
    error_type VARCHAR(100),
    retry_count INT DEFAULT 0,
    created_at TIMESTAMPTZ DEFAULT now(),
    resolved_at TIMESTAMPTZ,
    resolution VARCHAR(500)
);

-- Backfill query
INSERT INTO staging_orders (
    SELECT * FROM pipeline_dlq
    WHERE dag_id = 'etl_orders_daily'
    AND resolved_at IS NULL
    AND error_type = 'invalid_data_format'
);

Alerting

Pipeline Architecture Comparison

Common Airflow DAG Patterns

Sequential Pattern

start → extract → validate → load → transform → load_mart → complete

Best for: simple pipelines, single-source ingestion, weekly batch jobs.

Fan-Out Pattern

start → extract_all
  ├── validate_orders → load_orders → transform_orders → load_order_mart
  ├── validate_customers → load_customers → transform_customers → load_customer_mart
  └── validate_inventory → load_inventory → transform_inventory → load_inventory_mart
complete

Best for: multi-source pipelines, independent domain processing.

Conditional Branch Pattern

start → check_data_availability
  ├── [data available] → extract → validate → load → transform → complete
  └── [no data] → skip_run → complete (notify: no data today)

Best for: source systems with unreliable data delivery schedules.

Join Pattern

start → extract_orders ────┐
start → extract_payments ──┤→ join_orders_payments → validate_join → load_mart → complete
                            └── wait_for_both (sensor)

Best for: pipelines requiring data from multiple sources before downstream processing.

Error Handling Topology

Source → Extract Task
  ├── Success → Validate Task
  │   ├── Row count OK → Load Staging
  │   │   ├── Success → Transform
  │   │   └── Failure → Alert, halt pipeline
  │   └── Row count FAIL → Halt pipeline (schema/volume issue)
  └── Failure (retryable) → Retry (3x, exponential backoff)
      └── All retries exhausted → DLQ + Alert + Halt

Pipeline SLA Dashboard Metrics

Orchestration Comparison

Additional ETL Tools

Apache NiFi

NiFi provides a visual, no-code approach to data routing and transformation. Drag-and-drop processor chaining, data provenance tracking, backpressure, and priority queuing. Ideal for ingestion from heterogeneous sources and protocol translation. Deploy as a standalone cluster with ZooKeeper.

Mage.ai

Mage.ai is a modern open-source ETL tool with Python-native pipeline definition. Pipelines are blocks connected in a DAG with @transformer and @loader decorators. Auto-generated UI, real-time monitoring, and built-in dbt/Spark/BigQuery integration.

Kestra

Kestra uses declarative YAML for pipeline definitions with a powerful orchestration engine. Supports batch and event-driven workflows with built-in error handling, retries, and SLA monitoring. Plugin ecosystem covers ETL, dbt, Python, and cloud services.

Cloud ETL Services

AWS Glue: serverless Spark-based ETL with schema crawler and auto-generated catalog. Azure Data Factory: 90+ built-in connectors with mapping data flows and trigger-based orchestration. GCP Dataflow: fully-managed Apache Beam for batch and streaming with auto-scaling and exactly-once semantics.

Pipeline CI/CD

Testing Pipeline

# .github/workflows/dbt-ci.yml
jobs:
  dbt-ci:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: dbt-labs/dbt-action@v1
        with:
          dbt_command: dbt build --select state:modified+ --target ci
      - uses: dbt-labs/dbt-action@v1
        with:
          dbt_command: dbt docs generate

Deployment Pipeline

# Deploy steps:
# 1. PR merged to main → trigger staging deployment
# 2. Build and test in staging environment
# 3. Run dbt run --full-refresh on staging (weekly)
# 4. Smoke tests pass (row counts, freshness)
# 5. Promote to production: dbt run --target prod
# 6. Run dbt test --target prod (post-deployment validation)
# 7. Rollback: dbt run --select fct_orders --vars '{"prev_version": true}'

Rules

• ELT over ETL for cloud warehouses
• One DAG per domain, max 20 tasks per DAG
• Every task has retry, timeout, and SLA
• Incremental by default, full refresh exception
• dbt tests on every model, run in CI and pipeline
• Failed records land in DLQ, never silently dropped
• Pipeline halted if source data fails validation
• All transformations idempotent
• Monitor data freshness, not just pipeline success
• Track row count trends for anomaly detection
• dbt artifacts archived for lineage tracking
• Backfills are run as separate DAG runs with catchup
• Pipeline code reviewed and tested before production deploy
• Every pipeline has a documented owner and on-call rotation

References

• references/cloud-etl-services.md — Cloud ETL Services
• references/data-pipeline-cicd.md — Data Pipeline CI/CD
• references/etl-elt-patterns.md — ETL/ELT Patterns
• references/etl-pipeline-design.md — ETL Pipeline Design
• references/nifi-mage-patterns.md — Apache NiFi and Mage.ai ETL Patterns
• references/pipeline-monitoring.md — Pipeline Monitoring

Handoff

data-data-quality for validation rules and data contract enforcement data-data-warehouse for target schema design and optimization