Idempotenza nelle Pipeline di Dati: Rieseguire un Job Senza Rompere Nulla

Perché rieseguire un job è la norma, non l'eccezione

Nessuna pipeline gira una volta sola. Un job viene rieseguito perché:

• è fallito a metà (timeout, API giù, disco pieno);
• hai trovato un bug nella logica e devi rifare gli ultimi N giorni;
• sono arrivati dati in ritardo dal gestionale o dal provider;
• hai aggiunto una colonna e devi ricalcolare la storia (backfill);
• il cluster è caduto e l'orchestratore ha riprovato da solo.

Se in tutti questi casi il risultato finale è identico a quello che avresti ottenuto eseguendo il job una volta sola, il job è idempotente. Se invece ogni retry aggiunge righe, raddoppia i fatturati o sposta i timestamp, il problema non è il retry: è il design.

L'idempotenza è ciò che rende sicuro il "riprova". Senza di essa ogni automazione diventa una scommessa, e l'unica difesa resta controllare a mano che i numeri tornino.

La definizione, senza matematica pesante

In matematica una funzione è idempotente se applicarla due volte dà lo stesso risultato di applicarla una volta:

f(f(x)) = f(x)

In una pipeline: eseguire il job partendo da uno stato già elaborato deve produrre lo stesso output che si otterrebbe partendo da zero.

Due precisazioni importanti:

1. Non significa "non fare nulla se il job è già stato eseguito". Significa che anche eseguendolo, l'output non cambia.
2. È diversa da exactly once. L'idempotenza non impedisce che il job parta due volte: rende innocuo il fatto che parta due volte. È un obiettivo molto più realistico, soprattutto quando in mezzo ci sono API esterne e sistemi distribuiti.

Le tre strategie che coprono quasi tutti i casi

1. Sovrascrivere la partizione

La strategia più semplice e la più sottovalutata. Se il job produce fatti ben delimitati nel tempo (vendite di un giorno, fatturato di un mese, snapshot di magazzino), non aggiornare: sostituisci.

• In dbt, un modello incrementale può usare una strategia di delete+insert sulla partizione.
• In Spark, mode("overwrite") sulla partizione giusta.
• In BigQuery, WRITE_TRUNCATE con partizione in scrittura.

Il requisito è che la partizione sia deterministica: rieseguendo il job per il 3 marzo, devi sapere esattamente quali righe vengono toccate e quali no. Se la logica scrive "tutto quello che è arrivato dall'ultima esecuzione", non stai partizionando: stai inseguendo l'arrivo dei dati.

Il rischio principale è il dato in ritardo: se una fattura del 3 marzo arriva il 10 marzo, la partizione del 3 marzo va rifatta. Non è un difetto della strategia, è un requisito operativo: devi sapere quali partizioni ricontrollare e con che frequenza.

2. Upsert su chiave stabile

Quando l'entità evolve nel tempo (un ordine che cambia stato, un'anagrafica che si aggiorna), l'overwrite non basta: serve un upsert su chiave.

In PostgreSQL:

INSERT INTO ordini_giornalieri (data, negozio_id, fatturato)
VALUES ('2025-03-01', 42, 1250.00)
ON CONFLICT (data, negozio_id)
DO UPDATE SET fatturato = EXCLUDED.fatturato;

In un warehouse che supporta MERGE:

MERGE INTO dw.ordini AS t
USING staging.ordini AS s
  ON t.ordine_id = s.ordine_id
WHEN MATCHED THEN
  UPDATE SET t.stato = s.stato, t.totale = s.totale
WHEN NOT MATCHED THEN
  INSERT (ordine_id, stato, totale)
  VALUES (s.ordine_id, s.stato, s.totale);

Il punto critico è la chiave: deve essere stabile e unica. Un id autoincrementale generato a ogni esecuzione non è una chiave, è un contatore: cambierà a ogni run e l'upsert aggiornerà righe diverse ogni volta.

Attenzione alle cancellazioni: se una riga sparisce dalla sorgente, l'upsert non la tocca e resta nel warehouse per sempre. Serve una strategia esplicita: soft delete con flag, oppure un confronto tra set (righe presenti nella destinazione ma non più nella sorgente).

3. Chiave di idempotenza e deduplica

Per eventi, log e stream, dove le righe non "evolvono" ma si accumulano, la soluzione è una chiave di idempotenza: un identificatore che accompagna l'evento e che il database rifiuta se già visto.

CREATE TABLE eventi (
  event_id   uuid PRIMARY KEY,
  tipo       text NOT NULL,
  payload    jsonb NOT NULL,
  ricevuto_il timestamptz NOT NULL DEFAULT now()
);

INSERT INTO eventi (event_id, tipo, payload)
VALUES ('8f14e45f-ea4c-4b3e-9a4d-1c2b3a4d5e6f', 'click', '{}')
ON CONFLICT (event_id) DO NOTHING;

Se la sorgente non fornisce una chiave, si può costruire un hash del contenuto (per esempio md5 di colonne significative) oppure deduplicare a valle con una finestra:

WITH numerati AS (
  SELECT *,
         ROW_NUMBER() OVER (
           PARTITION BY ordine_id
           ORDER BY caricato_il DESC
         ) AS rn
  FROM raw.eventi
)
SELECT * FROM numerati WHERE rn = 1;

Questa seconda forma è più fragile: dipende dall'ordine di arrivo e richiede che il job venga rieseguito su tutta la finestra, non solo sull'ultimo pezzo.

