Qualità dei dati nella pipeline di ingestione

In un lakehouse su Microsoft Fabric, la qualità del dato si governa nel layer di trasformazione (tipicamente il passaggio Bronze → Silver), prima che i dati alimentino modelli semantici o report. Le tre patologie più frequenti sono duplicati, valori mancanti e dati in ritardo (late-arriving). Gestirle correttamente è il prerequisito per un carico incrementale idempotente: rieseguire la stessa pipeline non deve moltiplicare né corrompere le righe.

Deduplicazione con window function

Il SELECT DISTINCT non basta quando le righe sono “quasi” identiche o quando serve conservare la versione più recente. Il pattern robusto usa una window function con ROW_NUMBER() partizionata sulla business key:

WITH ranked AS (
  SELECT *, ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY updated_at DESC) AS rn
  FROM source)
SELECT * FROM ranked WHERE rn = 1
  • PARTITION BY definisce cosa rende due righe “lo stesso record”.
  • ORDER BY decide quale copia sopravvive (la più recente, quella con più campi valorizzati, ecc.).
  • RANK() va usato solo se accetti pareggi; per garantire una riga singola serve ROW_NUMBER().

Trattamento di null e valori mancanti

A livello associate la scelta non è “eliminare i null” ma decidere la strategia in base alla semantica:

  • Imputazione con COALESCE/fillna per default sensati (es. 0 per metriche additive, 'Unknown' per attributi dimensionali).
  • Filtro delle righe incomplete solo se la chiave o i campi obbligatori mancano.
  • Constraint a valle: in Delta puoi imporre NOT NULL o CHECK per far fallire il carico invece di propagare dati sporchi.

Attenzione: COUNT(colonna) ignora i null e col = NULL è sempre falso — usa IS NULL.

Dati in ritardo (late-arriving)

I record che arrivano fuori sequenza temporale non vanno trattati come nuovi inserimenti ciechi. La strategia corretta è l’upsert basato su chiave con confronto del timestamp: si aggiorna la riga esistente solo se il record entrante è più recente, altrimenti lo si scarta. Questo evita che un batch tardivo sovrascriva uno stato più aggiornato.

Upsert incrementale con MERGE su Delta Lake

L’operazione MERGE di Delta Lake è il cuore del carico incrementale idempotente. Confronta una tabella target con una source su una merge condition e applica azioni distinte per i tre rami:

MERGE INTO silver.customers AS t
USING updates AS s
  ON t.customer_id = s.customer_id
WHEN MATCHED AND s.updated_at > t.updated_at
  THEN UPDATE SET *
WHEN NOT MATCHED
  THEN INSERT *
  • WHEN MATCHED → aggiorna (o elimina) i record esistenti; la condizione extra sul timestamp gestisce i late-arriving.
  • WHEN NOT MATCHED (by target) → inserisce i nuovi record.
  • WHEN NOT MATCHED BY SOURCE → utile per chiudere o marcare come eliminate le righe non più presenti nella source.

SCD Type 1 e Type 2

MERGE implementa direttamente le slowly changing dimensions:

  • SCD Type 1: sovrascrive il valore nel ramo MATCHED (nessuno storico).
  • SCD Type 2: nel ramo MATCHED chiude la versione corrente (is_current = false, end_date) e un secondo passaggio inserisce la nuova versione, preservando lo storico.

Manutenzione della tabella

MERGE riscrive interi file Parquet e, su carichi frequenti, genera molti file piccoli che degradano le letture. La contromisura è la table maintenance periodica:

  • OPTIMIZE (con V-Order in Fabric) compatta i file e ottimizza la lettura per Power BI.
  • Z-ORDER / clustering sulle colonne di merge accelera lo skipping dei file.
  • VACUUM rimuove i file obsoleti oltre il periodo di retention.

Trappole tipiche d’esame

  • Scenario: il MERGE crea duplicati a ogni esecuzione → la merge condition non è su una chiave univoca; la partition/ON deve usare la business key corretta, altrimenti ogni riga risulta “NOT MATCHED” e viene reinserita.
  • Scenario: serve mantenere solo l’ultima versione di ogni record duplicato → usa ROW_NUMBER() partizionato sulla chiave con ORDER BY sul timestamp, non DISTINCT né RANK() (che lascia pareggi).
  • Scenario: un batch tardivo sovrascrive dati più recenti → aggiungi la condizione WHEN MATCHED AND s.updated_at > t.updated_at, non un UPDATE incondizionato.
  • Scenario: dopo settimane di micro-batch MERGE le query rallentano → pianifica OPTIMIZE + VACUUM (small files problem), non aumentare la capacity.
  • Scenario: devi conservare lo storico delle variazioni di una dimensione → SCD Type 2 con chiusura della riga corrente e insert della nuova, non un semplice UPDATE (che è Type 1 e perde lo storico).