ETL vs ELT: Data Integration Patterns

import pandas as pd
from sqlalchemy import create_engine

# EXTRACT: Pull data from source database
source_engine = create_engine("postgresql://source_db:5432/app")
raw_orders = pd.read_sql("""
    SELECT order_id, customer_id, product_id, quantity,
           unit_price, order_date, status
    FROM orders
    WHERE order_date >= '2025-01-01'
""", source_engine)

raw_customers = pd.read_sql("""
    SELECT customer_id, name, email, country, created_at
    FROM customers
""", source_engine)

# TRANSFORM: Clean, validate, and join data outside the warehouse
# Remove duplicates
raw_orders = raw_orders.drop_duplicates(subset=["order_id"])

# Calculate derived fields
raw_orders["total_amount"] = raw_orders["quantity"] * raw_orders["unit_price"]
raw_orders["order_date"] = pd.to_datetime(raw_orders["order_date"])
raw_orders["order_month"] = raw_orders["order_date"].dt.to_period("M")

# Join orders with customers
enriched_orders = raw_orders.merge(
    raw_customers[["customer_id", "country"]],
    on="customer_id",
    how="left"
)

# Filter out cancelled orders
enriched_orders = enriched_orders[enriched_orders["status"] != "cancelled"]

# Aggregate to monthly summary
monthly_summary = enriched_orders.groupby(["order_month", "country"]).agg(
    total_orders=("order_id", "count"),
    total_revenue=("total_amount", "sum"),
    unique_customers=("customer_id", "nunique"),
).reset_index()

# LOAD: Write transformed data to the warehouse
warehouse_engine = create_engine("snowflake://warehouse/analytics")
monthly_summary.to_sql(
    "monthly_revenue_by_country",
    warehouse_engine,
    schema="analytics",
    if_exists="replace",
    index=False
)
print(f"Loaded {len(monthly_summary)} rows into analytics.monthly_revenue_by_country")

-- ELT: Raw data is already loaded into the warehouse by an ingestion tool
-- (Fivetran, Airbyte, etc.) into a raw schema

-- Step 1: Staging model — clean and rename raw data
CREATE OR REPLACE VIEW staging.stg_orders AS
SELECT
    id AS order_id,
    customer_id,
    product_id,
    quantity,
    unit_price,
    quantity * unit_price AS total_amount,
    CAST(created_at AS DATE) AS order_date,
    DATE_TRUNC('month', created_at) AS order_month,
    status,
    _fivetran_synced AS loaded_at
FROM raw.app_public.orders
WHERE status != 'cancelled'
  AND id IS NOT NULL;

-- Step 2: Create enriched model by joining in the warehouse
CREATE OR REPLACE TABLE analytics.orders_enriched AS
SELECT
    o.order_id,
    o.customer_id,
    c.name AS customer_name,
    c.country,
    o.product_id,
    p.product_name,
    p.category,
    o.quantity,
    o.unit_price,
    o.total_amount,
    o.order_date,
    o.order_month
FROM staging.stg_orders o
LEFT JOIN staging.stg_customers c ON o.customer_id = c.customer_id
LEFT JOIN staging.stg_products p ON o.product_id = p.product_id;

-- Step 3: Aggregate for reporting
CREATE OR REPLACE TABLE analytics.monthly_revenue AS
SELECT
    order_month,
    country,
    COUNT(DISTINCT order_id) AS total_orders,
    SUM(total_amount) AS total_revenue,
    COUNT(DISTINCT customer_id) AS unique_customers,
    SUM(total_amount) / COUNT(DISTINCT order_id) AS avg_order_value
FROM analytics.orders_enriched
GROUP BY order_month, country
ORDER BY order_month DESC, total_revenue DESC;

# Hybrid approach: ETL for sensitive data, ELT for the rest

# ETL: Mask PII before loading into the warehouse
def mask_pii_and_load(records: list[dict]):
    """ETL pipeline that masks PII before warehouse load."""
    import hashlib

    masked = []
    for record in records:
        masked.append({
            "user_id": record["user_id"],
            "email_hash": hashlib.sha256(
                record["email"].encode()
            ).hexdigest(),
            "country": record["country"],
            "signup_date": record["signup_date"],
            # Original email is never stored in the warehouse
        })

    load_to_warehouse(masked, "raw.users_masked")

# ELT: Everything else loads raw, transforms in SQL via dbt
# Fivetran syncs orders, products, events into raw schema
# dbt transforms raw -> staging -> analytics

# dbt_project.yml — The transformation layer in an ELT architecture
name: 'analytics'
version: '1.0.0'
profile: 'snowflake_prod'

models:
  analytics:
    staging:        # Clean and rename raw data
      +materialized: view
      +schema: staging
    intermediate:   # Business logic and joins
      +materialized: ephemeral
    marts:          # Final tables for analytics
      +materialized: table
      +schema: analytics

# The raw schema is populated by Fivetran/Airbyte (Extract + Load)
# dbt handles all transformations (Transform)
# This separation is the essence of ELT