Apache Airflow: orchestrare pipeline dati ETL con Python

In sintesi. Apache Airflow non è un cron job evoluto: è un orchestratore che descrive una pipeline come un grafo di task (DAG), gestisce dipendenze, retry, backfill e monitoraggio da un’interfaccia web. In questa guida vedi i concetti chiave, l’architettura di Airflow 3, una pipeline ETL completa in Python per un caso reale di smart agriculture (sensori → InfluxDB → PostgreSQL → Grafana), la gestione degli errori con retry e trigger rules, l’installazione con Docker e quando Airflow è la scelta sbagliata.

Tutto comincia quasi sempre con un cron job. Uno script Python che ogni notte alle 2:00 scarica il CSV delle stazioni agrometeo, lo pulisce e lo carica nel database. Funziona per mesi. Poi un giorno il fornitore cambia il formato di una colonna, lo script fallisce, nessuno se ne accorge. Tre settimane dopo l’agronomo chiede perché la dashboard mostra un buco proprio nel periodo delle gelate tardive. E lì scopri che di cron job ce ne sono diciassette, su tre server diversi, e nessuno sa con certezza quale dipende da quale.

Apache Airflow nasce per risolvere esattamente questo problema. Creato in Airbnb nel 2014 e oggi progetto top-level della Apache Software Foundation, è diventato lo standard de facto per orchestrare pipeline di dati. In questa puntata della serie Stack Digitale 2026 lo usiamo per chiudere il cerchio: i dati che Kafka trasporta e che InfluxDB conserva devono essere elaborati, aggregati e sincronizzati con altri sistemi. Qualcuno deve coordinare tutto.

Cos’è Airflow (e cosa non è)

Airflow è un orchestratore: decide cosa eseguire, quando, in che ordine e cosa fare se qualcosa va storto. Non è un motore di elaborazione dati. Il lavoro pesante — leggere un CSV da un milione di righe, calcolare un’aggregazione — lo fanno Python, il database o uno strumento esterno; Airflow lancia quel lavoro, ne controlla l’esito e decide il passo successivo.

Altrettanto importante sapere cosa Airflow non è: non è uno strumento di streaming. Se hai bisogno di reagire a un evento in pochi millisecondi, la risposta è Kafka. Airflow lavora per esecuzioni discrete — ogni ora, ogni notte, quando arriva un file — ed eccelle nel batch. Le due tecnologie sono complementari, non alternative.

Airflow 3: cosa è cambiato

Se hai letto tutorial scritti prima del 2025, alcune cose sono cambiate. Airflow 3.0 (aprile 2025) è stata la release più importante dalla nascita del progetto, e nel 2026 siamo alla serie 3.3. Le novità che contano per chi inizia:

• Nuova interfaccia web in React, più veloce, con viste Grid e Graph ridisegnate.
• Versionamento dei DAG: ogni esecuzione è legata alla versione del codice che l’ha generata, e la UI mostra la storia delle modifiche.
• Task SDK: i DAG si scrivono importando da airflow.sdk, un’interfaccia stabile e separata dagli interni di Airflow.
• Asset (prima si chiamavano Dataset): un DAG può partire quando un altro ha aggiornato un dato, non solo a orario.
• Addio SLA, benvenuti Deadline Alerts (da Airflow 3.1) per essere avvisati quando un’esecuzione sfora i tempi.
• Con la 3.3 sono arrivati anche lo stato persistente dei task tra un retry e l’altro e un SDK sperimentale per scrivere task in Java o Go.

I concetti fondamentali: DAG, task e operatori

DAG: il grafo della pipeline

DAG sta per Directed Acyclic Graph, grafo diretto aciclico. Diretto perché ogni collegamento ha un verso: il task B parte dopo il task A. Aciclico perché non puoi tornare indietro: nessun task può dipendere, direttamente o indirettamente, da se stesso. In Airflow un DAG è un file Python che descrive i task e le loro dipendenze, più alcune proprietà: quando deve girare (schedule), da quando (start_date), quante volte ritentare in caso di errore.

