data-pipeline

name	data-pipeline
description	Data pipeline design and implementation guide covering ETL/ELT workflows, data quality validation, scheduling, and orchestration. Use when building data ingestion pipelines, creating ETL jobs, setting up data quality checks, designing stream processing flows, or planning data warehouse integration. Triggers on tasks involving data extraction, transformation, loading, data orchestration (Airflow, Dagster), CDC, or batch/stream processing architecture.
metadata	{"version":"1.0.0"}

Data Pipeline Best Practices

A comprehensive guide for designing, building, and maintaining reliable data pipelines — covering ETL/ELT patterns, data quality validation, orchestration, stream processing, and monitoring.

When to Apply

• Building a new data pipeline (batch or streaming)
• Designing ETL/ELT workflows for data warehouse integration
• Setting up data quality checks and validation
• Choosing between batch and stream processing
• Implementing Change Data Capture (CDC)
• Setting up pipeline orchestration and scheduling
• Troubleshooting failing or slow pipelines

---

1. Pipeline Architecture Patterns

1.1 ETL vs ELT

Aspect	ETL (Extract-Transform-Load)	ELT (Extract-Load-Transform)
Transform location	Staging server	Target data warehouse
Best for	Small-medium data, complex transforms	Large data, cloud warehouses
Tools	Informatica, SSIS, DataStage	dbt, Snowflake, BigQuery
Flexibility	Lower (transform before load)	Higher (transform in SQL)
Performance	Bottleneck on transform server	Leverages warehouse compute

Recommendation: Use ELT for modern cloud data warehouses. Use ETL when target has limited compute or data needs heavy pre-processing.

1.2 Batch vs Stream Processing

Aspect	Batch	Stream
Latency	Minutes to hours	Seconds to milliseconds
Throughput	Very high	Moderate
Complexity	Lower	Higher
Use case	Reports, analytics	Real-time dashboards, alerts
Tools	Airflow, Spark	Kafka, Flink, Spark Streaming

Decision rule: If data freshness requirements are > 15 minutes, use batch. If < 1 minute, use stream. Between 1-15 minutes, evaluate cost vs. urgency.

1.3 Lambda vs Kappa Architecture

Architecture	Description	When to Use
Lambda	Separate batch + stream layers	Need both real-time and historical processing
Kappa	Single stream processing layer	Simpler architecture, all data through stream

Recommendation: Start with Lambda if you have clear batch and real-time use cases. Prefer Kappa for new projects to reduce complexity.

---

2. Data Extraction

2.1 Extraction Patterns

Pattern	Description	Best For
Full load	Extract entire dataset each run	Small tables, initial load
Incremental (timestamp)	Extract rows modified since last run	Tables with updated_at column
Incremental (CDC)	Capture row-level changes	Large tables, frequent updates
Incremental (ID)	Extract rows with ID > last_max_id	Auto-increment tables

2.2 Change Data Capture (CDC)

Source DB → CDC Tool → Event Stream → Downstream Pipeline
                │
                ├── Debezium (PostgreSQL, MySQL)
                ├── AWS DMS
                ├── Fivetran
                └── Airbyte

CDC Best Practices:

• Always capture before and after images for updates
• Include operation type (INSERT, UPDATE, DELETE) in the event
• Use a dedicated replication user with minimal permissions
• Monitor CDC lag and set alerts for delays > 5 minutes
• Design for idempotency — downstream must handle duplicate events

2.3 API Extraction Guidelines

# Good: Paginated extraction with rate limiting and retry
def extract_from_api(base_url, params, max_retries=3):
    all_records = []
    page = 1
    while True:
        try:
            response = requests.get(
                f"{base_url}",
                params={**params, "page": page},
                timeout=30
            )
            response.raise_for_status()
            data = response.json()

            if not data.get("items"):
                break

            all_records.extend(data["items"])
            page += 1

            # Respect rate limits
            remaining = int(response.headers.get("X-RateLimit-Remaining", 1))
            if remaining <= 1:
                time.sleep(int(response.headers.get("Retry-After", 60)))

        except RequestException as e:
            if max_retries > 0:
                time.sleep(2 ** (max_retries - page))  # Exponential backoff
                max_retries -= 1
            else:
                raise

    return all_records

---

3. Data Transformation

3.1 Transformation Layer Principles

