dbt + Python Models: Transforms Voorbij SQL

Inleiding: Voorbij de Grenzen van SQL

SQL is en blijft de ruggengraat van data-transformaties. Maar wie al een tijdje met dbt werkt, kent de situaties waarin SQL simpelweg niet meer volstaat: complexe machine learning feature engineering, statistische berekeningen, API-aanroepen tijdens transformaties, of het gebruik van Python-bibliotheken zoals scikit-learn, prophet of statsmodels. Precies voor dit soort use cases introduceerde dbt Labs in versie 1.3 de ondersteuning voor Python models — en sindsdien is het één van de meest besproken features in de data engineering community.

In 2026 is de adoptie van dbt Python models flink gegroeid. Cloud-platformen zoals Databricks, Snowflake en BigQuery bieden native Python-uitvoering, waardoor data engineers nu hun volledige transformatielogica — zowel SQL als Python — kunnen beheren binnen één dbt-project. Dat betekent één lineage-graph, één governance-model, één CI/CD-pipeline. De belofte: het beste van twee werelden.

In deze blog duiken we diep in hoe dbt Python models werken, welke platformen ze ondersteunen, hoe je realistische transformaties schrijft met pandas en Spark, en wanneer je ze wél of juist niet moet inzetten. Met concrete codevoorbeelden, best practices en een eerlijke vergelijking met alternatieven.

Hoe Werken dbt Python Models?

Voordat we code schrijven, is het belangrijk te begrijpen hoe dbt Python models onder de motorkap werken. Het verschilt fundamenteel van SQL models, en dat heeft implicaties voor performance, debugging en deployment.

Platform Support in 2026

Praktische Codevoorbeelden

Voorbeeld 1: Basis Python Model (Snowflake / Snowpark)

Het eenvoudigste Python model: een bestaand SQL-model ophalen, een kolom toevoegen met Python-logica, en het resultaat terugschrijven.

# models/python/customer_enriched.py

import pandas as pd

def model(dbt, session):
    # Configureer het model als tabel
    dbt.config(
        materialized="table",
        packages=["pandas", "scikit-learn"]
    )

    # Refereer naar een upstream SQL model
    customers_df = dbt.ref("stg_customers").to_pandas()

    # Voeg een klant-segment toe op basis van LTV-berekening
    def bereken_segment(row):
        if row["lifetime_value"] >= 5000:
            return "Platinum"
        elif row["lifetime_value"] >= 1000:
            return "Gold"
        elif row["lifetime_value"] >= 200:
            return "Silver"
        else:
            return "Bronze"

    customers_df["klant_segment"] = customers_df.apply(
        bereken_segment, axis=1
    )

    # Bereken percentielrang binnen segment
    customers_df["ltv_percentiel"] = customers_df.groupby("klant_segment")[
        "lifetime_value"
    ].rank(pct=True).round(3)

    return customers_df

Voorbeeld 2: Tijdreeksforecast met Prophet (Databricks / PySpark)

Een realistisch use case: verkoopcijfers per product forecasten met Facebook Prophet, direct in dbt.

# models/python/sales_forecast.py

def model(dbt, session):
    dbt.config(
        materialized="table",
        packages=["prophet", "pandas"]
    )

    import pandas as pd
    from prophet import Prophet

    # Haal historische verkoopdata op (SQL model upstream)
    sales_df = dbt.ref("fct_daily_sales").to_pandas()
    sales_df["ds"] = pd.to_datetime(sales_df["order_date"])
    sales_df["y"] = sales_df["revenue"]

    resultaten = []

    # Forecast per product_category
    for categorie in sales_df["product_category"].unique():
        subset = sales_df[sales_df["product_category"] == categorie][["ds", "y"]].copy()

        if len(subset) < 30:  # Minimale datavereiste
            continue

        # Train Prophet model
        m = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            changepoint_prior_scale=0.05
        )
        m.fit(subset)

        # Forecast 90 dagen vooruit
        toekomst = m.make_future_dataframe(periods=90)
        forecast = m.predict(toekomst)

        forecast["product_category"] = categorie
        forecast["model_versie"] = "prophet_v1"
        forecast["gegenereerd_op"] = pd.Timestamp.now()

        resultaten.append(
            forecast[["ds", "yhat", "yhat_lower", "yhat_upper",
                      "product_category", "model_versie", "gegenereerd_op"]]
        )

    return pd.concat(resultaten, ignore_index=True)

Voorbeeld 3: Feature Engineering voor ML (Snowflake Snowpark)

Snowpark native DataFrames gebruiken voor schaalbare feature engineering zonder conversie naar pandas:

# models/python/ml_features_transacties.py

from snowflake.snowpark import functions as F
from snowflake.snowpark.window import Window

def model(dbt, session):
    dbt.config(materialized="table")

    # Laad transactiedata als Snowpark DataFrame (lazy evaluation!)
    transacties = dbt.ref("fct_transacties")

    # Window-functies voor rolling aggregaties
    klant_window_7d = Window.partition_by("klant_id").order_by(
        F.col("transactie_datum")
    ).rows_between(-6, 0)

    klant_window_30d = Window.partition_by("klant_id").order_by(
        F.col("transactie_datum")
    ).rows_between(-29, 0)

    # Feature berekeningen — volledig in Snowflake gepushed
    features = transacties.with_column(
        "gem_bedrag_7d", F.avg("bedrag").over(klant_window_7d)
    ).with_column(
        "gem_bedrag_30d", F.avg("bedrag").over(klant_window_30d)
    ).with_column(
        "aantal_transacties_7d", F.count("transactie_id").over(klant_window_7d)
    ).with_column(
        "bedrag_volatiliteit_30d", F.stddev("bedrag").over(klant_window_30d)
    ).with_column(
        "is_weekend", F.dayofweek(F.col("transactie_datum")).isin([0, 6])
    ).select(
        "klant_id",
        "transactie_datum",
        "transactie_id",
        "gem_bedrag_7d",
        "gem_bedrag_30d",
        "aantal_transacties_7d",
        "bedrag_volatiliteit_30d",
        "is_weekend"
    )

    return features

