ETL Design Skill

Design ETL (Extract-Transform-Load) and ELT (Extract-Load-Transform) workflows. Cover extraction strategies, transformation patterns, loading approaches, schema management, and orchestration.

Workflow

Step 1: Choose ETL vs ELT

Recommendation: Use ELT for most modern data warehouse projects. Use ETL when transformations require complex logic not easily expressed in SQL, or when data must be cleansed before it is safe to store.

Step 2: Design Extraction

Source Types and Strategies

Incremental Extraction Patterns

Timestamp-based:

-- Extract records modified since last successful run
SELECT *
FROM source_table
WHERE updated_at > :last_watermark
  AND updated_at <= :current_timestamp
ORDER BY updated_at

Log-based (CDC):

Source DB -> WAL/Binlog -> Debezium -> Kafka -> Warehouse

Sequence-based:

-- Use an auto-incrementing ID for append-only tables
SELECT *
FROM events
WHERE event_id > :last_max_id
ORDER BY event_id

Full refresh (when incremental is not possible):

-- Replace entire table; use for small dimension tables
TRUNCATE TABLE target_table;
INSERT INTO target_table SELECT * FROM source_table;

Step 3: Design the Transformation Layer

Use a layered transformation architecture:

Raw/Landing -> Staging -> Intermediate -> Marts/Presentation

Layer 1: Raw / Landing

• Exact copy of source data
• No transformations
• Append-only with load timestamp
• Serves as audit trail and reprocessing source

-- Raw table structure
CREATE TABLE raw.orders (
    _loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP(),
    _source_file STRING,
    _raw_data VARIANT  -- Store entire payload as semi-structured
);

Layer 2: Staging

• Cleaned and typed
• Deduplicated
• Column names standardized
• Source-specific logic applied

-- dbt staging model pattern
-- models/staging/stg_orders.sql
WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
),
renamed AS (
    SELECT
        order_id::INTEGER AS order_id,
        customer_id::INTEGER AS customer_id,
        TRIM(status) AS order_status,
        amount::DECIMAL(12,2) AS order_amount_usd,
        TO_TIMESTAMP(created_at) AS created_at,
        TO_TIMESTAMP(updated_at) AS updated_at
    FROM source
),
deduplicated AS (
    SELECT *
    FROM renamed
    QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) = 1
)
SELECT * FROM deduplicated

Layer 3: Intermediate

• Business logic applied
• Joins across sources
• Calculations and derived fields
• Grain transformations

-- models/intermediate/int_order_items_enriched.sql
SELECT
    oi.order_item_id,
    oi.order_id,
    o.customer_id,
    o.order_status,
    p.product_name,
    p.category,
    oi.quantity,
    oi.unit_price,
    oi.quantity * oi.unit_price AS line_total,
    CASE
        WHEN o.order_status = 'refunded' THEN 0
        WHEN o.order_status = 'partially_refunded' THEN oi.quantity * oi.unit_price * (1 - o.refund_pct)
        ELSE oi.quantity * oi.unit_price
    END AS net_revenue
FROM {{ ref('stg_order_items') }} oi
JOIN {{ ref('stg_orders') }} o ON oi.order_id = o.order_id
JOIN {{ ref('stg_products') }} p ON oi.product_id = p.product_id

Layer 4: Marts / Presentation

• Business-user-facing tables
• Pre-aggregated for common queries
• Documented with descriptions
• Optimized for BI tool access

-- models/marts/fct_daily_revenue.sql
SELECT
    DATE_TRUNC('day', o.created_at) AS revenue_date,
    p.category AS product_category,
    c.segment AS customer_segment,
    COUNT(DISTINCT o.order_id) AS order_count,
    COUNT(DISTINCT o.customer_id) AS customer_count,
    SUM(oi.net_revenue) AS net_revenue,
    SUM(oi.net_revenue) / NULLIF(COUNT(DISTINCT o.order_id), 0) AS avg_order_value