Task e operatori

Un task è un singolo passo della pipeline. Per definirlo usi un operatore, cioè un modello pronto per un tipo di lavoro:

• PythonOperator (o il decoratore @task): esegue una funzione Python. È quello che userai l’80% delle volte.
• BashOperator: esegue un comando shell, utile per script esistenti o strumenti a riga di comando.
• SQLExecuteQueryOperator: esegue una query su un database configurato come connection (PostgreSQL, MySQL, Snowflake…).
• Sensori: task speciali che aspettano una condizione, come l’arrivo di un file in una cartella o su un bucket S3.

Esistono centinaia di operatori nei provider package ufficiali per AWS, Google Cloud, Azure, dbt, Slack e molti altri servizi.

Dipendenze

Le dipendenze si dichiarano con l’operatore >>: estrai >> valida >> carica significa “prima estrai, poi valida, poi carica”. Con la TaskFlow API, se un task usa il valore restituito da un altro, la dipendenza viene creata automaticamente.

L’architettura di Airflow 3

Capire i componenti ti aiuta a diagnosticare i problemi quando qualcosa non parte.

• DAG processor: legge i file Python nella cartella dags/ e li trasforma in una rappresentazione serializzata.
• Scheduler: il cervello. Controlla quali DAG devono partire e quali task hanno le dipendenze soddisfatte.
• Executor: decide dove girano i task — nello stesso processo (LocalExecutor), su una flotta di worker (CeleryExecutor) o in pod dedicati (KubernetesExecutor).
• Worker: i processi che eseguono concretamente il codice dei task. In Airflow 3 comunicano tramite API e non accedono più direttamente al database.
• Metadata database: tipicamente PostgreSQL. Contiene lo stato di ogni esecuzione, i log di riferimento, le connection.
• API server: serve l’interfaccia web e l’API REST.

Tutorial: la pipeline Smart Agriculture in quattro fasi

Il caso d’uso viene dal progetto P3 del corso di Advanced Data Modeling del Politecnico di Torino. Un consorzio di aziende vitivinicole del Monferrato raccoglie dati da sensori nei vigneti. Ogni notte, un file CSV con le letture del giorno viene depositato in una cartella di staging. La pipeline deve:

1. Estrarre e validare il CSV (colonne attese, valori plausibili, niente duplicati).
2. Caricare le letture in InfluxDB, con tag farm, plot, sensor.
3. Calcolare i KPI giornalieri per appezzamento: temperature minime e massime, Growing Degree Days, ore di stress idrico.
4. Scrivere i KPI in PostgreSQL, accanto allo storico delle rese e degli interventi agronomici, e lanciare un alert se c’è rischio gelata.

Ecco il DAG completo con la TaskFlow API:

from datetime import datetime, timedelta
import pandas as pd
from airflow.sdk import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook

STAGING = "/opt/airflow/data/staging"

