import pandas as pd

Pipeline orchestration: Airflow and Prefect

Quando le pipeline diventano molte, l’ordine di esecuzione non può più vivere nelle chat, in cron sparsi o nella memoria del team. Airflow e Prefect servono a rendere espliciti dipendenze, retry, scheduling, parametri e responsabilità. Questa lezione affronta il passaggio da uno script che funziona a un workflow che si può gestire, e va letta come design di dipendenze: cosa deve partire prima, cosa può fallire senza bloccare tutto, cosa va ritentato e cosa richiede un intervento umano.

Cosa risolve davvero un orchestratore

Il problema non è conoscere Airflow o Prefect in astratto. Il problema è decidere quale strumento adottare quando hai dieci script schedulati a orari diversi e nessuno sa quale dipende da quale. Un orchestratore non rende affidabile una pipeline fragile, ma rende visibili le sue fragilità, e questa visibilità è già metà del lavoro operativo.

La scelta tra Airflow, Prefect o un servizio gestito parte da fattori concreti: il numero e la complessità delle dipendenze, il volume dei workflow, la necessità di backfill, il livello di osservabilità richiesto e le competenze disponibili nel team. Inquadrare bene questi vincoli prima di scrivere codice evita di trovarsi legati a uno strumento che combatte contro il modo in cui lavori.

Una mappa per la decisione

Conviene seguire una sequenza che lega la scelta tecnica a un esito misurabile, invece di adottare uno strumento perché è di moda.

Step	Question to ask	Expected output
Decision	Che cosa cambia se orchestriamo meglio le pipeline?	Scelta esplicita
Signal	Quale dato osservabile riduce l’incertezza?	Metrica o evento
Baseline	Rispetto a cosa interpretiamo il risultato?	Credible comparison
Vincolo	Che cosa può falsare la lettura?	Assunzione da dichiarare
Action	Quale passo operativo segue?	Raccomandazione controllabile

Nel contesto dell’infrastructure ops, la prima domanda non è quale metrica calcolare, ma quale decisione dovrà essere presa grazie all’analisi. Una dashboard o un grafico di costi hanno valore solo se riducono incertezza decisionale. Se non cambiano una scelta, sono documentazione o teatro analitico.

Rendere visibili le assunzioni

Per rendere analizzabile la scelta, definisci prima l’unità di lavoro, che qui è il job, la pipeline, il servizio, l’alert, il deploy o la risorsa cloud. Poi collega quell’unità a una metrica osservabile come SLA, MTTR, costo, failure rate, freshness e qualità dati. Infine dichiara la decisione attesa, che può essere un runbook, un alert, una policy di costo, una pipeline CI/CD o un piano di recovery.

Element	Operational Definition	Controllo minimo
Unit of analysis	Job, pipeline, servizio, alert, deploy o risorsa cloud	Definizione stabile e tracciabile
Variabile osservata	SLA, MTTR, costo, failure rate, freshness	Definizione condivisa con il team
Baseline	Periodo precedente, benchmark o scenario controfattuale	Credible comparison
Soglia decisionale	Punto in cui scatta un alert o un intervento	Criterio scritto prima dell’incidente
Rischio residuo	Errore che resta anche dopo l’orchestrazione	Sensitivity check o revisione

La formalizzazione è solida quando un altro ingegnere può riprodurre la logica, criticare le assunzioni e arrivare alla stessa decisione partendo dagli stessi dati.

Airflow: the veteran

Airflow (Airbnb, 2014) is the de facto standard. It models workflows as DAGs (Directed Acyclic Graphs) in Python:

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG('etl_daily', start_date=datetime(2024,1,1), schedule='@daily') as dag:
    extract = BashOperator(task_id='extract', bash_command='python extract.py')
    transform = BashOperator(task_id='transform', bash_command='dbt run --select staging')
    validate = BashOperator(task_id='validate', bash_command='dbt test')
    load = BashOperator(task_id='load', bash_command='dbt run --select marts')
    extract >> transform >> validate >> load

I pregi sono la maturità, una community enorme e operatori per ogni sistema immaginabile. I difetti sono la complessità di gestione, uno scheduling statico e un backfill spesso problematico.

Prefect e Dagster: la nuova generazione

