Il Problema dei Modelli Incrementali Naif

Il modello incrementale base visto nel modulo 04 - filtro su MAX(timestamp) e merge per unique_key - è sufficiente per molti casi d'uso. Ma quando il volume cresce oltre il miliardo di righe, quando i dati arrivano in ritardo di ore o giorni, o quando la sorgente è un sistema CDC che emette operazioni di tipo INSERT/UPDATE/DELETE, servono pattern più sofisticati.

Questo modulo esplora tre scenari avanzati: la strategia microbatch (introdotta in dbt 1.9) per dataset time-series enormi, l'integrazione con il Change Data Capture per replicare fedelmente le modifiche da database sorgente, e le tecniche per garantire l'idempotenza in presenza di late-arriving data.

Recap: Le Strategie Incrementali Standard

Prima di procedere con i pattern avanzati, ricapitoliamo il quadro completo delle strategie disponibili in dbt 1.8+:

-- Le 4 strategie standard (+ microbatch in 1.9):

-- 1. append: solo INSERT, nessun UPDATE, nessuna deduplicazione
-- Casi d'uso: audit log, event stream immutabili, fact tables append-only
{{ config(materialized='incremental', incremental_strategy='append') }}

-- 2. merge: UPSERT su unique_key (default con unique_key definita)
-- Casi d'uso: dimensioni che cambiano, ordini aggiornabili, qualsiasi SCD-1
{{ config(materialized='incremental', unique_key='id', incremental_strategy='merge') }}

-- 3. delete+insert: elimina partizioni interessate e reinserisce
-- Casi d'uso: BigQuery partitioned tables, re-processing per data
{{ config(materialized='incremental', unique_key=['id', 'date'],
          incremental_strategy='delete+insert',
          partition_by={'field': 'date', 'data_type': 'date'}) }}

-- 4. insert_overwrite: sovrascrive intere partizioni (Spark, BigQuery)
-- Casi d'uso: dataset partizionati per data, Spark/Databricks
{{ config(materialized='incremental', incremental_strategy='insert_overwrite',
          partition_by={'field': 'date', 'data_type': 'date'}) }}

Microbatch: La Strategia per Dataset Time-Series Enormi

La strategia microbatch è stata introdotta in dbt 1.9 per risolvere un problema specifico: modelli incrementali time-series dove il volume per singola esecuzione è ancora troppo grande, o dove il re-processing di finestre temporali specifiche deve essere semplice e atomico.

Come Funziona Microbatch

Invece di eseguire un'unica query incrementale grande, microbatch divide l'esecuzione in lotti temporali (batch) definiti dal campo event_time e dalla batch_size. Ogni batch è processato in modo atomico e indipendente.

-- models/marts/events/fct_events_microbatch.sql
{{ config(
    materialized='incremental',
    incremental_strategy='microbatch',
    event_time='event_timestamp',    -- colonna timestamp per la suddivisione in batch
    begin='2024-01-01',              -- data di inizio per il full refresh iniziale
    batch_size='day',                -- granularità di ogni batch: hour | day | month | year
    lookback=1                       -- ri-processa anche N batch precedenti (late-arriving data)
) }}

SELECT
    event_id,
    user_id,
    session_id,
    event_type,
    page_url,
    referrer_url,
    device_type,
    country_code,
    event_timestamp,
    DATE_TRUNC('day', event_timestamp)  AS event_date,
    DATE_TRUNC('hour', event_timestamp) AS event_hour
FROM {{ ref('stg_events') }}

-- NOTA IMPORTANTE con microbatch:
-- NON usare is_incremental() con microbatch.
-- dbt gestisce automaticamente il filtro temporale su event_time.
-- La query deve essere scritta come se processasse TUTTI i dati --
-- dbt aggiunge automaticamente il filtro: WHERE event_timestamp >= batch_start AND event_timestamp < batch_end

Esecuzione e Re-Processing con Microbatch

-- Esecuzione normale: processa i batch mancanti dall'ultima run
dbt run --select fct_events_microbatch

-- Re-processare un giorno specifico (molto utile per correzioni dati):
dbt run \
  --select fct_events_microbatch \
  --event-time-start "2024-06-15" \
  --event-time-end "2024-06-16"

-- Re-processare un intervallo (es. dopo una correzione alla sorgente):
dbt run \
  --select fct_events_microbatch \
  --event-time-start "2024-06-01" \
  --event-time-end "2024-07-01"