FROM {{ ref('int_order_items_enriched') }} oi
JOIN {{ ref('stg_orders') }} o ON oi.order_id = o.order_id
JOIN {{ ref('dim_customers') }} c ON o.customer_id = c.customer_id
JOIN {{ ref('dim_products') }} p ON oi.product_id = p.product_id
GROUP BY 1, 2, 3

Step 4: Design Loading Strategy

dbt materialization mapping:

# dbt_project.yml
models:
  staging:
    +materialized: view           # Always fresh, low cost
  intermediate:
    +materialized: ephemeral      # Inline, no table created
  marts:
    facts:
      +materialized: incremental  # Append new data
      +incremental_strategy: merge
      +unique_key: id
    dimensions:
      +materialized: table        # Full refresh daily

Step 5: Handle Schema Evolution

Prepare for upstream schema changes:

def detect_schema_changes(current_schema, expected_schema):
    """Detect and categorize schema changes."""
    current_cols = set(current_schema.keys())
    expected_cols = set(expected_schema.keys())

    changes = {
        'new_columns': current_cols - expected_cols,      # Added upstream
        'dropped_columns': expected_cols - current_cols,   # Removed upstream
        'type_changes': [],                                # Type changed
    }

    for col in current_cols & expected_cols:
        if current_schema[col] != expected_schema[col]:
            changes['type_changes'].append({
                'column': col,
                'was': expected_schema[col],
                'now': current_schema[col]
            })

    return changes

Schema evolution strategies:

• Additive only: Accept new columns, alert on drops or type changes
• Schema-on-read: Store raw data as JSON/VARIANT, parse in transformations
• Versioned schemas: Maintain schema registry, version transformations
• Contract testing: Fail pipeline if schema violates contract

Step 6: Design Orchestration

# Airflow DAG pattern
extract_orders >> validate_orders >> load_raw_orders
extract_customers >> validate_customers >> load_raw_customers

[load_raw_orders, load_raw_customers] >> run_dbt_staging
run_dbt_staging >> run_dbt_intermediate >> run_dbt_marts
run_dbt_marts >> run_data_tests >> notify_on_completion

Key orchestration principles:

• Define dependencies explicitly (DAG)
• Set appropriate timeouts for each task
• Configure retry policies (count, delay, backoff)
• Use sensors or triggers for event-driven starts
• Separate extraction from transformation schedules when needed

Output Format

Deliver an ETL design document with:

1. Architecture overview showing data flow from source to presentation
2. Extraction design per source with incremental strategy
3. Transformation layer design with model lineage
4. Loading strategy with materialization choices
5. Schema evolution plan
6. Orchestration DAG with dependencies and schedules
7. Testing strategy with data quality checks at each layer

Examples

Example 1: SaaS Analytics ETL

Fivetran syncs from Stripe, Salesforce, and product DB. dbt transforms through staging, intermediate, and mart layers. Daily full refresh for dimensions, incremental for facts.

Example 2: Event Stream ELT

Kafka events land in S3, Snowpipe loads to raw. dbt transforms events into session-level and user-level aggregations. Micro-batch with 15-minute freshness SLA.

Example 3: Financial Data ETL

Extract from ERP via JDBC, validate against source row counts, transform with fiscal calendar mapping, load with partition-based idempotency, reconcile totals post-load.

Best Practices

1. Land raw data first. Always store the unmodified source data before transforming.
2. Version your transformations. Use dbt or version-controlled SQL so changes are auditable.
3. Test at every layer. Schema tests on staging, business logic tests on intermediate, freshness tests on marts.
4. Document lineage. Every column in a mart table should be traceable back to its source.
5. Design for reprocessing. When transformation logic changes, you should be able to rebuild from raw.
6. Separate concerns. One model per transformation step. Do not combine extraction, cleaning, and business logic.
7. Use incremental where possible. Full refreshes become expensive as data grows.
8. Monitor data freshness. SLA breaches should trigger alerts before stakeholders notice.
9. Handle late-arriving data. Design a strategy for records that arrive after their partition was processed.
10. Plan for backfills from day one. Make it easy to reprocess a date range when business logic changes.