Il confronto in breve

Nella pratica si usano insieme: overwrite per i fatti aggregati, upsert per le dimensioni, chiave di idempotenza per l'ingestione grezza.

Il nemico silenzioso: i contatori

Ci sono operazioni che sembrano innocue e non sono idempotenti per costruzione:

UPDATE statistiche SET visite = visite + 1 WHERE pagina = '/home';

Rieseguire il job gonfia il numero. Stesso discorso per INSERT INTO log ... senza deduplica, per le code a cui si accoda un messaggio, per i file scritti con un nome che contiene l'ora di esecuzione.

La regola generale è semplice: se il job legge lo stato attuale e lo modifica in modo incrementale, non è idempotente. Se invece calcola un risultato a partire da una sorgente e lo scrive in modo deterministico, lo è.

Altri classici:

• usare now() o la data corrente come colonna di business (data_elaborazione va bene come metadato, non come partizione logica);
• derivare la finestra di lavoro da "quando è girato il job l'ultima volta";
• dipendere dall'ordine di arrivo dei file;
• generare nomi di file casuali a ogni run.

Side effect fuori dal database

Il caso più difficile è quello in cui il job non scrive solo su un database: manda email, chiama un'API, notifica un canale, accoda un webhook.

Qui l'idempotenza sul database non basta. Le opzioni concrete sono due:

• Chiave di idempotenza trasmessa al destinatario: molte API accettano un header Idempotency-Key; se lo stesso valore viene inviato due volte, il servizio restituisce la prima risposta senza rieseguire l'operazione. Se l'API non lo supporta, si può mantenere lato nostro una tabella di "operazioni già inviate" con la chiave come primary key.
• Transactional outbox: invece di chiamare l'API dentro il job, si scrive l'intenzione di chiamarla in una tabella, nella stessa transazione che aggiorna i dati. Un secondo processo legge la tabella, invia e marca come inviato. Se il job viene rieseguito, la transazione si ripete ma la riga outbox esiste già e non viene ricreata.

Il pattern outbox ha anche un vantaggio operativo: separa il calcolo dall'integrazione. Il job dei dati non fallisce perché il servizio di notifica è lento.

Stato, watermark e backfill

Ogni pipeline incrementale ha uno stato: "fino a qui ho elaborato". Se lo stato vive fuori dal database (un file, una variabile d'ambiente, un campo sull'orchestratore), prima o poi si disallinea dai dati.

Due regole che riducono molto il problema:

1. Salva lo stato nella stessa transazione dei dati. Se aggiorni il watermark e i fatti in due passaggi separati, un crash in mezzo produce o perdita o duplicazione.
2. Deriva il lavoro dai dati quando possibile. Se puoi capire quali righe mancano interrogando la destinazione, non hai bisogno di uno stato separato: hai un job che si auto-ricostruisce.

Un backfill è solo un caso particolare di riesecuzione: la differenza è che parte da una finestra scelta a mano invece che dall'ultima esecuzione. Se il job è parametrico rispetto al tempo (accetta un data_da e un data_a), il backfill è gratis.

Come verificare che un job sia idempotente

Il test è banale e per questo viene quasi sempre saltato: esegui il job due volte di fila e confronta l'output.

import hashlib
import pandas as pd

def fingerprint(engine, query: str) -> str:
    df = pd.read_sql(query, engine)
    df = df.sort_values(by=list(df.columns)).reset_index(drop=True)
    return hashlib.sha256(df.to_csv(index=False).encode()).hexdigest()

prima = fingerprint(engine, "select * from mart_vendite order by data, negozio_id")
esegui_job()
dopo = fingerprint(engine, "select * from mart_vendite order by data, negozio_id")

assert prima == dopo, "il job non è idempotente: l'output è cambiato"

Un esempio per capire cosa aspettarsi: se la tabella contiene 100 righe dopo il primo run e 100 righe con gli stessi valori dopo il secondo, il job è idempotente. Se ne contiene 200, hai un INSERT puro; se ne contiene 100 ma con il fatturato raddoppiato, hai un UPDATE ... SET x = x + ....

Due varianti di questo test valgono la pena:

• eseguire due volte saltando la prima parte (solo la scrittura, con staging già pronto), per isolare il problema;
• eseguire su dati di input in ordine diverso, per scoprire dipendenze nascoste dall'ordinamento.

In CI, con un mini-dataset di poche righe, questo test costa secondi e intercetta la maggior parte delle regressioni. È molto più economico di scoprire il problema in produzione il giorno dopo un backfill.

Conclusione

• L'idempotenza non è una proprietà teorica: è ciò che rende un retry sicuro e un backfill possibile senza panico.
• Le tre strategie di base sono overwrite di partizione, upsert su chiave stabile e chiave di idempotenza con deduplica; nella pratica convivono nella stessa pipeline.
• Tutto ciò che incrementa un contatore o dipende dall'ora di esecuzione rompe l'idempotenza. I metadati con now() vanno bene, le colonne di business no.
• Gli effetti collaterali fuori dal database (email, API, webhook) richiedono una chiave di idempotenza concordata o il pattern transactional outbox.
• Lo stato di avanzamento va salvato nella stessa transazione dei dati, o derivato dai dati stessi.
• Il test decisivo è eseguire il job due volte e confrontare un'impronta dell'output: poche righe di codice, in CI, e molti mal di testa in meno.

Foto di copertina di Kirill Sh su Unsplash.