-- Full refresh dall'inizio (usa il 'begin' configurato):
dbt run --select fct_events_microbatch --full-refresh

-- Output tipico durante l'esecuzione:
-- Running microbatch model fct_events_microbatch
--   Batch 2024-06-13 00:00:00 to 2024-06-14 00:00:00 [1 of 3]: OK (123,456 rows)
--   Batch 2024-06-14 00:00:00 to 2024-06-15 00:00:00 [2 of 3]: OK (98,234 rows)
--   Batch 2024-06-15 00:00:00 to 2024-06-16 00:00:00 [3 of 3]: OK (45,678 rows)

Vantaggi di Microbatch vs Incremental Standard

Change Data Capture: Replicare le Modifiche dal Database Sorgente

Il Change Data Capture (CDC) cattura ogni INSERT, UPDATE e DELETE che avviene nel database sorgente e lo propaga nella pipeline. È lo standard per replicare dati in tempo quasi reale senza impatto sul sistema transazionale.

Pattern CDC con Debezium + Kafka

Lo stack più comune per CDC con dbt: Debezium legge il WAL (Write-Ahead Log) di PostgreSQL o MySQL e pubblica i cambiamenti su Kafka. Un connector Kafka (Fivetran, Airbyte, o custom) carica i messaggi nella raw table del warehouse. dbt poi consolida la storia delle modifiche.

-- La raw table CDC tipica dal connettore ha questa struttura:
-- Ogni riga rappresenta un'operazione (INSERT/UPDATE/DELETE)

-- raw.orders_cdc_stream:
-- _op          VARCHAR  -- 'r' (read/snapshot), 'c' (create/insert), 'u' (update), 'd' (delete)
-- _ts_ms       BIGINT   -- timestamp dell'operazione in milliseconds
-- _after       VARIANT  -- stato DOPO la modifica (JSON) -- NULL per DELETE
-- _before      VARIANT  -- stato PRIMA della modifica (JSON) -- NULL per INSERT
-- order_id     VARCHAR  -- chiave primaria (estratta dal messaggio)
-- customer_id  VARCHAR  -- ...altri campi estratti...

-- models/staging/stg_orders_cdc.sql
-- Normalizza i messaggi CDC in una struttura flat

{{ config(
    materialized='incremental',
    unique_key=['order_id', '_ts_ms'],
    incremental_strategy='append'   -- append-only: ogni operazione è una riga
) }}

SELECT
    order_id,
    _op                                          AS operation_type,
    TO_TIMESTAMP(_ts_ms / 1000)                  AS operation_timestamp,

    -- Estrai i campi dall'JSON _after (per INSERT e UPDATE)
    CASE WHEN _op != 'd'
         THEN _after:customer_id::VARCHAR
         ELSE _before:customer_id::VARCHAR
    END                                          AS customer_id,

    CASE WHEN _op != 'd'
         THEN _after:status::VARCHAR
         ELSE NULL
    END                                          AS status,

    CASE WHEN _op != 'd'
         THEN _after:total_amount::DECIMAL(10,2)
         ELSE NULL
    END                                          AS total_amount,

    CASE WHEN _op != 'd'
         THEN _after:updated_at::TIMESTAMP
         ELSE NULL
    END                                          AS updated_at,

    _op = 'd'                                    AS is_deleted

FROM {{ source('raw', 'orders_cdc_stream') }}

{% if is_incremental() %}
WHERE TO_TIMESTAMP(_ts_ms / 1000) > (
    SELECT MAX(operation_timestamp) FROM {{ this }}
)
{% endif %}

-- models/marts/dim_orders_from_cdc.sql
-- Consolida la storia CDC: prende solo l'ultimo stato di ogni ordine

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    merge_update_columns=['customer_id', 'status', 'total_amount', 'updated_at', 'is_deleted', 'last_operation_at']
) }}

WITH latest_per_order AS (
    -- Prendi solo l'operazione più recente per ogni ordine
    SELECT
        order_id,
        customer_id,
        status,
        total_amount,
        updated_at,
        is_deleted,
        operation_timestamp AS last_operation_at,
        -- Row number per prendere solo il record più recente
        ROW_NUMBER() OVER (
            PARTITION BY order_id
            ORDER BY operation_timestamp DESC
        ) AS rn
    FROM {{ ref('stg_orders_cdc') }}

    {% if is_incremental() %}
    -- In modalità incrementale: considera le operazioni recenti + lookback
    WHERE operation_timestamp > (
        SELECT DATEADD('hour', -2, MAX(last_operation_at))
        FROM {{ this }}
    )
    {% endif %}
)
SELECT
    order_id,
    customer_id,
    status,
    total_amount,
    updated_at,
    is_deleted,
    last_operation_at
