Vraag tien data engineers wat ETL is en je krijgt tien antwoorden. De afkorting staat voor Extract, Transform, Load — data uit bronnen halen, transformeren naar het juiste formaat, en laden in het warehouse. Klinkt eenvoudig. In de praktijk gaan hier de meeste projecten kapot, want de hoeken zitten in robuustheid, herstartbaarheid en kosten.

ETL versus ELT

De volgorde van de operaties is veranderd door de cloud. In de pre-cloud-era was compute schaars: je transformeerde data in een tussenstap (op een ETL-server), omdat het warehouse te duur was om die werkdruk te dragen. Met Snowflake, BigQuery en consorten heeft ELT de overhand genomen: laad de ruwe data eerst in het warehouse, en transformeer met SQL in het warehouse zelf. dbt is hier de standaardtool voor.

Aspect	ETL (traditioneel)	ELT (modern)
Transformatie	In ETL-engine (Informatica, SSIS)	In warehouse (SQL/dbt)
Tussenopslag	Vaak file-based / staging server	Stagingtabellen in warehouse
Schaal	Beperkt door ETL-server	Onbeperkt (warehouse compute)
Schema-on	Write	Vaak read (raw blijft ruw)
Tools	Informatica, Talend, SSIS	Fivetran/Airbyte + dbt
Kostenmodel	Licenties + servers	Pay-per-query

Extract: data uit de bron halen

De extract-stap kent drie hoofdstrategieën:

• Full extract — kopieer elke run de hele bron. Eenvoudig, maar duur en traag bij grote datasets.
• Incremental extract — haal alleen rijen op die zijn gewijzigd sinds de laatste run, op basis van een watermark (bijv. updated_at).
• Change Data Capture (CDC) — abonneer op de transactielog van de bron-database (bijv. via Debezium, AWS DMS, Fivetran). Detecteert ook deletes, wat watermark-gebaseerde extracts missen.

Een eenvoudige incremental extract in Python:

import pandas as pd
from sqlalchemy import create_engine

src = create_engine("postgresql://user:pwd@source-db/app")
dwh = create_engine("snowflake://user:pwd@account/warehouse")

# Lees laatste watermark uit DWH metadata-tabel
with dwh.begin() as cx:
    last_ts = cx.execute(
        "SELECT MAX(load_ts) FROM meta.last_load WHERE table_name = 'orders'"
    ).scalar() or "1900-01-01"

# Haal alleen gewijzigde rijen op
query = f"""
    SELECT * FROM orders
    WHERE updated_at > '{last_ts}'
"""
df = pd.read_sql(query, src)

# Schrijf naar staging
df.to_sql("stg_orders", dwh, if_exists="append", index=False)

# Update watermark
with dwh.begin() as cx:
    cx.execute(
        "INSERT INTO meta.last_load VALUES ('orders', :ts)",
        {"ts": df["updated_at"].max()}
    )

print(f"{len(df)} rijen geladen, watermark naar {df['updated_at'].max()}")

CDC met Debezium en Kafka — een minimaal opzet

Voor een betrouwbare CDC-pipeline op een PostgreSQL-bron leg je een Debezium-connector op de transactielog (WAL) en publiceer je de wijzigingen op een Kafka-topic. Een sink-connector schrijft elke wijziging naar een staging-tabel in het warehouse:

# debezium-postgres-connector.yaml
name: orders-cdc
config:
  connector.class: io.debezium.connector.postgresql.PostgresConnector
  database.hostname: source-db.internal
  database.port: 5432
  database.user: cdc_reader
  database.dbname: app
  database.server.name: app_prod
  table.include.list: public.orders,public.customers
  plugin.name: pgoutput
  publication.name: dbz_app_pub
  slot.name: dbz_app_slot
  tombstones.on.delete: true
  topic.prefix: cdc

De CDC-events bevatten op ("c" = create, "u" = update, "d" = delete) en zowel de before- als after-state van elke rij. Daarmee kun je in silver een correcte SCD2-historie reconstrueren, ook bij deletes — iets wat een watermark-extract structureel mist.