@dag(
    schedule="@daily",
    start_date=datetime(2026, 3, 1),
    catchup=False,
    default_args={"retries": 3, "retry_delay": timedelta(minutes=10)},
    tags=["agro", "iot"],
)
def smart_agriculture_pipeline():

    @task
    def extract_and_validate(ds=None):
        df = pd.read_csv(f"{STAGING}/sensors_{ds}.csv", parse_dates=["timestamp"])
        expected = {"timestamp", "farm", "plot", "sensor", "air_temp", "soil_temp", "soil_moisture"}
        missing = expected - set(df.columns)
        if missing:
            raise ValueError(f"Colonne mancanti: {missing}")
        df = df.drop_duplicates().query("-30 < air_temp < 50")
        path = f"{STAGING}/clean_{ds}.parquet"
        df.to_parquet(path)
        return path  # passa solo il percorso, non i dati

    @task
    def load_to_influx(path):
        from influxdb_client import InfluxDBClient, WriteOptions
        df = pd.read_parquet(path).set_index("timestamp")
        with InfluxDBClient.from_env_properties() as client:
            with client.write_api(write_options=WriteOptions(batch_size=5000)) as w:
                w.write(bucket="agro", record=df, data_frame_measurement_name="sensors",
                        data_frame_tag_columns=["farm", "plot", "sensor"])
        return path

    @task
    def compute_daily_kpi(path):
        df = pd.read_parquet(path)
        kpi = df.groupby(["farm", "plot"]).agg(
            t_min=("air_temp", "min"), t_max=("air_temp", "max"),
            soil_min=("soil_temp", "min"), moisture_avg=("soil_moisture", "mean"))
        kpi["gdd"] = ((kpi.t_min + kpi.t_max) / 2 - 10).clip(lower=0)
        return kpi.reset_index().to_dict("records")

    @task
    def write_to_postgres(rows, ds=None):
        hook = PostgresHook(postgres_conn_id="agro_pg")
        hook.insert_rows("daily_kpi", [(ds, r["farm"], r["plot"], r["t_min"], r["t_max"],
                          r["gdd"], r["moisture_avg"]) for r in rows],
                         replace=True, replace_index=["day", "farm", "plot"])
        return rows

    @task.short_circuit
    def frost_risk(rows):
        return any(r["soil_min"] < 2 for r in rows)

    @task
    def send_frost_alert(rows):
        plots = [r["plot"] for r in rows if r["soil_min"] < 2]
        print(f"ALLERTA GELATA negli appezzamenti: {plots}")  # qui: email, Slack, Telegram...

    path = extract_and_validate()
    kpi = compute_daily_kpi(load_to_influx(path))
    saved = write_to_postgres(kpi)
    frost_risk(saved) >> send_frost_alert(saved)

smart_agriculture_pipeline()

Tre dettagli che fanno la differenza

• Si passano riferimenti, non dati. Tra un task e l’altro i valori viaggiano tramite XCom, salvati nel database dei metadati. Per questo il primo task restituisce il percorso del file Parquet, non il DataFrame. Passare un milione di righe via XCom è uno degli errori più comuni.
• La data logica ds. Ogni esecuzione sa a quale giorno si riferisce. Se rilanci l’esecuzione del 15 marzo, il DAG legge il file del 15 marzo, non quello di oggi.
• Scritture idempotenti. L’insert_rows con replace=True sostituisce le righe dello stesso giorno invece di duplicarle. Puoi rieseguire il DAG dieci volte e il risultato non cambia.

Scheduling, catchup e backfill

Il parametro schedule accetta preset (@hourly, @daily, @weekly), espressioni cron ("0 2 * * *") o asset: “parti quando il DAG di import ha aggiornato la tabella X”.

Il concetto più importante è l’intervallo di dati. Un DAG giornaliero che gira nella notte tra il 15 e il 16 marzo elabora i dati del 15: la sua data logica è il 15. Se imposti catchup=True e una start_date di tre mesi fa, Airflow crea automaticamente tutte le esecuzioni mancanti. Il backfill è la stessa cosa fatta a comando: “ricalcola tutto marzo con la nuova formula dei GDD”. È qui che l’idempotenza smette di essere una buona pratica e diventa un requisito.

Gestire gli errori: retry, trigger rules e alert

In una pipeline reale qualcosa fallisce sempre: la rete, un’API esterna lenta, un file arrivato in ritardo. Airflow ti dà tre livelli di difesa.

Retry automatici

Con retries=3 e retry_delay=10 minuti, un errore temporaneo si risolve da solo. Puoi attivare anche retry_exponential_backoff per allungare progressivamente l’attesa, utile con API che applicano rate limit.

Trigger rules

Di default un task parte solo se tutti i task a monte sono riusciti (all_success). Ma a volte vuoi comportamenti diversi:

Callback e Deadline Alerts

Con on_failure_callback puoi eseguire una funzione quando un task fallisce definitivamente: tipicamente un messaggio su Slack o Teams con il link ai log. Con i Deadline Alerts di Airflow 3 vieni avvisato quando un’esecuzione non termina entro un tempo stabilito: se il report per l’agronomo deve essere pronto alle 7:00, vuoi saperlo alle 6:30 che è in ritardo, non alle 8:00.

