Airflow: retries non basta, il task deve essere idempotente

La telefonata arriva sempre uguale: «il fatturato di marzo sul cruscotto è il doppio di quello vero».
Nove volte su dieci non è un bug di calcolo. È un task che è stato eseguito due volte.

Rieseguire i task, per Airflow, è normale amministrazione: non è un caso limite, è come funziona.
Il problema è che quasi nessuno scrive il task tenendone conto.

Le tre strade per cui un task gira due volte

La prima la conoscono tutti: il retry automatico. Il task fallisce, Airflow riprova.
Vale la pena sapere che di default non riprova affatto — nella
configurazione
default_task_retries vale 0 e default_task_retry_delay vale 300 secondi. Se in
default_args hai messo "retries": 3, l’hai deciso tu, ed è la scelta giusta: una rete che
sfarfalla non deve far cadere una pipeline. Ma da quel momento ogni task del DAG può girare fino a
quattro volte.

La seconda è il clear dalla UI, quello che si fa alle otto di sera quando un task è rosso.
La documentazione descrive cosa succede senza giri di parole: «Clearing a task instance creates a
record of the task instance. The try_number of the current task instance is incremented, the
max_tries set to 0 and the state set to None, which causes the task to re-run».

La terza è il backfill, ed è quella che fa i danni grossi, perché rigira lo stesso task su
decine di intervalli storici. In Airflow 3 si lancia con airflow backfill create --from-date
--to-date. Da notare che in Airflow 3 il recupero automatico degli intervalli passati è spento
di default (scheduler.catchup_by_default=False, mentre in Airflow 2 era True): il rischio non è
più la pipeline che al primo deploy parte da sola su due anni di storico, è il backfill che lanci
tu, convinto che sia un’operazione innocua.

Cosa dice Airflow, in una riga

La pagina
Best Practices è netta:
«You should treat tasks in Airflow equivalent to transactions in a database. This implies that you
should never produce incomplete results from your tasks», e «the tasks should produce the same
outcome on every re-run».

Tradotto: un task non deve aggiungere il proprio risultato. Deve essere il proprio risultato,
per l’intervallo che gli compete. Se lo rilanci dieci volte, dopo la decima la tabella deve essere
identica a come era dopo la prima.

Il codice

Questo è il task che prima o poi ti raddoppia i numeri:

from airflow.sdk import dag, task

@task
def carica_vendite():
    righe = leggi_sorgente()
    db.executemany(
        "INSERT INTO vendite (giorno, negozio, importo) VALUES (%s, %s, %s)",
        righe,
    )

Non c’è niente di sbagliato finché gira una volta sola. Alla seconda, le righe sono il doppio.

La versione che regge la rilettura è questa — cancella la propria fetta, poi la riscrive, dentro
un’unica transazione:

@task
def carica_vendite(data_interval_start=None, data_interval_end=None):
    righe = leggi_sorgente(data_interval_start, data_interval_end)
    with db.transaction():
        db.execute(
            "DELETE FROM vendite WHERE giorno >= %s AND giorno < %s",
            (data_interval_start, data_interval_end),
        )
        db.executemany(
            "INSERT INTO vendite (giorno, negozio, importo) VALUES (%s, %s, %s)",
            righe,
        )

Due cose sono cambiate, ed entrambe contano.

La finestra è esplicita. data_interval_start e data_interval_end sono le variabili di
contesto che identificano l’intervallo di dati del run — >= sull’inizio e < sulla fine, così i
confini non si sovrappongono. La documentazione lo dice come regola: «Read and write in a specific
partition. Never read the latest available data in a task». Se il task legge “gli ultimi dati
disponibili”, rieseguirlo il mese dopo produce un risultato diverso, ed è esattamente ciò che non
deve succedere.

Il DELETE e l’INSERT stanno nella stessa transazione. Se salta la corrente in mezzo, non resti
con la fetta cancellata e non riscritta. È la parte “never produce incomplete results”.

Dove il database te lo permette, l’alternativa è l’UPSERT — INSERT ... ON CONFLICT DO UPDATE su
PostgreSQL — e anche qui la documentazione è esplicita: «Do not use INSERT during a task re-run, an
INSERT statement might lead to duplicate rows in your database. Replace it with UPSERT». Serve però
una chiave naturale vera. Se non ce l’hai, il delete-then-insert sulla partizione è più onesto.

L’errore che non si vede: now()

Questo sembra innocuo e non lo è:

ieri = datetime.now().date() - timedelta(days=1)

Airflow lo dice chiaramente: «The Python datetime now() function gives the current datetime
object. This function should never be used inside a task, especially to do the critical
computation, as it leads to different outcomes on each run». Il task che usa now() calcola una
cosa diversa a ogni esecuzione: un backfill su marzo, lanciato oggi, ti riscrive marzo con i dati di
oggi. La data la dà il run, non l’orologio.

I parametri che vengono dopo, non prima

depends_on_past=True impedisce a un task di partire se la sua esecuzione precedente non è andata a
buon fine, e max_active_runs limita i run contemporanei dello stesso DAG (il default a livello di
configurazione è 16, che per una pipeline che scrive sulla stessa tabella è tanto).

Sono utili, ma sono un contenimento, non una soluzione: riducono le occasioni in cui due esecuzioni
si pestano i piedi, non rendono il task sicuro da rieseguire. Se il task è idempotente, servono di
rado. Se non lo è, non bastano — prima o poi qualcuno fa un clear.

Stesso discorso per la retry policy dichiarativa documentata nella 3.3, che permette di decidere per
tipo di eccezione se ritentare (RETRY), fallire subito (FAIL) o lasciar fare alla logica
standard (DEFAULT). È ottima per non riprovare tre volte su un errore di sintassi SQL. Non ha
niente a che vedere con la pulizia dei dati.

In pratica

Prima di mandare in produzione un DAG, una domanda sola: se lancio questo task due volte di
seguito sullo stesso intervallo, la tabella cambia? Se la risposta è sì, il task non è finito,
qualsiasi cosa dicano i retry.

È una verifica che costa cinque minuti e si fa a mano, in sviluppo. Il doppione in produzione,
invece, non lo trovi tu: lo trova il cliente, guardando un numero che non torna. E a quel punto non
stai sistemando una pipeline, stai ricostruendo la fiducia in un cruscotto.