ETL/ELT Patterns and Best Practices: Real-World Strategies for Data Practitioners

ETL/ELT Patterns and Best Practices: Real-World Strategies for Data Practitioners

ETL (Extract, Transform, Load) and ELT (Extract, Load, Transform) represent fundamentally different philosophies for moving and processing data. As your data volumes grow and transformation complexity increases, choosing the right pattern—and implementing it correctly—becomes critical to system reliability and performance. This guide explores real patterns you'll encounter, common pitfalls that catch experienced practitioners, and practical techniques grounded in PostgreSQL and PySpark.

ETL vs. ELT: When and Why

The traditional ETL pattern transforms data before loading it into your target system. This made sense when storage was expensive and compute was centralized. You'd extract from a source, clean and validate in a staging area, then load only validated data.

ELT inverts this: load raw data first, transform in place. Modern cloud data warehouses (like PostgreSQL with appropriate indexing and partitioning) make this viable. The trade-off is clear:

ETL advantages: Smaller storage footprint, cleaner target system, easier to audit transformations before load, better for regulated environments.

ELT advantages: Faster initial load, flexibility to re-transform without re-extracting, leverage warehouse compute power, easier to recover from transformation errors.

In practice, most mature systems use hybrid patterns. You might ELT for high-volume fact data but ETL for slowly-changing dimensions. The pattern depends on your source reliability, transformation complexity, and recovery requirements.

Real Pattern 1: The Staging Layer Approach (ETL)

This is the workhorse pattern for regulated industries and systems requiring audit trails. Data flows: Source → Staging (raw) → Staging (cleaned) → Production.

The key insight: never transform directly into production. Staging layers give you rollback capability and allow validation before commit.

-- Pattern: Two-stage staging in PostgreSQL
-- Stage 1: Raw load (minimal validation)
CREATE TABLE stg_orders_raw (
    source_id TEXT,
    customer_id TEXT,
    amount TEXT,
    order_date TEXT,
    loaded_at TIMESTAMP DEFAULT NOW()
);

-- Stage 2: Cleaned (business logic applied)
CREATE TABLE stg_orders_clean (
    order_id BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    amount NUMERIC(12,2) NOT NULL,
    order_date DATE NOT NULL,
    validation_status VARCHAR(20),
    cleaned_at TIMESTAMP DEFAULT NOW()
);

-- Transformation logic
INSERT INTO stg_orders_clean (customer_id, amount, order_date, validation_status)
SELECT 
    CAST(customer_id AS BIGINT),
    CAST(amount AS NUMERIC),
    CAST(order_date AS DATE),
    CASE 
        WHEN CAST(amount AS NUMERIC) < 0 THEN 'INVALID_AMOUNT'
        WHEN CAST(customer_id AS BIGINT) IS NULL THEN 'MISSING_CUSTOMER'
        ELSE 'VALID'
    END
FROM stg_orders_raw
WHERE loaded_at > NOW() - INTERVAL '1 hour'
ON CONFLICT DO NOTHING;

This pattern's strength: you can inspect stg_orders_clean before running the final merge into production. If validation fails, you've caught it before corrupting your fact tables.

Common pitfall: Developers skip the validation stage to save time, then spend weeks debugging why customer IDs are NULL in production. The staging layer is your insurance policy.

Real Pattern 2: Incremental Load with Change Data Capture (CDC)

Full refreshes don't scale. Once your source table has millions of rows, reprocessing everything nightly becomes prohibitively expensive. CDC patterns load only what changed.

Three approaches exist:

Timestamp-based: Track updated_at columns. Simple but fails if updates are backdated or clocks drift.

Sequence-based: Use database sequences or log position. More reliable but requires source system cooperation.

Hash-based: Compare row hashes to detect changes. Expensive but catches all modifications.

Here's a practical timestamp-based approach:

-- Track last successful load
CREATE TABLE etl_checkpoint (
    table_name VARCHAR(100) PRIMARY KEY,
    last_loaded_at TIMESTAMP,
    load_status VARCHAR(20)
);

-- Incremental extraction
WITH last_checkpoint AS (
    SELECT COALESCE(last_loaded_at, '1900-01-01'::TIMESTAMP) as cutoff
    FROM etl_checkpoint
    WHERE table_name = 'orders'
)
SELECT 
    o.id,
    o.customer_id,
    o.amount,
    o.updated_at,
    'INSERT' as change_type