FROM latest_per_order
WHERE rn = 1
  AND NOT is_deleted  -- esclude i record cancellati dalla vista finale

Pattern SCD-2 da CDC: Mantenere la Storia Completa

-- models/marts/dim_orders_history.sql
-- SCD Type 2 completo da stream CDC: ogni cambiamento di stato viene storicizzato

{{ config(materialized='incremental', unique_key='order_state_id') }}

WITH cdc_with_prev AS (
    SELECT
        order_id,
        operation_type,
        operation_timestamp,
        status,
        total_amount,
        customer_id,
        -- Stato precedente per rilevare cambiamenti reali
        LAG(status) OVER (PARTITION BY order_id ORDER BY operation_timestamp)        AS prev_status,
        LAG(total_amount) OVER (PARTITION BY order_id ORDER BY operation_timestamp)  AS prev_total,
        -- Prossimo timestamp per calcolare valid_to
        LEAD(operation_timestamp) OVER (PARTITION BY order_id ORDER BY operation_timestamp) AS next_op_at
    FROM {{ ref('stg_orders_cdc') }}
    WHERE NOT is_deleted

    {% if is_incremental() %}
    AND operation_timestamp > (
        SELECT DATEADD('hour', -1, MAX(valid_from)) FROM {{ this }}
    )
    {% endif %}
),
filtered_changes AS (
    -- Includi solo righe dove qualcosa è effettivamente cambiato
    SELECT *
    FROM cdc_with_prev
    WHERE operation_type = 'c'                  -- nuovi inserimenti
       OR status != prev_status                  -- cambio di stato
       OR ABS(total_amount - prev_total) > 0.01  -- cambio importo
)
SELECT
    -- Chiave surrogata per questa versione del record
    {{ dbt_utils.generate_surrogate_key(['order_id', 'operation_timestamp']) }} AS order_state_id,
    order_id,
    customer_id,
    status,
    total_amount,
    operation_timestamp                          AS valid_from,
    COALESCE(next_op_at, '9999-12-31'::TIMESTAMP) AS valid_to,
    next_op_at IS NULL                           AS is_current
FROM filtered_changes

Idempotenza: La Proprietà Fondamentale

Un modello incrementale è idempotente se eseguirlo più volte con gli stessi dati produce sempre lo stesso risultato. È una garanzia critica in produzione: retry automatici, re-processing per correzioni, e full refresh non devono produrre risultati diversi.

Anti-pattern: Modelli Non Idempotenti

-- ANTI-PATTERN: questo modello NON è idempotente

{{ config(materialized='incremental') }}

SELECT
    order_id,
    customer_id,
    total_amount,
    CURRENT_TIMESTAMP AS inserted_at  -- PROBLEMA: ogni re-run produce timestamp diversi
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
WHERE order_id NOT IN (SELECT order_id FROM {{ this }})  -- PROBLEMA: lento su tabelle grandi
{% endif %}

-- Il problema: se la pipeline ri-processa lo stesso giorno per una correzione,
-- CURRENT_TIMESTAMP cambierà e avrai righe duplicate o inconsistenti.

-- CORRETTO: modello idempotente

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

SELECT
    order_id,
    customer_id,
    total_amount,
    created_at,           -- usa il timestamp dalla sorgente, non CURRENT_TIMESTAMP
    updated_at            -- timestamp di aggiornamento della sorgente
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
-- Filtro basato su updated_at (non su NOT IN, che è O(n*m)):
WHERE updated_at > (
    SELECT MAX(updated_at) FROM {{ this }}
)
{% endif %}

-- Con merge su unique_key='order_id':
-- * Re-run dello stesso range di dati: MERGE aggiornerà le righe esistenti
--   con gli stessi valori → nessun cambiamento netto (idempotente)
-- * Nuovi dati: MERGE inserirà le nuove righe
-- * Dati corretti retroattivamente: MERGE aggiornerà correttamente

Testing dell'Idempotenza

-- tests/assert_incremental_idempotent.sql
-- Verifica che eseguire il modello due volte non cambi il conteggio delle righe

-- Questo test è un placeholder: l'idempotenza vera si verifica in CI
-- confrontando il conteggio prima e dopo un secondo run sugli stessi dati.

-- Test pratico: verifica che non ci siano duplicati sulla unique_key
SELECT
    order_id,
    COUNT(*) AS occurrences