Prefect e Dagster, arrivati dal 2020 in poi, affrontano i limiti di Airflow con un approccio più moderno. Lo scheduling è dinamico, perché i parametri possono cambiare a runtime e non solo a definizione. Il retry è nativo con backoff esponenziale, senza configurazioni esterne. C’è caching dei task, quindi se un task è già stato eseguito con gli stessi input viene saltato. E al posto degli operatori in stile YAML si lavora con primitive Python, cioè normali funzioni con decoratori.

from prefect import flow, task

@task(retries=3, retry_delay_seconds=60)
def extract():
    return fetch_from_api()

@flow
def etl_pipeline():
    data = extract()
    transform(data)

Pattern di orchestrazione ricorrenti

Alcuni schemi tornano in quasi tutti i progetti. Il fan-out/fan-in fa generare a un task N task paralleli e poi li riunisce, ed è il pattern tipico per partizionare i dati per data e processarli in parallelo. Il conditional branching rende il DAG non lineare, come in if data_is_valid: run_etl() else: send_alert(). Il backfill serve a riprocessare i dati storici: in Airflow cambi la start_date e ci pensa catchup, mentre in Prefect è un flow separato. I sensor, infine, attendono che una condizione esterna sia vera, per esempio che un file su S3 esista, prima di procedere.

Netflix e la disciplina delle metriche

Netflix è un esempio utile perché ha costruito molte decisioni di prodotto intorno a segnali comportamentali osservabili: completamento degli episodi, tempo di ricerca prima della riproduzione, abbandono dopo pochi minuti, ritorno nella settimana successiva, efficacia delle raccomandazioni. Il punto non è che ogni azienda debba copiare Netflix, ma che il dato non viene trattato come ornamento, viene trattato come infrastruttura decisionale.

Quando Netflix valuta una modifica all’esperienza, per esempio una nuova riga di raccomandazioni, una diversa immagine di copertina o un algoritmo di ranking, non misura solo il click immediato. Guarda anche segnali di qualità, cioè se l’utente guarda davvero il contenuto, se torna nei giorni successivi e se riduce il tempo speso a cercare. Questa disciplina impedisce di ottimizzare vanity metric che sembrano positive nel breve periodo ma erodono valore nel lungo. Lo stesso vale per l’orchestrazione: se la metrica che osservi non aiuta a scegliere tra due azioni alternative, stai misurando rumore.

SQL example: building a control view

Il pattern seguente è volutamente generico ma eseguibile nella maggior parte dei warehouse moderni. L’obiettivo è creare una base analitica con metrica, segmento e finestra temporale, così da confrontare periodi e gruppi senza riscrivere la logica ogni volta.

WITH base_events AS (
  SELECT
    user_id,
    account_id,
    event_type,
    event_time,
    DATE_TRUNC('week', event_time) AS week,
    source,
    device_type
  FROM events
  WHERE event_time >= CURRENT_DATE - INTERVAL '180 days'
    AND user_id IS NOT NULL
),
weekly_user_metrics AS (
  SELECT
    week,
    user_id,
    COALESCE(source, 'unknown') AS source,
    COALESCE(device_type, 'unknown') AS device_type,
    COUNT(*) AS total_events,
    COUNT(DISTINCT DATE(event_time)) AS active_days,
    COUNT(DISTINCT event_type) AS event_diversity,
    MAX(CASE WHEN event_type IN ('purchase', 'subscribe', 'activation') THEN 1 ELSE 0 END) AS reached_key_outcome
  FROM base_events
  GROUP BY week, user_id, source, device_type
)
SELECT
  week,
  source,
  device_type,
  COUNT(DISTINCT user_id) AS users,
  ROUND(AVG(active_days), 2) AS avg_active_days,
  ROUND(AVG(event_diversity), 2) AS avg_event_diversity,
  ROUND(AVG(reached_key_outcome) * 100, 2) AS key_outcome_rate
FROM weekly_user_metrics
GROUP BY week, source, device_type
ORDER BY week, source, device_type;

La query non pretende di essere la risposta finale. Serve a creare una superficie di osservazione fatta di trend, segmenti, differenze tra canali e variazioni nel tempo. Per un orchestratore lo stesso schema vale sulle run: confrontare durata, failure rate e costo tra pipeline diverse aiuta a capire quale workflow merita attenzione.