Transform: opschonen en harmoniseren

De transform-laag doet meer dan formatteren. Typische taken:

• Cleansing — trimmen, casing, null-handling, datatype-conversies.
• Deduplicatie — late-arriving data, retries, dubbele extracts.
• Joining / lookups — bron-IDs vertalen naar surrogate keys.
• Business rules — een "actieve klant" is iemand met een order in laatste 90 dagen, etc.
• Aggregations — pre-aggregeren voor performance.
• Versioning — SCD Type 2 historie bouwen.

Hier een dbt-model dat orders dedupliceert en business logic toepast:

-- models/silver/orders.sql
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

WITH source AS (
    SELECT * FROM {{ ref('stg_orders') }}
    {% if is_incremental() %}
      WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
    {% endif %}
),
deduped AS (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) AS rn
    FROM source
)
SELECT
    order_id,
    customer_id,
    order_date::DATE                    AS order_date,
    UPPER(TRIM(status))                 AS status,
    COALESCE(total_amount, 0)           AS total_amount,
    CASE WHEN total_amount > 1000 THEN 'high'
         WHEN total_amount > 100  THEN 'medium'
         ELSE 'low'
    END                                 AS order_size_tier,
    updated_at
FROM deduped
WHERE rn = 1

Load: idempotency is alles

Een load is idempotent als je hem twee keer kunt draaien zonder dat de uitkomst verandert. Bij een retry, een netwerkstoring of een herstart: de tabel moet er hetzelfde uit blijven zien als bij één geslaagde run.

Drie veelgebruikte loadstrategieën:

• Append-only — voeg nieuwe rijen toe, raak bestaande niet aan. Werkt voor immutable feiten (verkooptransacties).
• Upsert / MERGE — update als bestaand, anders insert. Voor mutable entiteiten (klanten, producten).
• Full refresh — drop en herlaad. Voor kleine dimensies of als de bron geen updated_at heeft.

De universele MERGE-statement (werkt in Snowflake, BigQuery, Redshift, Synapse):

MERGE INTO dim_customer AS tgt
USING stg_customers AS src
   ON tgt.customer_id = src.customer_id

WHEN MATCHED AND (
       tgt.name      <> src.name
    OR tgt.segment   <> src.segment
    OR tgt.country   <> src.country
) THEN UPDATE SET
       name      = src.name,
       segment   = src.segment,
       country   = src.country,
       updated_at = CURRENT_TIMESTAMP

WHEN NOT MATCHED THEN INSERT (customer_id, name, segment, country, created_at)
   VALUES (src.customer_id, src.name, src.segment, src.country, CURRENT_TIMESTAMP);

De WHEN MATCHED AND-clausule is een belangrijk detail: zonder die check update je elke rij elke run, ook als er niets veranderd is. Dat geeft onnodige churn, lastigere debugging en hogere kosten.

Slowly Changing Dimensions in praktijk

Voor SCD Type 2 (historie bewaren) wordt het complexer. Hoofdstuk 5 gaat hier diep op in, maar als voorproefje:

-- 1. Sluit verlopen versies
UPDATE dim_customer
SET    valid_to = CURRENT_DATE - 1,
       is_current = FALSE
FROM   stg_customers s
WHERE  dim_customer.customer_id = s.customer_id
  AND  dim_customer.is_current = TRUE
  AND  (dim_customer.segment <> s.segment OR dim_customer.country <> s.country);

-- 2. Voeg nieuwe versie toe
INSERT INTO dim_customer (customer_id, name, segment, country, valid_from, valid_to, is_current)
SELECT s.customer_id, s.name, s.segment, s.country, CURRENT_DATE, '9999-12-31', TRUE
FROM   stg_customers s
LEFT JOIN dim_customer d
  ON d.customer_id = s.customer_id AND d.is_current = TRUE
WHERE  d.customer_id IS NULL
   OR  d.segment <> s.segment
   OR  d.country <> s.country;

Orkestratie

Je pipelines moeten ergens worden gestart, in de juiste volgorde, met dependencies. Populaire orkestrators:

