Una guida passo-passo per ingestion incrementale, gestione dei tardi e backfill, con dbt e un orchestratore minimale.
Prerequisiti
Per seguire questa guida servono: Python 3.11+, un database PostgreSQL (o SQLite per le prove), e familiarità con la riga di comando. Tutto il codice è su GitHub, linkato in fondo.
Step 1 — Setup del progetto
Creiamo la struttura minima. L’obiettivo è tenere tutto in meno di 200 righe, senza framework esterni che non siano strettamente necessari. Ogni dipendenza aggiunta è un debito di manutenzione.
- Crea un virtualenv dedicato al progetto.
- Installa le dipendenze base: sqlalchemy, dbt-core, click.
- Configura il file profiles.yml per il tuo database.
Step 2 — Implementazione
La logica core è semplice: leggere i dati incrementalmente, trasformarli con SQL dichiarativo, e scrivere i risultati in una tabella di destinazione. Il trucco sta nella gestione degli errori e nel backfill.
@asset(io_manager_key="warehouse")
def daily_revenue(context, orders):
df = (
orders
.filter("status = 'paid'")
.groupby("day")
.agg(revenue=("amount", "sum"))
)
context.log.info(f"rows={len(df)}")
return dfNota l’assenza di stato condiviso: la funzione riceve i dati in ingresso e restituisce un dataframe. Questa è la proprietà che rende il backfill banale.
Una pipeline è buona quando puoi ri-eseguirla tre volte e ottenere lo stesso risultato. Tutto il resto è contorno.
— principio di idempotenza
Step 3 — Test e deploy
Ogni modello dbt ha un test associato. Li eseguiamo in CI ad ogni push. In produzione, un cron job lancia il tutto ogni 15 minuti. Se qualcosa fallisce, un alert Slack avvisa il team entro 60 secondi.
Il codice completo, inclusi i Dockerfile, è disponibile nel repository collegato.