FROM source_orders o
CROSS JOIN last_checkpoint
WHERE o.updated_at > last_checkpoint.cutoff
AND o.updated_at <= NOW() - INTERVAL '5 minutes'  -- lag for late arrivals
ORDER BY o.updated_at;

Notice the 5-minute lag: sources often have clock skew or delayed writes. Waiting prevents missing records.

Common pitfall: Loading without the lag window. You'll mysteriously miss records that were written seconds before your extraction ran. Always add a safety margin.

Real Pattern 3: Streaming ELT with PySpark

For high-frequency data (events, logs, sensor data), batch ETL becomes impractical. Streaming patterns process data as it arrives, but introduce complexity around state management and exactly-once semantics.

Here's a practical streaming pattern using PySpark Structured Streaming:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, schema_of_json, window
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

spark = SparkSession.builder.appName("EventETL").getOrCreate()

# Read from Kafka (or any streaming source)
df_raw = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .option("startingOffsets", "latest") \
    .load()

# Parse JSON and apply transformations
schema = StructType([
    StructField("user_id", StringType()),
    StructField("event_type", StringType()),
    StructField("value", DoubleType())
])

df_parsed = df_raw.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

# Aggregate in 5-minute windows
df_aggregated = df_parsed.groupBy(
    window(col("timestamp"), "5 minutes"),
    col("user_id")
).agg({"value": "sum"})

# Write to PostgreSQL (checkpoint-aware)
query = df_aggregated.writeStream \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://localhost/warehouse") \
    .option("dbtable", "events_aggregated") \
    .option("checkpointLocation", "/tmp/checkpoint") \
    .outputMode("append") \
    .start()

query.awaitTermination()

The checkpointLocation is critical: it allows Spark to recover from failures without reprocessing or duplicating data.

Common pitfall: Omitting checkpoint locations or storing them on ephemeral storage. Your streaming job crashes, restarts, and reprocesses the last 6 hours of data. Checkpoints must be on durable storage (HDFS, S3, or local disk with replication).

Real Pattern 4: Slowly Changing Dimensions (SCD)

Dimensions (customers, products, locations) change slowly. How you handle these changes dramatically affects query correctness and storage.

SCD Type 1: Overwrite old values. Simplest, loses history.

SCD Type 2: Add new rows with validity dates. Preserves history, doubles storage.

SCD Type 3: Keep current and previous values in same row. Hybrid approach.

Type 2 is most common in analytics. Here's the pattern:

-- Dimension table with SCD Type 2
CREATE TABLE dim_customer (
    customer_key BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    name VARCHAR(255),
    email VARCHAR(255),
    valid_from DATE NOT NULL,
    valid_to DATE,
    is_current BOOLEAN DEFAULT TRUE,
    UNIQUE(customer_id, valid_from)
);

-- Merge logic (upsert with history)
WITH new_data AS (
    SELECT DISTINCT
        customer_id,
        name,
        email
    FROM stg_customers_clean
),
changed_rows AS (
    SELECT 
        d.customer_key,
        n.customer_id,
        n.name,
        n.email
    FROM dim_customer d
    FULL OUTER JOIN new_data n ON d.customer_id = n.customer_id
    WHERE d.is_current = TRUE
    AND (d.name != n.name OR d.email != n.email OR n.customer_id IS NULL)
)
UPDATE dim_customer d
SET valid_to = CURRENT_DATE - INTERVAL '1 day',
    is_current = FALSE
FROM changed_rows c
WHERE d.customer_key = c.customer_key;

INSERT INTO dim_customer (customer_id, name, email, valid_from, is_current)
SELECT 
    customer_id,
    name,
    email,
    CURRENT_DATE,
    TRUE
FROM new_data
WHERE customer_id NOT IN (SELECT customer_id FROM dim_customer WHERE is_current = TRUE)
ON CONFLICT DO NOTHING;

This pattern maintains a complete audit trail while keeping queries simple (filter on is_current = TRUE for current state).

Common pitfall: Not indexing on customer_id and is_current. Your dimension lookups will become table scans. Always index on the columns you join and filter on.

Real Pattern 5: Error Handling and Retry Logic

Production ETL fails. Networks timeout, sources go offline, data is malformed. Robust systems handle failures gracefully.

