ETL/ELT Patterns¶

ETL (Extract-Transform-Load) and ELT (Extract-Load-Transform) are data integration patterns. This sheet covers extraction strategies, transformation patterns, loading strategies, incremental processing, and modern lakehouse approaches.

Table of Contents¶

Installation / Setup
Core Syntax
Common Operations
Practical Examples
Common Patterns
Best Practices
Common Mistakes
Troubleshooting
Related Cheat Sheets
Official Documentation

Installation / Setup¶

# Common libraries
import pandas as pd
import pyspark.sql.functions as F
from pyspark.sql import SparkSession

# Airflow for orchestration
from airflow import DAG
from airflow.operators.python import PythonOperator

Core Syntax¶

ETL vs ELT¶

Pattern	Description	When to Use
ETL	Extract → Transform → Load	Small data; complex transformations; data warehouse limits
ELT	Extract → Load → Transform	Large data; cloud data warehouses; flexible schemas
Streaming	Real-time extraction and processing	Low-latency requirements

ETL Flow:
Source → Extract → Stage → Transform → Load → Destination

ELT Flow:
Source → Extract → Load → Transform (in destination) → Destination

Extraction Strategies¶

# Full extraction
df = pd.read_csv('full_export.csv')
df = spark.read.parquet('s3://bucket/full_export/')

# Incremental extraction (by timestamp)
query = """
    SELECT * FROM orders
    WHERE updated_at > '{last_run}'
"""
df = pd.read_sql(query, conn, params={'last_run': last_run})

# Incremental extraction (by ID)
query = """
    SELECT * FROM orders
    WHERE id > {last_id}
"""

# Change Data Capture (CDC)
query = """
    SELECT * FROM orders
    WHERE operation IN ('INSERT', 'UPDATE', 'DELETE')
    AND timestamp > '{last_run}'
"""

# Full snapshot with merge
df_new = extract_current()
df_old = load_previous()
merged = merge_snapshots(df_new, df_old)

Transformation Patterns¶

# Cleaning
df = df.dropna(subset=['critical_column'])
df['email'] = df['email'].str.lower().str.strip()

# Normalization
df['name'] = df['name'].str.title()

# Denormalization
df = df.merge(dimensions, on='dimension_id')

# Aggregation
daily = df.groupby('date').agg({'amount': 'sum', 'quantity': 'sum'})

# Pivoting
pivot = df.pivot_table(index='date', columns='category', values='revenue')

# Unpivoting
melted = pd.melt(df, id_vars=['date'], var_name='category', value_name='revenue')

Loading Patterns¶

# Full load (truncate and insert)
df.to_sql('target_table', conn, if_exists='replace', index=False)

# Incremental append
df.to_sql('target_table', conn, if_exists='append', index=False)

# Upsert (merge)
# Using pandas
pd.merge(target, source, on='id', how='outer', suffixes=('', '_new'))

