ETL vs ELT — Architecture Differences, Trade-offs & Migration Guide

# etl_pipeline.py
# Traditional ETL: transform before loading to warehouse
import json
from datetime import datetime

class ETLPipeline:
    def __init__(self, pipeline_name):
        self.name = pipeline_name

    def extract(self, source_type, connection_string):
        """Extract raw data from source."""
        print(f"[EXTRACT] Reading from {source_type}: {connection_string}")
        raw_data = [
            {"user_id": 1, "email": "alice"@example".com", "signup_date": "2026-01-15", "ssn": "123-45-6789"},
            {"user_id": 2, "email": "bob"@example".com", "signup_date": "2026-02-20", "ssn": "987-65-4321"},
            {"user_id": 3, "email": None, "signup_date": None, "ssn": "456-78-9012"},
        ]
        print(f"  Extracted {len(raw_data)} raw records")
        return raw_data

    def transform(self, data):
        """Transform: clean, validate, mask PII."""
        print(f"[TRANSFORM] Processing {len(data)} records")
        transformed = []
        for row in data:
            if not row.get("user_id"):
                continue
            if not row.get("email"):
                row["email"] = "unknown@unknown.com"
            transformed.append({
                "user_id": row["user_id"],
                "email": row["email"],
                "signup_date": row.get("signup_date") or datetime.now().strftime("%Y-%m-%d"),
                "has_pii": False,
                "ssn_masked": "***-**-****" if row.get("ssn") else None,
            })
        print(f"  Transformed: {len(transformed)} clean records")
        return transformed

    def load(self, data, target):
        """Load transformed data to warehouse."""
        print(f"[LOAD] Writing {len(data)} records to {target}")
        for row in data:
            print(f"  INSERT INTO {target} VALUES ({row['user_id']}, '{row['email']}', ...)")
        return {"status": "loaded", "records": len(data)}

    def run(self):
        print(f"\n=== ETL Pipeline: {self.name} ===\n")
        raw = self.extract("PostgreSQL", "postgresql://source:5432/ecommerce")
        clean = self.transform(raw)
        result = self.load(clean, "Snowflake.ANALYTICS.USERS")
        print(f"\nResult: {json.dumps(result, indent=2)}")

etl = ETLPipeline("User Data ETL")
etl.run()

# elt_pipeline.py
# ELT: load raw data, transform in warehouse
class ELTPipeline:
    def __init__(self, name):
        self.name = name

    def extract(self, source_type, connection_string):
        """Extract data from source."""
        print(f"[EXTRACT] Reading from {source_type}: {connection_string}")
        data = [
            {"user_id": 1, "email": "alice"@example".com", "signup_date": "2026-01-15", "ssn": "123-45-6789"},
            {"user_id": 2, "email": "bob"@example".com", "signup_date": "2026-02-20", "ssn": "987-65-4321"},
            {"user_id": 3, "email": None, "signup_date": None, "ssn": "456-78-9012"},
        ]
        print(f"  Extracted {len(data)} raw records")
        return data

    def load_raw(self, data, storage_location):
        """Load data as-is to data lake or raw schema."""
        print(f"[LOAD] Writing {len(data)} raw records to {storage_location}")
        for row in data:
            print(f"  INSERT INTO {storage_location} VALUES ({row})")
        return {"status": "loaded", "records": len(data)}

    def transform_in_warehouse(self, query):
        """Execute transformation SQL in the warehouse."""
        print(f"[TRANSFORM] Running SQL in warehouse:")
        print(f"  {query}")
        print(f"  Result: 3 rows affected (PII masked, nulls handled)")
        return {"status": "transformed", "sql": query}

    def run(self):
        print(f"\n=== ELT Pipeline: {self.name} ===\n")
        raw = self.extract("PostgreSQL", "postgresql://source:5432/ecommerce")
        load_result = self.load_raw(raw, "Snowflake.RAW.USER_DATA")
        sql = """
        CREATE OR REPLACE TABLE ANALYTICS.CLEAN_USERS AS
        SELECT
            user_id,
            COALESCE(email, 'unknown@unknown.com') AS email,
            COALESCE(signup_date, CURRENT_DATE) AS signup_date,
            '***-**-****' AS ssn_masked
        FROM RAW.USER_DATA;
        """
        transform_result = self.transform_in_warehouse(sql)
        print(f"\nResult: {json.dumps(transform_result, indent=2)}")