Python example: checking stability and anomalies

Una metrica operativa utile deve essere stabile abbastanza da orientare le decisioni e sensibile abbastanza da segnalare i cambiamenti reali. In Python puoi controllare le variazioni anomale settimana su settimana, per esempio su failure rate o durata media delle run.

# df contiene: week, segment, users, key_outcome_rate
# key_outcome_rate espresso in percentuale, es. 12.4

df = df.sort_values(['segment', 'week']).copy()
df['previous_rate'] = df.groupby('segment')['key_outcome_rate'].shift(1)
df['wow_change_pp'] = df['key_outcome_rate'] - df['previous_rate']
df['rolling_mean'] = df.groupby('segment')['key_outcome_rate'].transform(
    lambda s: s.rolling(4, min_periods=2).mean()
)
df['rolling_std'] = df.groupby('segment')['key_outcome_rate'].transform(
    lambda s: s.rolling(4, min_periods=2).std()
)
df['z_score'] = (df['key_outcome_rate'] - df['rolling_mean']) / df['rolling_std']

anomalies = df[df['z_score'].abs() >= 2].sort_values('z_score')
print(anomalies[['week', 'segment', 'key_outcome_rate', 'wow_change_pp', 'z_score']])

Il valore del controllo è pratico: eviti di reagire a ogni oscillazione casuale, ma vedi subito quando una variazione merita un’indagine. In azienda questo alimenta gli alert, le review settimanali e le retrospettive operative.

Errori tipici e come evitarli

Il primo errore è ragionare su dati aggregati troppo presto, perché una media globale nasconde due segmenti che si muovono in direzioni opposte. Il secondo è non controllare la qualità del dato, dato che eventi duplicati, tracking incompleto, timezone incoerenti e cambi di definizione producono conclusioni false. Il terzo è confondere correlazione e causalità: se le pipeline che usano un certo orchestratore falliscono di meno, non è detto che sia merito dello strumento, potrebbero gestire workload più semplici.

C’è anche un errore più sottile, che è usare l’orchestrazione come etichetta tecnica invece che come criterio di scelta. Succede quando il team presenta un numero senza dire quale decisione cambia, quale baseline lo rende interpretabile e quale rischio resta aperto. Per ridurre questi rischi, ogni analisi dovrebbe contenere la definizione esplicita della metrica, il confronto per segmento e la verifica contro un periodo precedente o un gruppo di controllo.

Esercizio applicato

Al livello base scrivi in cinque righe quale decisione concreta dovrebbe migliorare grazie all’orchestrazione, indicando metrica, unità di analisi, baseline e rischio principale. Se non riesci a nominare la decisione, il problema è ancora troppo astratto.

Al livello intermedio costruisci una tabella con quattro colonne: segnale osservato, interpretazione prudente, controllo necessario e azione consigliata. Inserisci almeno un caso in cui il segnale da solo non basta per decidere.

Al livello research-grade trasforma l’esercizio in un memo decisionale con assunzioni, criteri di esclusione, soglia di intervento, sensitivity check e proposta di monitoraggio dopo la decisione. Come materiale usa log delle pipeline, metriche dell’orchestratore, cost explorer, dati CI/CD, incident report e dashboard ops. Se non hai dati reali, crea un dataset sintetico con almeno 200 righe, una dimensione temporale, una dimensione di segmento e una metrica di outcome.

Checkpoint

Prima di chiudere, verifica di saper rispondere a queste cinque domande.

1. Quale decisione concreta dovrebbe migliorare grazie a questa lezione?
2. Which unit of analysis makes the problem measurable?
3. Quale baseline useresti per evitare una lettura isolata?
4. Quale assunzione, se falsa, cambierebbe la conclusione?
5. Quale controllo presenteresti prima di mettere in produzione?

Summary

Orchestrare le pipeline è uno strumento decisionale, non un esercizio di stile tecnico. Airflow resta la scelta matura quando servono operatori e community, mentre Prefect e Dagster convengono dove contano scheduling dinamico, retry nativo e caching. Il valore nasce quando colleghi problema, dati, metrica, segmentazione e azione. Una buona analisi non termina con “il job è girato”, ma con una frase operativa: quale decisione prendiamo, con quale livello di confidenza e quale metrica useremo per sapere se avevamo ragione.