ETL vs ELT: Modern Data Integration Patterns

# etl_pipeline.py
import pandas as pd
from sqlalchemy import create_engine

class ETLPipeline:
    def __init__(self, source_config, target_config):
        self.source = create_engine(source_config['connection'])
        self.target = create_engine(target_config['connection'])
    
    def extract(self, query: str) -> pd.DataFrame:
        """Extract data from source."""
        return pd.read_sql(query, self.source)
    
    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        """Transform data."""
        # Clean PII
        df = df.drop(columns=['ssn', 'credit_card'], errors='ignore')
        
        # Standardize formats
        df['created_at'] = pd.to_datetime(df['created_at'])
        
        # Remove duplicates
        df = df.drop_duplicates()
        
        # Add derived fields
        df['year'] = df['created_at'].dt.year
        df['month'] = df['created_at'].dt.month
        
        return df
    
    def load(self, df: pd.DataFrame, table_name: str):
        """Load to target."""
        df.to_sql(table_name, self.target, if_exists='replace', index=False)
    
    def run(self, query: str, table_name: str):
        """Execute full ETL pipeline."""
        print("Extracting data...")
        df = self.extract(query)
        
        print("Transforming data...")
        df = self.transform(df)
        
        print(f"Loading {len(df)} rows...")
        self.load(df, table_name)
        
        print("ETL pipeline complete!")

# dbt_project.yml
name: analytics_warehouse
version: '1.0.0'

models:
  analytics:
    +materialized: table
    
staging:
  +materialized: view

-- models/staging/stg_orders.sql
{{ config(materialized='view') }}

SELECT
    order_id,
    customer_id,
    CAST(total_amount AS DECIMAL(10,2)) AS total_amount,
    status,
    CAST(created_at AS TIMESTAMP) AS created_at,
    -- PII removed at transformation layer
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at) AS order_sequence
FROM {{ source('raw', 'orders_raw') }}
WHERE created_at >= '2024-01-01'

-- models/marts/fct_orders.sql
{{ config(materialized='table') }}

WITH customer_orders AS (
    SELECT
        customer_id,
        COUNT(*) AS total_orders,
        SUM(total_amount) AS lifetime_value,
        MIN(created_at) AS first_order_date,
        MAX(created_at) AS last_order_date
    FROM {{ ref('stg_orders') }}
    GROUP BY customer_id
)

SELECT
    o.*,
    c.lifetime_value,
    c.total_orders,
    DATEDIFF(day, c.first_order_date, c.last_order_date) AS customer_tenure_days,
    CASE 
        WHEN c.total_orders >= 10 THEN 'VIP'
        WHEN c.total_orders >= 5 THEN 'Regular'
        ELSE 'New'
    END AS customer_segment
FROM {{ ref('stg_orders') }} o
LEFT JOIN customer_orders c ON o.customer_id = c.customer_id

# airbyte_connection.yaml
source:
  name: postgres
  config:
    host: "source-db.example.com"
    port: 5432
    database: "sales"
    username: "replication_user"
    
destination:
  name: bigquery
  config:
    project_id: "analytics-project"
    dataset_id: "staging"

sync_mode: incremental
stream:
  - name: orders
    cursor_field: updated_at
    destination_sync_mode: append_dedup

# Fivetran uses declarative configuration
# connectors/fivetran_connector.json
{
  "schema": "salesforce",
  "tables": [
    {
      "name": "account",
      "sync_mode": "incremental",
      "destination": "postgres"
    },
    {
      "name": "opportunity",
      "sync_mode": "incremental"
    }
  ]
}