04 · ETL & ELT: Pipelines, Patterns & Trade-Offs

graph LR
    A["Operational<br/>System"]
    B["Extract<br/>(Full + Incremental)"]
    C["Transform<br/>(Central Server)"]
    D["Load<br/>(into Warehouse)"]
    E["Warehouse<br/>(Clean Data)"]

    A -->|"1. Extract"| B
    B -->|"2. Transform<br/>Normalize, Aggregate,<br/>Apply Business Logic"| C
    C -->|"3. Load"| D
    D --> E

graph LR
    A["Operational<br/>System"]
    B["Extract<br/>(Full + Incremental)"]
    C["Load<br/>(Raw into Warehouse)"]
    D["Transform<br/>(SQL/dbt in Warehouse)"]
    E["Warehouse<br/>(All Data + Clean)"]

    A -->|"1. Extract"| B
    B -->|"2. Load Raw<br/>No transformation"| C
    C -->|"3. Transform<br/>Dimensional models,<br/>Business logic"| D
    D --> E

graph LR
    A["Operational<br/>DB"]
    B["Airflow DAG<br/>Daily at 2 AM"]
    C["Extract<br/>Incremental<br/>Last 24h"]
    D["Transform<br/>or Load-Transform"]
    E["Warehouse"]

    B -->|"Trigger"| C
    C -->|"Read new data"| A
    C --> D
    D -->|"Write"| E

graph LR
    A["Operational<br/>System"]
    B["Event Stream<br/>Kafka, Pub/Sub"]
    C["Streaming<br/>Consumer<br/>Flink, Spark"]
    D["Cloud Storage<br/>(Staging)"]
    E["Warehouse<br/>(Micro-batches)"]

    A -->|"Emit events<br/>in real-time"| B
    B --> C
    C -->|"Buffer 5-min<br/>windows"| D
    D -->|"Bulk insert<br/>every 5 min"| E

Real-time events: Stream → Staging table (1-min micro-batches)
Batch corrections: Daily batch → Reconciliation + SCD Type 2

graph LR
    A["Operational<br/>DB"]
    B["CDC Tool<br/>Debezium,<br/>GCP Datastream"]
    C["Event Stream<br/>Kafka,<br/>Change Log"]
    D["Warehouse<br/>Apply Changes<br/>Upsert/Delete"]

    A -->|"Transaction Logs"| B
    B -->|"Parse deltas"| C
    C -->|"Apply in order"| D

Run 1: Extract 1000 rows, Load to warehouse ✓
Run 2: Network fails after load, before commit
Run 3: Retry load...
  → Old system: Loads 1000 rows again (duplicates!)
  → Idempotent system: Detects already loaded, skips

# Airflow: Track last extracted timestamp
last_run_timestamp = get_state('customer_last_extract')
→ SELECT * FROM customers WHERE updated_at > '{last_run_timestamp}'
→ Load to warehouse
→ set_state('customer_last_extract', current_timestamp)

# If load fails and we retry:
→ Timestamp hasn't updated, so re-extraction gets same data
→ Warehouse deduplication handles duplicates

BEGIN TRANSACTION;
  DELETE FROM silver.customers WHERE date_loaded = TODAY();
  INSERT INTO silver.customers 
    SELECT * FROM bronze.customers
    WHERE date_loaded = TODAY();
COMMIT;

-- Merge is idempotent: running twice = running once
MERGE INTO gold.dim_customer t
USING silver.customers s
  ON t.customer_id = s.customer_id 
   AND t.is_current = TRUE
WHEN MATCHED AND (t.email != s.email) THEN
  UPDATE SET 
    is_current = FALSE,
    end_date = TODAY()
WHEN NOT MATCHED THEN
  INSERT (customer_id, email, ..., is_current)
  VALUES (s.customer_id, s.email, ..., TRUE);

Fact table loaded at 2 AM:
  Sale ID 12345: Date 2024-01-15, Amount 100

At 6 AM, source system correction arrives:
  Sale ID 12345: Actually Date 2024-01-10 (was in wrong day batch)

Problem: Daily aggregations for 2024-01-10 are already published!

-- Nightly 2 AM: Load daily fact
INSERT INTO gold.fact_sales
SELECT * FROM silver.sales 
WHERE sale_date = CURRENT_DATE() - 1;

-- Late-arriving: If date is old, delete & reload
MERGE INTO gold.fact_sales t
USING (
  SELECT * FROM silver.sales 
  WHERE sale_date BETWEEN CURRENT_DATE() - 7 AND CURRENT_DATE()
) s
  ON t.sale_key = s.sale_key
WHEN MATCHED THEN UPDATE SET t.* = s.*
WHEN NOT MATCHED THEN INSERT *;

-- fact_sales: Original data
CREATE TABLE gold.fact_sales (...);

-- fact_sales_corrections: Override table for late changes
CREATE TABLE gold.fact_sales_corrections (
  sale_key STRING,
  new_amount DECIMAL,
  new_date DATE,
  reason STRING,
  effective_date DATE
);

-- Query layer: Union & apply corrections
SELECT 
  f.*,
  COALESCE(c.new_amount, f.amount) as amount_corrected,
  COALESCE(c.new_date, f.date) as date_corrected
FROM gold.fact_sales f
LEFT JOIN gold.fact_sales_corrections c 
  ON f.sale_key = c.sale_key 
  AND c.effective_date <= CURRENT_DATE();

-- Store both timestamps
CREATE TABLE silver.sales (
  event_time TIMESTAMP,     -- When sale happened
  processing_time TIMESTAMP, -- When we received it

  -- Allows late-arriving analysis
) CLUSTER BY event_time;

-- Late-arriving: Data with old event_time, new processing_time
SELECT 
  DATE(event_time) as sale_date,
  COUNT(*) as count_by_event_time,
  COUNT(CASE WHEN processing_time > DATE(event_time) + 1 DAY 
    THEN 1 END) as late_arriving_count
FROM silver.sales
GROUP BY 1;

Bronze (Raw) → Quality Check → Silver (Clean) → Validation → Gold (Ready)
                    ↓                              ↓              ↓
            - Duplicates?        - Nulls expected?   - FK valid?
            - Schema mismatch?   - Out of range?     - Aggregates match?
            - Malformed dates?   - Format correct?   - No orphans?

# dbt: Declarative quality checks
version: 2

models:
  - name: silver_customers
    columns:
      - name: customer_id
        tests:
          - unique         # No duplicates
          - not_null       # Always present

      - name: email
        tests:
          - unique
          - not_null
          - regex_match:
              pattern: '^.+@.+\..+$'  # Valid email format

      - name: age
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 150