Key principles:

Idempotency: Running the same ETL twice produces the same result. Use ON CONFLICT clauses and deduplication logic.

Exponential backoff: Don't hammer a failing source. Wait 1 second, then 2, then 4, then 8.

Dead letter queues: Capture records that fail validation. Process them separately, not in the main pipeline.

import time
from datetime import datetime

def extract_with_retry(source_query, max_retries=3):
    """Extract with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return source_query()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt
            print(f"Attempt {attempt + 1} failed. Retrying in {wait_time}s")
            time.sleep(wait_time)

def validate_and_route(record):
    """Route valid records to main table, invalid to DLQ."""
    try:
        # Validation logic
        if not record.get('customer_id'):
            raise ValueError("Missing customer_id")
        if record.get('amount', 0) < 0:
            raise ValueError("Negative amount")
        return ('valid', record)
    except Exception as e:
        return ('invalid', {
            'original_record': record,
            'error': str(e),
            'timestamp': datetime.now().isoformat()
        })

# Usage
records = extract_with_retry(lambda: source_db.query("SELECT * FROM orders"))
valid_records = []
dlq_records = []

for record in records:
    status, processed = validate_and_route(record)
    if status == 'valid':
        valid_records.append(processed)
    else:
        dlq_records.append(processed)

# Load valid records
load_to_warehouse(valid_records)
# Log invalid records for investigation
log_to_dlq(dlq_records)

Common pitfall: Failing the entire ETL job when one record is malformed. Separate validation failures from system failures. A bad customer record shouldn't block loading 10,000 good orders.

Real Pattern 6: Monitoring and Observability

You can't fix what you can't see. Production ETL requires comprehensive monitoring.

Essential metrics:

• Record counts: Rows extracted, transformed, loaded. Compare to expected volumes.
• Latency: Time from source update to warehouse availability. SLAs depend on this.
• Error rates: Percentage of records failing validation. Sudden spikes indicate data quality issues.
• Checkpoint lag: For streaming, how far behind real-time are you?

-- Monitoring table
CREATE TABLE etl_metrics (
    run_id UUID PRIMARY KEY,
    pipeline_name VARCHAR(100),
    start_time TIMESTAMP,
    end_time TIMESTAMP,
    records_extracted BIGINT,
    records_loaded BIGINT,
    records_failed BIGINT,
    status VARCHAR(20),
    error_message TEXT
);

-- Query to detect anomalies
SELECT 
    pipeline_name,
    DATE(start_time) as run_date,
    AVG(records_loaded) as avg_records,
    MIN(records_loaded) as min_records,
    MAX(records_loaded) as max_records,
    STDDEV(records_loaded) as stddev_records
FROM etl_metrics
WHERE start_time > NOW() - INTERVAL '30 days'
GROUP BY pipeline_name, DATE(start_time)
HAVING MIN(records_loaded) < AVG(records_loaded) * 0.5
ORDER BY run_date DESC;

This query flags days where load volume dropped by 50% or more—a common sign of upstream issues.

Choosing Your Pattern: Decision Framework

No single pattern fits all scenarios. Use this framework:

Choose ETL (transform before load) if: You need strict data governance, have complex validation rules, or operate in regulated industries. Storage is cheap; data quality is expensive.

Choose ELT (load then transform) if: You need fast ingestion, have flexible transformation requirements, or want to re-transform without re-extracting. Your warehouse has sufficient compute.

Choose streaming if: Latency matters (sub-minute freshness required), data arrives continuously, or you need real-time aggregations.

Use CDC if: Your source is large (millions+ rows) and changes are incremental. Full refreshes become prohibitively expensive.

Use SCD Type 2 if: You need historical analysis ("what was the customer's address on this date?"). Accept the storage overhead.

Summary: Practical Takeaways

Mature ETL systems share common characteristics: they separate concerns (staging layers), handle failures gracefully (retry logic, DLQs), track state (checkpoints, CDC), and provide visibility (monitoring). The specific pattern—ETL vs. ELT, batch vs. streaming—matters less than applying these principles consistently.

Start simple. Use the staging layer pattern until you understand your data quality. Add CDC when full refreshes become slow. Introduce streaming only when latency requirements demand it. Each pattern adds complexity; add it only when the business case justifies it.