1. Idempotency — Running the same transform twice produces the same result
2. Deterministic — Same input always produces the same output
3. Testable — Each transform can be unit tested independently
4. Traceable — Data lineage from source to target is documented
5. Atomic — Transform succeeds completely or fails completely (no partial writes)

3.2 Transformation Patterns

Pattern	Description	Example
Filter	Remove rows that don't meet criteria	WHERE status = 'active'
Aggregate	Group and compute metrics	SUM(amount) GROUP BY date
Join	Combine data from multiple sources	Orders + Users + Products
Pivot	Rows to columns	Monthly metrics as columns
Lookup	Enrich with reference data	Map country code to name
Deduplicate	Remove duplicate records	ROW_NUMBER() OVER (PARTITION BY id)
Type cast	Convert data types	String date to TIMESTAMP

3.3 dbt-Style Transformation (Recommended)

models/
  staging/
    stg_users.sql          -- Raw source → cleaned staging
    stg_orders.sql
    stg_products.sql
  intermediate/
    int_order_items.sql    -- Join staging tables
    int_user_orders.sql
  marts/
    finance/
      fct_revenue.sql      -- Business-facing fact table
      dim_products.sql     -- Business-facing dimension table
    marketing/
      fct_campaign_attribution.sql

Key conventions:

• Staging (stg_) — 1:1 with source, minimal cleaning (rename, type cast)
• Intermediate (int_) — Business logic, joins, deduplication
• Marts (fct_/dim_) — Final tables for reporting and analytics
• Every model has tests (not_null, unique, accepted_values, relationships)

---

4. Data Loading

4.1 Load Strategies

Strategy	When to Use	Characteristics
Append	Event logs, transactions	Fast, no conflict resolution
Upsert (Merge)	Dimension tables, latest state	Handles inserts + updates
Replace	Full refresh snapshots	Simple but destructive
Incremental	Large fact tables	Only process new/changed data

4.2 Upsert Pattern (SQL)

-- PostgreSQL upsert example
INSERT INTO target_table (id, name, amount, updated_at)
SELECT id, name, amount, updated_at
FROM staging_table
ON CONFLICT (id)
DO UPDATE SET
    name = EXCLUDED.name,
    amount = EXCLUDED.amount,
    updated_at = EXCLUDED.updated_at
WHERE target_table.updated_at < EXCLUDED.updated_at;

4.3 Loading Best Practices

• Load to a staging table first, validate, then merge into target
• Use bulk loading for large datasets (COPY in PostgreSQL, BULK INSERT in SQL Server)
• Disable indexes during bulk load, rebuild after
• Set appropriate batch sizes (10k-100k rows per batch)
• Always verify row counts after loading

---

5. Data Quality Validation

5.1 Quality Check Framework

Implement checks at every pipeline stage:

Extract → [Schema Check] → [Freshness Check] → Transform → [Data Test] → Load → [Reconciliation]

5.2 Essential Data Quality Checks

Check Type	Description	Example
Not null	Critical fields have values	user_id IS NOT NULL
Uniqueness	No duplicate primary keys	COUNT(id) = COUNT(DISTINCT id)
Referential integrity	FK references exist	order.user_id IN (SELECT id FROM users)
Range check	Values within valid range	amount BETWEEN 0 AND 1000000
Format check	Values match expected format	email ~ '^[^@]+@[^@]+\.[^@]+$'
Freshness	Data is not stale	MAX(updated_at) > NOW() - INTERVAL '24 hours'
Volume check	Row count within expected range	COUNT(*) BETWEEN 10000 AND 50000
Distribution check	No unexpected skew	status values match historical distribution

5.3 Data Quality Implementation

class DataQualityCheck:
    def __init__(self, df, table_name):
        self.df = df
        self.table_name = table_name
        self.results = []

    def check_not_null(self, columns):
        for col in columns:
            null_count = self.df[col].isnull().sum()
            self.results.append({
                "table": self.table_name,
                "check": "not_null",
                "column": col,
                "passed": null_count == 0,
                "details": f"{null_count} null values"
            })

    def check_row_count(self, min_rows=1, max_rows=None):
        count = len(self.df)
        passed = count >= min_rows and (max_rows is None or count <= max_rows)
        self.results.append({
            "table": self.table_name,
            "check": "row_count",
            "passed": passed,
            "details": f"{count} rows (expected {min_rows}-{max_rows})"
        })

    def run_all(self):
        failed = [r for r in self.results if not r["passed"]]
        if failed:
            raise DataQualityError(f"{len(failed)} checks failed", failed)
        return self.results