• Apache Airflow — de standaard, Python-DAGs, breed ecosysteem.
• Prefect / Dagster — moderner, type-safe, betere developer experience.
• Azure Data Factory — visueel, native in Microsoft-stack.
• dbt Cloud / Cron — voor SQL-only transformaties.
• GitHub Actions / Cloud Composer — voor lichte workloads.

Een Airflow-DAG voor een dagelijkse load:

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator

with DAG(
    dag_id="dwh_daily_load",
    start_date=datetime(2026, 1, 1),
    schedule="0 4 * * *",          # elke nacht om 04:00
    catchup=False,
    max_active_runs=1,             # voorkom overlappende runs
    default_args={"retries": 2, "retry_delay": 300},
) as dag:

    extract_orders   = PythonOperator(task_id="extract_orders",   python_callable=run_extract, op_args=["orders"])
    extract_customers = PythonOperator(task_id="extract_customers", python_callable=run_extract, op_args=["customers"])

    transform = DbtCloudRunJobOperator(task_id="dbt_build", job_id=12345)

    quality = PythonOperator(task_id="quality_checks", python_callable=run_ge_suite)

    [extract_orders, extract_customers] >> transform >> quality

Foutafhandeling die echt werkt

• Retries met exponential backoff — netwerkstoringen lossen zichzelf op.
• Dead Letter Queue — rijen die niet door validatie komen apart bewaren, niet weggooien.
• Atomic loads — laad in een staging-tabel, doe een atomic SWAP of rename. Geen halve states.
• Alerting — een gefaalde run om 04:00 mag niet pas om 09:00 ontdekt worden.
• Re-runbaarheid — kun je zonder repercussies een dag opnieuw laden?

Backfill: hoe je historie veilig opnieuw laadt

Vroeg of laat ontdek je een bug in je transformaties die het laatste kwartaal aan data corrumpeert. Of een bronsysteem doet retroactieve correcties op rijen van vorige maand. In beide gevallen heb je een betrouwbare backfill-strategie nodig:

1. Parametriseer je pipeline op een datumbereik, niet op "gisteren". Een DAG die --start-date en --end-date accepteert is herbruikbaar voor backfills, hercorrupted dagen en initial loads.
2. Bewaar bronhistorie in bronze. Als bronze append-only is, kun je elke transformatie deterministisch opnieuw draaien zonder de bron opnieuw te bevragen.
3. Gebruik partitioning op load-datum in silver/gold. Een backfill kan dan veilig één partitie overschrijven zonder de rest te raken — bijvoorbeeld via Snowflake's DELETE + INSERT in transactie, of BigQuery's partition-overwrite.
4. Schaal compute tijdens de backfill omhoog. Op Snowflake een grotere warehouse-grootte, op Databricks meer workers. Backfills zijn intensiever dan reguliere loads.

Performance-tips voor zware loads

• Bulk in plaats van row-by-row — een COPY INTO is 100-1000× sneller dan losse INSERT-statements.
• Cluster keys / partitioning — partitioneer fact tables op event-datum, niet op load-datum.
• Vermijd UDFs in hot paths — een Python-UDF in een SQL-statement breekt vectorisatie. Schrijf het als pure SQL waar het kan.
• Monitor data skew — een join op een kolom met 95% NULL-waarden gaat één worker overbelasten. Debug met EXPLAIN.
• Materialise tussenresultaten als ze door meerdere downstream models gebruikt worden — beter één keer berekenen dan tien keer.

Anti-patterns

• "Truncate + reload" in productie tijdens werkuren — zichtbare gaps voor gebruikers.
• Hardcoded watermarks in SQL — onmogelijk om historisch opnieuw te laden.
• Geen logging — als het pipeline-debug-uur slaat, weet je niets meer terug.
• Eén grote DAG die uren draait — maak hem modulair, met losse tabellen die je apart kunt herladen.
• Business logic in extract-laag — de extract moet ruwheid bewaren, transformatie hoort in silver.

Veelgestelde vragen