Voorbeeld 4: Incrementele Python Model

Python models ondersteunen ook incrementele materialisatie — essentieel voor grote datasets in productie:

# models/python/events_scored.py

def model(dbt, session):
    dbt.config(
        materialized="incremental",
        unique_key="event_id",
        on_schema_change="sync_all_columns",
        packages=["scikit-learn", "joblib", "pandas"]
    )

    import pandas as pd
    import joblib

    # Haal nieuwe events op (incrementeel)
    events_df = dbt.ref("stg_events").to_pandas()

    if dbt.is_incremental:
        # Filter op alleen nieuwe records
        # (dbt beheert de merge-logica op basis van unique_key)
        pass  # dbt.is_incremental filtert automatisch

    # Laad pre-trained model uit stage (Snowflake internal stage)
    # of gebruik een lichtgewicht berekening
    events_df["fraud_score"] = (
        events_df["bedrag"] * 0.3 +
        events_df["snelheid_kmh"].fillna(0) * 0.7
    ).clip(0, 1)

    events_df["risico_label"] = events_df["fraud_score"].apply(
        lambda s: "HOOG" if s > 0.7 else ("MIDDEL" if s > 0.4 else "LAAG")
    )

    events_df["gescoord_op"] = pd.Timestamp.now()

    return events_df

Vergelijking met Alternatieven

dbt Python models zijn niet de enige manier om Python-transformaties in je data pipeline te integreren. Hier een eerlijke vergelijking met de meest gebruikte alternatieven in 2026:

Praktijkvoorbeeld: E-commerce Churn Scoring

Best Practices & Production Tips

1. Houd Python Models Lean — SQL doet het zware werk

Python models zijn duurder (compute-tijd, kosten) dan SQL models. Gebruik SQL voor filtering, joining en aggregaties. Gebruik Python alleen voor wat SQL niet kan: bibliotheken, complexe algoritmen, externe API-calls. Een stelregel: als het in SQL kan, doe het in SQL.

2. Gebruik Platform-Native DataFrames Zo Lang Mogelijk

# ❌ Vermijd dit op grote datasets:
df = dbt.ref("grote_tabel").to_pandas()  # Trekt alles naar memory!

# ✅ Gebruik Snowpark/PySpark native:
df = dbt.ref("grote_tabel")  # Lazy, pushdown naar platform
df_gefilterd = df.filter(F.col("datum") >= "2025-01-01")
df_geaggregeerd = df_gefilterd.group_by("klant_id").agg(F.sum("bedrag"))
# Alleen op het einde converteren als nodig voor Python libs:
resultaat = df_geaggregeerd.to_pandas()

3. Pinnen van Package Versies

# models/python/mijn_model.py
def model(dbt, session):
    dbt.config(
        materialized="table",
        # Pinnen voorkomt breaking changes in productie!
        packages=["scikit-learn==1.4.2", "pandas==2.2.1", "prophet==1.1.5"]
    )

4. Incrementele Modellen voor Grote Datasets

Gebruik altijd materialized="incremental" voor Python models die op grote tabellen werken. Full refreshes van miljoenen rijen via Python zijn duur en langzaam. Definieer een duidelijke unique_key en filter-logica.

5. Foutafhandeling en Logging

def model(dbt, session):
    dbt.config(materialized="table")
    import pandas as pd
    import logging

    logger = logging.getLogger(__name__)

    df = dbt.ref("stg_transacties").to_pandas()

    # Valideer inputdata voordat je transformeert
    if df.empty:
        raise ValueError("Upstream model stg_transacties heeft geen data!")

    if df["bedrag"].isna().sum() / len(df) > 0.1:
        logger.warning(
            f"Meer dan 10% null-waarden in 'bedrag': "
            f"{df['bedrag'].isna().sum()} rijen"
        )

    # Transformatielogica...
    return df

6. Scheiding van Model Code en Business Logica

Voor complexe Python models: extraheer business logica naar aparte Python-functies of klassen. Dit maakt unit-testing buiten dbt mogelijk en houdt het model-bestand overzichtelijk.

7. Documenteer Python Models in schema.yml

# models/schema.yml
models:
  - name: churn_scores
    description: "Dagelijkse churn-scores per klant, berekend met gradient boosting model."
    config:
      materialized: table
    columns:
      - name: klant_id
        description: "Primaire sleutel"
        tests:
          - not_null
          - unique
      - name: churn_kans
        description: "Kans op churn in de komende 30 dagen (0.0 - 1.0)"
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1

Lokaal Testen en Debuggen

Een bekende frustratie: dbt Python models kun je niet lokaal runnen zoals reguliere Python-scripts — ze draaien in de cloud. Dit maakt debugging complexer. Hier zijn strategieën om dit te mitigeren:

Conclusie: Wanneer Wel of Niet Gebruiken?

dbt Python models zijn in 2026 volwassen genoeg voor productiegebruik op de grote cloud-platformen. Ze lossen een echte pijn op: de kloof tussen SQL-transformaties en Python-gebaseerde data science logica. Door alles in één dbt-project te beheren, win je aan governance, observability en ontwikkelaarservaring.

De toekomst van dbt Python models ziet er veelbelovend uit: betere lokale development experiences, bredere platform-ondersteuning en diepere integratie met ML-platformen staan op de roadmap van dbt Labs. Wie nu investeert in het leren van Python models, positioneert zich sterk voor de data engineering uitdagingen van de komende jaren.