---

6. Pipeline Orchestration

6.1 Orchestration Tools Comparison

Tool	Best For	Language	Cloud Native
Apache Airflow	General-purpose, large community	Python	Yes (MWAA, Cloud Composer)
Dagster	Data asset-centric, modern	Python	Yes (Dagster Cloud)
Prefect	Simple, developer-friendly	Python	Yes (Prefect Cloud)
Apache Spark	Big data processing	Scala/Python	Yes (EMR, Dataproc)
AWS Step Functions	AWS-native workflows	JSON	Yes

6.2 DAG Design Best Practices

Good DAG design:
┌──────────┐     ┌──────────┐     ┌──────────┐
│ Extract  │────▶│ Transform│────▶│  Load    │
│   API    │     │  & Test  │     │ Warehouse│
└──────────┘     └──────────┘     └──────────┘
                                         │
                                    ┌────▼────┐
                                    │ Validate│
                                    │ & Alert │
                                    └─────────┘

Bad DAG design (monolithic):
┌──────────────────────────────────────┐
│  Extract + Transform + Load + Test  │
└──────────────────────────────────────┘

Rules:

• Each task should do one thing and do it well
• Tasks should be idempotent — safe to re-run
• Set appropriate retries (usually 2-3 with exponential backoff)
• Define clear dependencies between tasks
• Add data sensors to wait for upstream data availability
• Set SLAs and alert on missed deadlines

6.3 Pipeline Scheduling

Schedule	Use Case
@daily / 0 5 * * *	Standard daily reports
@hourly	Near real-time dashboards
@weekly	Weekly summaries, aggregations
@monthly	Monthly financial reports
Event-triggered	On file arrival, API webhook

---

7. Error Handling & Monitoring

7.1 Error Handling Strategy

Level 1: Retry (transient errors)
  └── Network timeout, rate limit → Retry with backoff (3 attempts)

Level 2: Alert (recoverable errors)
  └── Schema change, missing data → Alert team, pause pipeline

Level 3: Dead letter queue (unrecoverable errors)
  └── Bad data format → Route to DLQ for manual review

7.2 Monitoring Checklist

• Pipeline run status (success/failure/running)
• Data freshness (time since last successful run)
• Row count comparison (source vs target)
• Pipeline duration trend (detect slow-downs)
• Error rate and error types
• Resource utilization (CPU, memory, disk)
• Data quality check pass/fail rates

7.3 Alerting Rules

Metric	Warning	Critical
Pipeline duration	> 2x average	> 4x average
Data freshness	> 2 hours late	> 6 hours late
Row count delta	> 10% deviation	> 30% deviation
Consecutive failures	2 in a row	5 in a row
Quality check failures	Any non-critical	Any critical check

---

8. Pipeline Testing Strategy

8.1 Test Types

Test Type	What to Test	When
Unit test	Individual transform functions	Every commit
Integration test	Source → staging → target	Before deploy
Data test	Data quality assertions	Every pipeline run
Snapshot test	Output doesn't change unexpectedly	Schema changes
Performance test	Pipeline completes within SLA	Scaling changes

8.2 Testing Best Practices

• Test with realistic sample data (anonymized production subset)
• Test edge cases: empty input, null values, duplicate records, special characters
• Test failure scenarios: source unavailable, bad schema, network timeout
• Maintain a test data fixture that evolves with the schema
• Run data tests as part of the pipeline (not just in CI)

---

9. Quick Reference: Pipeline Anti-Patterns

Anti-Pattern	Problem	Solution
No data quality checks	Silent data corruption	Add validation at every stage
Monolithic DAG	Hard to debug and retry	Split into small, focused tasks
No idempotency	Duplicate data on retry	Design idempotent operations
No monitoring	Failures go unnoticed	Set up alerts and dashboards
Hard-coded configs	Environment-specific bugs	Use configuration management
No data lineage	Can't trace data issues	Document source-to-target mapping
Ignore small failures	Cascading data quality issues	Fail fast, alert immediately
No version control	Can't rollback pipeline changes	Version all pipeline code