import json
elt = ELTPipeline("User Data ELT")
elt.run()

# etl_vs_elt_decision.py
# Score your use case for ETL vs ELT
class ArchitectureScorer:
    def __init__(self):
        self.etl_score = 0
        self.elt_score = 0

    def evaluate(self, criteria, value):
        """Score a criterion toward ETL or ELT."""
        scoring = {
            "data_volume": {"small": (1, 1), "medium": (1, 2), "large": (1, 3)},
            "transformation_complexity": {
                "simple_sql": (1, 3), "python_heavy": (3, 1), "ml_transforms": (2, 2)
            },
            "compliance_needs": {"pii_masking_before_load": (3, 0), "none": (1, 2)},
            "team_skills": {"sql_expert": (1, 3), "python_expert": (3, 1), "balanced": (2, 2)},
            "schema_volatility": {
                "stable": (2, 2), "changes_weekly": (3, 1), "changes_daily": (1, 3)
            },
            "raw_data_access": {"needed_direct": (1, 3), "only_clean": (3, 1)},
        }
        if criteria in scoring and value in scoring[criteria]:
            e, e2 = scoring[criteria][value]
            self.etl_score += e
            self.elt_score += e2
        return self

    def recommend(self):
        print(f"\n=== Architecture Recommendation ===\n")
        print(f"ETL Score: {self.etl_score}")
        print(f"ELT Score: {self.elt_score}")
        if self.elt_score > self.etl_score:
            print(f"Recommendation: ELT (score {self.elt_score} > {self.etl_score})")
        elif self.etl_score > self.elt_score:
            print(f"Recommendation: ETL (score {self.etl_score} > {self.elt_score})")
        else:
            print("Recommendation: Either — use ETL for sensitive data, ELT for analytics")

scorer = ArchitectureScorer()
scorer.evaluate("data_volume", "large")
scorer.evaluate("transformation_complexity", "simple_sql")
scorer.evaluate("compliance_needs", "none")
scorer.evaluate("team_skills", "sql_expert")
scorer.evaluate("schema_volatility", "changes_weekly")
scorer.evaluate("raw_data_access", "needed_direct")
scorer.recommend()

# pipeline_cost_model.py
# Compare total cost of ETL vs ELT for a given workload
class PipelineCostModel:
    def __init__(self, data_size_gb, daily_new_rows, team_size):
        self.data_size_gb = data_size_gb
        self.daily_new_rows = daily_new_rows
        self.team_size = team_size

    def etl_cost_monthly(self):
        etl_compute = self.data_size_gb * 0.05 * 30
        etl_storage_transient = 50
        etl_maintenance = self.team_size * 500
        return {
            "compute": round(etl_compute, 2),
            "storage": etl_storage_transient,
            "maintenance": etl_maintenance,
            "total": round(etl_compute + etl_storage_transient + etl_maintenance, 2),
        }

    def elt_cost_monthly(self):
        raw_storage = self.data_size_gb * 0.02 * 30
        transform_compute = self.data_size_gb * 0.02 * 30
        elt_maintenance = self.team_size * 200
        return {
            "compute": round(transform_compute, 2),
            "storage": round(raw_storage, 2),
            "maintenance": elt_maintenance,
            "total": round(transform_compute + raw_storage + elt_maintenance, 2),
        }

    def compare(self):
        etl = self.etl_cost_monthly()
        elt = self.elt_cost_monthly()
        print(f"\n=== Pipeline Cost Model ({self.data_size_gb}GB/day) ===\n")
        print(f"{'Cost Component':<20} {'ETL ($)':<15} {'ELT ($)':<15}")
        print("-" * 50)
        print(f"{'Compute':<20} {etl['compute']:<15} {elt['compute']:<15}")
        print(f"{'Storage':<20} {etl['storage']:<15} {elt['storage']:<15}")
        print(f"{'Maintenance':<20} {etl['maintenance']:<15} {elt['maintenance']:<15}")
        print("-" * 50)
        print(f"{'Total':<20} {etl['total']:<15} {elt['total']:<15}")
        savings = round(etl['total'] - elt['total'], 2)
        print(f"\nELT saves ${savings}/month ({round(savings/etl['total']*100)}%)" if savings > 0 else f"\nETL saves ${abs(savings)}/month")

model = PipelineCostModel(data_size_gb=500, daily_new_rows=50000000, team_size=3)
model.compare()