ETL vs ELT: Choosing the Right Approach

ETL vs ELT: Choosing the Right Data Transformation Paradigm

Data transformation is the most consequential architectural decision in any data pipeline. The choice between ETL (Extract-Transform-Load) and ELT (Extract-Load-Transform) determines your latency profile, cost structure, scalability ceiling, and operational complexity. This lesson provides a rigorous analysis of both paradigms with production-grade implementation strategies.

Why Transformation Strategy Matters

How ETL Works:

• Raw data is extracted from source systems
• Transformed in an intermediate processing engine (Spark, Informatica, DataStage)
• Loaded into a target data warehouse or data lake
• Transformation happens before persistence

How ELT Works:

• Raw data is extracted from source systems
• Loaded directly into a scalable storage layer (Snowflake, BigQuery, Redshift)
• Transformations applied after loading using SQL or dbt
• Raw data is preserved for replay and reprocessing

Why the Shift to ELT Happened:

1. Cloud storage costs dropped dramatically — storing raw data became cheap
2. MPP warehouses emerged — elastic compute for in-warehouse transformations
3. Raw data preservation — enables replay, debugging, and schema evolution

Key Insight: ETL remains the correct choice for streaming use cases, PII-heavy data that cannot be stored raw, and scenarios where downstream systems require pre-aggregated, denormalized feeds.

Architecture Diagram

ETL vs ELT Comparison

Production Implementation

ETL Pipeline with Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, udf, when
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
import logging

logger = logging.getLogger(__name__)

class ETLPipeline:
    """
    Production ETL pipeline with Spark.
    
    Parameters:
        spark (SparkSession): Active Spark session for distributed computation
        source_config (dict): Source system configuration including paths and connection strings
        target_config (dict): Target system configuration including paths and write modes
    """
    def __init__(self, spark: SparkSession, source_config: dict, target_config: dict):
        self.spark = spark
        self.source_config = source_config
        self.target_config = target_config

    def extract(self, source_path: str, schema: StructType = None):
        """
        Extract data from source with optional schema enforcement.
        
        Parameters:
            source_path (str): Path to source data (S3, HDFS, local)
            schema (StructType): Optional schema for type enforcement. If None, schema is inferred.
            
        Returns:
            DataFrame: Raw data with enforced schema
        """
        logger.info(f"Extracting from {source_path}")
        if schema:
            return self.spark.read.schema(schema).parquet(source_path)
        return self.spark.read.parquet(source_path)

    def transform(self, df):
        """
        Apply cleansing, enrichment, and aggregation transformations.
        
        Transformation pipeline:
        1. Deduplication on business keys
        2. Null and invalid record filtering
        3. Left join with dimension table for enrichment
        4. Risk tier classification via conditional logic
        5. Daily aggregation by customer
        """
        # Step 1: Remove duplicates based on business key
        df_deduped = df.dropDuplicates(["transaction_id", "event_date"])

        # Step 2: Filter invalid records (null amounts and negative values)
        df_filtered = df_deduped.filter(
            col("amount").isNotNull() & (col("amount") > 0)
        )

        # Step 3: Enrich with lookup data via broadcast join
        customer_df = self.spark.read.parquet(self.source_config["customer_path"])
        df_enriched = df_filtered.join(
            customer_df, on="customer_id", how="left"
        ).withColumn(
            "risk_tier",
            when(col("credit_score") >= 750, "low")
            .when(col("credit_score") >= 500, "medium")
            .otherwise("high")
        )

        # Step 4: Aggregate daily metrics per customer
        df_agg = df_enriched.groupBy("customer_id", "event_date").agg(
            F.sum("amount").alias("daily_total"),
            F.count("*").alias("transaction_count"),
            F.avg("amount").alias("avg_amount")
        )

        return df_agg

    def load(self, df, target_path: str, mode: str = "overwrite"):
        """
        Load transformed data to target with date partitioning.
        
        Parameters:
            df (DataFrame): Transformed data to load
            target_path (str): Output path for parquet files
            mode (str): Write mode - 'overwrite', 'append', 'ignore', 'error'
        """
        logger.info(f"Loading {df.count()} records to {target_path}")
        df.write.mode(mode).partitionBy("event_date").parquet(target_path)

    def run(self, source_path: str, target_path: str):
        """Execute the full ETL pipeline with error handling and logging."""
        try:
            raw_df = self.extract(source_path)
            transformed_df = self.transform(raw_df)
            self.load(transformed_df, target_path)
            logger.info("ETL pipeline completed successfully")
            return {"status": "success", "records": transformed_df.count()}
        except Exception as e:
            logger.error(f"ETL pipeline failed: {e}")
            raise

ELT Pipeline with dbt

# dbt_project.yml equivalent logic in Python for orchestration
from dataclasses import dataclass
from typing import List, Dict
import subprocess
import logging

logger = logging.getLogger(__name__)