# Using PySpark Delta
delta_table.alias('target').merge(
    source.alias('source'),
    'target.id = source.id'
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

Common Operations¶

Incremental Processing¶

# Track last run
last_run_file = 'last_run.txt'

def get_last_run():
    try:
        with open(last_run_file) as f:
            return f.read().strip()
    except FileNotFoundError:
        return '1970-01-01'

def save_last_run(timestamp):
    with open(last_run_file, 'w') as f:
        f.write(timestamp)

# Incremental extraction
def extract_incremental():
    last_run = get_last_run()
    df = pd.read_sql(
        f"SELECT * FROM source WHERE updated_at > '{last_run}'",
        conn
    )
    return df

Data Quality Checks¶

def validate_data(df, rules):
    errors = []

    # Check for nulls in required columns
    for col in rules.get('required_columns', []):
        if df[col].isnull().any():
            errors.append(f"Nulls found in {col}")

    # Check value ranges
    for col, (min_val, max_val) in rules.get('ranges', {}).items():
        if df[col].min() < min_val or df[col].max() > max_val:
            errors.append(f"Values out of range in {col}")

    # Check uniqueness
    for col in rules.get('unique_columns', []):
        if df[col].duplicated().any():
            errors.append(f"Duplicates found in {col}")

    return errors

# Usage
rules = {
    'required_columns': ['id', 'name', 'email'],
    'ranges': {'age': (0, 150)},
    'unique_columns': ['id', 'email'],
}
errors = validate_data(df, rules)

Schema Evolution¶

# Track schema changes
schema_log = []

def log_schema(df, source):
    schema_log.append({
        'source': source,
        'timestamp': pd.Timestamp.now(),
        'columns': list(df.columns),
        'dtypes': df.dtypes.to_dict(),
    })

# Handle new columns
def merge_schemas(old_schema, new_schema):
    new_cols = set(new_schema) - set(old_schema)
    dropped_cols = set(old_schema) - set(new_schema)
    return new_cols, dropped_cols

Practical Examples¶

Example 1 — Daily ETL pipeline¶

import pandas as pd
from datetime import datetime, timedelta

def run_daily_etl():
    # Extract
    yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
    df = pd.read_sql(f"""
        SELECT * FROM orders
        WHERE order_date = '{yesterday}'
    """, conn)

    # Transform
    df['total'] = df['quantity'] * df['unit_price']
    df['tax'] = df['total'] * 0.1
    df['grand_total'] = df['total'] + df['tax']

    # Load
    df.to_sql('daily_orders', conn, if_exists='append', index=False)
    print(f"Loaded {len(df)} rows")

Example 2 — ELT with Delta Lake¶

# Extract: read raw data
raw_df = spark.read.csv('s3://raw/orders/', header=True, inferSchema=True)

# Load: write raw to Delta (bronze)
raw_df.write.format('delta').mode('append').saveAsTable('bronze_orders')

# Transform: clean and enrich (in SQL)
spark.sql("""
    CREATE OR REPLACE TABLE silver_orders AS
    SELECT
        id,
        LOWER(TRIM(customer_email)) as customer_email,
        CAST(order_date AS DATE) as order_date,
        quantity * unit_price as total,
        CURRENT_TIMESTAMP() as processed_at
    FROM bronze_orders
    WHERE quantity > 0
    AND customer_email IS NOT NULL
""")

# Aggregate (gold)
spark.sql("""
    CREATE OR REPLACE TABLE gold_daily_revenue AS
    SELECT
        order_date,
        SUM(total) as daily_revenue,
        COUNT(DISTINCT customer_email) as unique_customers
    FROM silver_orders
    GROUP BY order_date
""")

Example 3 — Streaming ELT¶

# Real-time extraction from Kafka
stream_df = spark.readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', 'localhost:9092') \
    .option('subscribe', 'orders') \
    .load()

# Parse and transform
parsed = stream_df \
    .select(F.from_json(F.col('value').cast('string'), schema).alias('data')) \
    .select('data.*')

# Load to Delta (append)
query = parsed.writeStream \
    .format('delta') \
    .outputMode('append') \
    .option('checkpointLocation', 's3://checkpoints/orders') \
    .toTable('silver_orders')

Common Patterns¶

Pattern: slowly changing dimensions (SCD)¶

# SCD Type 1: overwrite
new_df.to_sql('dimensions', conn, if_exists='replace')

# SCD Type 2: add new version
from datetime import datetime

new_df['valid_from'] = datetime.now()
new_df['valid_to'] = None
new_df['is_current'] = True

# Deactivate old records
old_df['valid_to'] = datetime.now()
old_df['is_current'] = False

Pattern: data vault¶

# Hub: business key
hubs = df[['id', 'business_key']].drop_duplicates()

# Links: relationships
links = df[['order_id', 'customer_id', 'product_id']].drop_duplicates()

# Satellites: descriptive attributes
satellites = df[['id', 'name', 'amount', 'timestamp']]

Pattern: audit trail¶

# Add audit columns
df['created_at'] = datetime.now()
df['updated_at'] = datetime.now()
df['created_by'] = 'etl_pipeline'
df['updated_by'] = 'etl_pipeline'
df['record_version'] = 1

Best Practices¶

Idempotent pipelines — same inputs = same outputs, safe to rerun.
Track lineage — know where data comes from and where it goes.
Incremental processing — avoid reprocessing everything.
Data quality checks — validate at each stage.
Audit logging — track who changed what and when.
Schema evolution — handle new columns gracefully.
Error handling — fail gracefully; log errors; alert.
Test with production-like data — catch issues early.

Common Mistakes¶

Mistake	Fix
Full reload every time	Use incremental processing
No data quality checks	Add validation at each stage
Hardcoding dates/times	Use dynamic date calculation
Not tracking lineage	Add metadata columns
Ignoring schema evolution	Handle new/removed columns
No error handling	Add try/except and alerting

Troubleshooting¶

Symptom	Cause / Fix
Pipeline fails silently	Add error handling and alerting
Duplicate data	Check idempotency; use upsert
Slow transformations	Optimize queries; add indexes
Data quality issues	Add validation; check source
Schema mismatch	Handle schema evolution; add columns

airflow.md — pipeline orchestration
data-pipelines.md — pipeline design patterns
apache-spark.md — distributed processing
delta-lake.md — Delta Lake operations
pandas — single-machine transformations

Official Documentation¶

ETL best practices — https://www.ibm.com/topics/etl
Delta Lake MERGE — https://docs.delta.io/latest/delta-utility.html

Last Updated¶

2026-09-06 — Initial version.

ETL/ELT Patterns¶

Table of Contents¶

Installation / Setup¶

Core Syntax¶

ETL vs ELT¶

Extraction Strategies¶

Transformation Patterns¶

Loading Patterns¶

Common Operations¶

Incremental Processing¶

Data Quality Checks¶

Schema Evolution¶

Practical Examples¶

Example 1 — Daily ETL pipeline¶

Example 2 — ELT with Delta Lake¶

Example 3 — Streaming ELT¶

Common Patterns¶

Pattern: slowly changing dimensions (SCD)¶

Pattern: data vault¶

Pattern: audit trail¶

Best Practices¶

Common Mistakes¶

Troubleshooting¶

Related Cheat Sheets¶

Official Documentation¶

Last Updated¶