Monitorare dalla UI

La differenza più evidente rispetto ai cron job è la visibilità. Dall’interfaccia web vedi ogni esecuzione di ogni DAG in una griglia colorata: verde riuscito, rosso fallito, arancione in retry. Un clic su un quadratino rosso ti porta al log di quel task in quella data. Puoi rilanciare un singolo task, marcarlo come riuscito o avviare un backfill senza toccare il terminale.

Installazione con Docker

Il modo più rapido per provare Airflow in locale è il file docker-compose.yaml ufficiale, disponibile nella documentazione:

mkdir airflow-lab && cd airflow-lab
curl -LfO "https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml"
mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env
docker compose up airflow-init
docker compose up -d
# interfaccia su http://localhost:8080 (utente e password: airflow)

Attenzione alle risorse: servono almeno 4 GB di RAM dedicati a Docker, meglio 8. È un problema concreto sui portatili degli studenti, e nelle note del corso suggeriamo alternative più leggere quando serve: Prefect (con un piano cloud gratuito) o, per casi semplicissimi, APScheduler dentro un’applicazione Python. Per un singolo computer c’è anche airflow standalone, che avvia tutto in un processo con SQLite: comodo per imparare, mai per la produzione.

Un secondo caso: la puntualità del trasporto pubblico

Il progetto P5 del corso mostra un altro schema tipico. I dati GTFS con gli orari programmati di GTT vivono in PostgreSQL; le posizioni e i ritardi dei mezzi, simulati con Kafka, finiscono in InfluxDB. Ogni notte un DAG calcola l’On-Time Performance per linea: extract_from_influxdb >> compute_otp >> write_to_pg >> refresh_view. Il punto interessante è il secondo task: il JOIN tra ritardi effettivi (InfluxDB) e orari programmati (PostgreSQL) non può farlo nessuno dei due database da solo. Lo fa Python dentro l’operatore, e Airflow garantisce che avvenga solo quando entrambe le sorgenti sono pronte. È l’orchestrazione multi-sistema che un task interno di InfluxDB non può offrire.

Airflow, cron o n8n? Scegliere lo strumento giusto

Se il tuo problema è “quando arriva una mail con allegato, salvalo su Drive e avvisa il commerciale”, Airflow è sproporzionato: basta un tool come quelli che abbiamo confrontato nell’articolo n8n vs Make.com. Se invece hai più database da tenere allineati, dati storici da ricalcolare e un bisogno reale di sapere cosa è successo a ogni esecuzione, Airflow ripaga ampiamente la curva di apprendimento.

Best practice in sintesi

• Task piccoli e atomici: un task, una responsabilità. Più facili da rilanciare e da debuggare.
• Idempotenza sempre: rieseguire un task per la stessa data deve dare lo stesso risultato.
• Niente dati pesanti in XCom: passa percorsi, chiavi, identificativi.
• Niente logica pesante al livello superiore del file: il DAG processor lo rilegge di continuo, quindi query e chiamate API vanno dentro i task.
• Credenziali nelle Connection di Airflow o in un secret manager, mai nel codice.
• Usa la data logica, mai datetime.now(), per decidere quali dati elaborare.

Il quadro si completa

Con Airflow lo stack della serie è completo: Forms e Sheets per raccogliere, Kafka per trasportare in tempo reale, InfluxDB per le serie temporali, Neo4j per le relazioni, Grafana per mostrare, Airflow per coordinare. Nel prossimo articolo della serie facciamo un passo indietro e guardiamo il disegno complessivo: perché le aziende più avanzate usano quattro o cinque database diversi, e come fanno a tenerli coerenti.

Hai script notturni che nessuno osa toccare, o dati che dovrebbero essere aggiornati ogni giorno e invece lo sono “quando qualcuno se ne ricorda”? Richiedi un audit dati e AI: mappiamo le tue pipeline attuali e capiamo insieme se serve Airflow o se basta mettere ordine.