@dataclass
class ELTPipeline:
    """
    Orchestrate an ELT pipeline using dbt transformations in a cloud warehouse.
    
    Parameters:
        warehouse_type (str): Target warehouse type - 'snowflake', 'bigquery', 'redshift'
        dbt_project_path (str): Path to dbt project root directory
        schemas (List[str]): List of schemas to process (e.g., ['raw', 'staging', 'marts'])
    """
    warehouse_type: str  # snowflake, bigquery, redshift
    dbt_project_path: str
    schemas: List[str]

    def extract_and_load(self, source_connections: Dict[str, str]) -> bool:
        """
        Phase 1: Extract from sources and load raw data into warehouse.
        
        Parameters:
            source_connections (Dict): Mapping of source names to connection strings
            
        Returns:
            bool: True if all sources loaded successfully, False otherwise
        """
        for source_name, connection_string in source_connections.items():
            logger.info(f"Loading raw data from {source_name}")
            # Use Fivetran, Airbyte, or custom loader for raw ingestion
            result = subprocess.run(
                ["python", "-m", " loaders.raw_loader",
                 "--source", source_name,
                 "--connection", connection_string,
                 "--target", f"raw.{source_name}"],
                capture_output=True, text=True
            )
            if result.returncode != 0:
                logger.error(f"Failed to load {source_name}: {result.stderr}")
                return False
        return True

    def transform(self, models: List[str] = None) -> bool:
        """
        Phase 2: Transform raw data using dbt models in the warehouse.
        
        Parameters:
            models (List[str]): Optional list of specific models to run. If None, runs all models.
            
        Returns:
            bool: True if all transformations succeeded, False otherwise
        """
        cmd = ["dbt", "run", "--profiles-dir", self.dbt_project_path]
        if models:
            cmd.extend(["--models", *models])

        logger.info(f"Running dbt transformations: {cmd}")
        result = subprocess.run(cmd, capture_output=True, text=True)

        if result.returncode != 0:
            logger.error(f"dbt transformation failed: {result.stderr}")
            return False

        logger.info("dbt transformations completed successfully")
        return True

    def test(self) -> bool:
        """
        Phase 3: Run dbt tests to validate data quality.
        
        Returns:
            bool: True if all tests passed, False otherwise
        """
        result = subprocess.run(
            ["dbt", "test", "--profiles-dir", self.dbt_project_path],
            capture_output=True, text=True
        )
        if result.returncode != 0:
            logger.error(f"dbt tests failed: {result.stderr}")
            return False
        return True

    def run(self, source_connections: Dict[str, str]) -> Dict:
        """
        Execute the full ELT pipeline: extract -> load -> transform -> test.
        
        Parameters:
            source_connections (Dict): Source system connection strings
            
        Returns:
            Dict: Pipeline execution result with status and phase information
        """
        results = {}

        # Extract and Load
        if not self.extract_and_load(source_connections):
            return {"status": "failed", "phase": "extract_load"}

        # Transform
        if not self.transform():
            return {"status": "failed", "phase": "transform"}

        # Test
        if not self.test():
            return {"status": "failed", "phase": "test"}

        return {"status": "success", "phases_completed": ["extract", "load", "transform", "test"]}

Best Practices

1. Always preserve raw data regardless of paradigm. Even in ETL, write raw extracts to a staging area before transformation for auditability and reprocessing.
2. Implement idempotent loads so that re-running a pipeline produces the same result. Use MERGE/UPSERT operations instead of INSERT.
3. Version your transformation logic using Git. Treat transformation code as first-class software, not SQL scripts.
4. Separate concerns: extraction, transformation, and loading should be independent, testable modules with clear interfaces.
5. Monitor pipeline SLAs with metrics for throughput, latency, error rate, and data freshness. Set alerts on anomalies.
6. Use schema evolution strategies that handle source schema changes without breaking downstream consumers.
7. Implement circuit breakers to stop pipeline execution when error rates exceed thresholds, preventing cascading failures.
8. Document data lineage from source to target. Use tools like OpenLineage or Apache Atlas for automated lineage tracking.
9. Test transformation logic with unit tests (per-function), integration tests (per-pipeline), and contract tests (per-output schema).
10. Design for failure: every pipeline stage should handle retries, dead letter queues, and partial completions gracefully.

Performance Benchmarks

See Also

• 017 - Apache Airflow: DAGs, Operators, and Scheduling - Orchestrate ETL/ELT pipelines with Airflow
• 021 - Apache Spark: RDDs, DataFrames, and the Catalyst Optimizer - ETL transformation engine fundamentals
• 024 - Data Ingestion Patterns: Batch, Streaming, CDC, and APIs - Extract phase patterns and strategies
• 025 - Data Quality: Validation Frameworks and Anomaly Detection - Quality checks for transformed data
• 023 - Batch vs Streaming: Lambda, Kappa, and Architecture Trade-offs - Processing paradigm selection framework
• 029 - Prefect: Modern Workflow Orchestration - Alternative orchestration for ETL/ELT