FROM {{ ref('fct_orders') }}
GROUP BY order_id
HAVING COUNT(*) > 1

-- Se questo test passa, il merge sta funzionando correttamente.

Gestione dei Late-Arriving Data

I late-arriving data sono eventi con timestamp passato che arrivano nella pipeline in ritardo. È un problema comune con sorgenti mobile (app offline), sistemi con retry, e pipeline batch che arrivano con ritardo variabile.

-- Pattern 1: Lookback window fissa
-- Semplice e robusto per la maggior parte dei casi

{{ config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge'
) }}

SELECT
    event_id,
    user_id,
    event_type,
    event_timestamp,
    received_at  -- quando è arrivato nel warehouse
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_timestamp >= (
    -- Lookback di 48 ore per catturare late-arriving data
    SELECT DATEADD('hour', -48, MAX(event_timestamp))
    FROM {{ this }}
)
{% endif %}

-- Pattern 2: Lookback adattivo basato sul lag storico
-- Più sofisticato: calcola il lookback in base al ritardo osservato

{{ config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge'
) }}

{% set lookback_hours = var('event_lookback_hours', 48) %}

WITH late_data_analysis AS (
    -- Calcola il ritardo medio degli ultimi 7 giorni
    -- (solo in produzione e quando la tabella esiste)
    {% if is_incremental() %}
    SELECT
        AVG(DATEDIFF('hour', event_timestamp, received_at)) AS avg_lag_hours,
        MAX(DATEDIFF('hour', event_timestamp, received_at)) AS max_lag_hours
    FROM {{ this }}
    WHERE received_at > DATEADD('day', -7, CURRENT_TIMESTAMP)
    {% else %}
    SELECT {{ lookback_hours }} AS avg_lag_hours, {{ lookback_hours }} AS max_lag_hours
    {% endif %}
),
dynamic_lookback AS (
    -- Usa il 95° percentile del ritardo storico + 20% di buffer
    SELECT GREATEST(
        {{ lookback_hours }},
        CEIL(max_lag_hours * 1.2)
    ) AS lookback_hours
    FROM late_data_analysis
)
SELECT
    e.event_id,
    e.user_id,
    e.event_type,
    e.event_timestamp,
    e.received_at
FROM {{ ref('stg_events') }} e
CROSS JOIN dynamic_lookback dl

{% if is_incremental() %}
WHERE e.event_timestamp >= DATEADD('hour', -dl.lookback_hours,
    (SELECT MAX(event_timestamp) FROM {{ this }}))
{% endif %}

Monitoring dei Modelli Incrementali

-- models/admin/incremental_health_check.sql
-- Modello di monitoraggio: traccia metriche degli incrementali in produzione

{{ config(materialized='table') }}

SELECT
    'fct_orders'                                         AS model_name,
    COUNT(*)                                             AS total_rows,
    MIN(created_at)                                      AS earliest_record,
    MAX(updated_at)                                      AS latest_record,
    DATEDIFF('hour', MAX(updated_at), CURRENT_TIMESTAMP) AS hours_since_last_update,
    COUNT(DISTINCT DATE_TRUNC('day', created_at))        AS distinct_days_covered,
    CURRENT_TIMESTAMP                                    AS check_timestamp
FROM {{ ref('fct_orders') }}

UNION ALL

SELECT
    'fct_events_hourly',
    COUNT(*),
    MIN(event_hour),
    MAX(event_hour),
    DATEDIFF('hour', MAX(event_hour), CURRENT_TIMESTAMP),
    COUNT(DISTINCT DATE_TRUNC('day', event_hour)),
    CURRENT_TIMESTAMP
FROM {{ ref('fct_events_hourly') }}

Conclusioni e Prossimi Passi

I modelli incrementali avanzati sono dove la data engineering su dbt richiede ragionamento attento su trade-off: microbatch semplifica il re-processing ma aggiunge complessità di configurazione; CDC richiede comprensione profonda del formato dei messaggi; idempotenza è non negoziabile in sistemi che gestiscono retry e correzioni dati.

Il criterio unificante: ogni modello incrementale deve rispondere alla domanda “cosa succede se lo eseguo due volte sugli stessi dati?” Se la risposta non è “lo stesso risultato”, il modello ha un bug.

Il modulo finale della serie chiude il cerchio con dbt in contesti enterprise: governance dei dati, Semantic Layer con MetricFlow e orchestrazione con Airflow.