Apache Airflow Spiegato: Una Guida Per Principianti

Introduzione

Le aziende moderne basate sui dati fanno tutte affidamento su pipeline di dati che recuperano, trasformano, arricchiscono e spostano informazioni da un luogo all'altro. Queste pipeline di dati possono coinvolgere molti passaggi (estrazione dei dati grezzi, pulizia, addestramento di modelli di machine learning, creazione di dashboard, ecc.). Inoltre, devono essere eseguite in un ordine molto specifico. Strumenti di orchestrazione dei flussi di lavoro come Apache Airflow aiutano a garantire che tutti questi passaggi vengano eseguiti al momento giusto e nell'ordine corretto. Rendono anche le pipeline facili da monitorare e gestire.

Airflow è stato creato da Airbnb nel 2014 ed è diventato una delle piattaforme open-source di orchestrazione dei flussi di lavoro più popolari. Viene utilizzato da migliaia di organizzazioni per automatizzare e monitorare pipeline batch. In questo articolo, spiegheremo Apache Airflow dalle basi. Gli argomenti che tratteremo includono:

• Cos'è Airflow, cosa fa e perché i team di dati lo usano.
• I concetti e i componenti principali di Airflow e come si integrano tra loro.
• Come le pipeline sono definite come Grafi Aciclici Diretti (DAG) e una panoramica di un semplice DAG in azione.
• Casi d'uso comuni come ETL/ELT, MLOps e altri.

Speriamo che questa sia un'introduzione pratica e basata su esempi ad Apache Airflow per ingegneri, studenti e team che stanno iniziando con l'orchestrazione dei flussi di lavoro.

Punti Chiave

• Apache Airflow è una piattaforma di orchestrazione dei flussi di lavoro basata su Python che gestisce, pianifica e monitora pipeline di dati complesse utilizzando DAG (Grafici Aciclici Diretti).
• I DAG definiscono i compiti e le loro dipendenze, permettendo ad Airflow di eseguire ogni passaggio nell'ordine corretto con retry integrati, logging e osservabilità.
• L'architettura di Airflow (Scheduler, Executor, Worker, Metadata DB, Web UI) consente un'esecuzione distribuita e scalabile e il monitoraggio in tempo reale dei pipeline.
• Airflow eccelle nei flussi di lavoro batch come lavori ETL/ELT, pipeline di addestramento ML, controlli di qualità dei dati e analisi pianificate — ma non è ideale per carichi di lavoro in tempo reale o ad altissima frequenza.
• Alternative come Prefect, Dagster, Luigi, Argo Workflows e Mage AI potrebbero essere più adatte a seconda delle esigenze, come i flussi di lavoro basati su eventi, l'esecuzione nativa su Kubernetes o interfacce moderne più user-friendly per gli sviluppatori.

Cos'è Apache Airflow?

Apache Airflow è un sistema open-source per la creazione, la pianificazione e il monitoraggio dei workflow. Un workflow di Airflow è un DAG (un grafo aciclico diretto), che è una serie di attività con relazioni chiaramente definite (dipendenze) e senza cicli (un'attività non può ripetersi a monte). La filosofia principale di Airflow è "configurazione come codice", quindi invece di utilizzare un'interfaccia utente con drag-and-drop, si scrivono script Python per rappresentare i workflow. Questo approccio offre una flessibilità significativa agli ingegneri: è possibile connettersi quasi a qualsiasi tecnologia tramite gli operatori di Airflow e le librerie Python. Inoltre, è possibile applicare le migliori pratiche di ingegneria del software (controllo delle versioni, testing, ecc.) alle definizioni delle pipeline.

Perché l'orchestrazione del flusso di lavoro è importante?

Il punto chiave è che spesso in qualsiasi flusso di lavoro dei dati ci sono più passaggi che devono essere eseguiti in sequenza. Un esempio semplice è la pipeline ETL (Extract/Transform/Load). Le pipeline possono diventare rapidamente più complesse man mano che aggiungi più attività, dipendenze, ramificazioni e così via, e devono anche essere eseguite con una cadenza regolare. In questi casi, l'intervento manuale o gli strumenti di pianificazione semplici spesso diventano insufficienti. Airflow risolve questo problema agendo come orchestratore centrale. Ha un grafo aciclico diretto (DAG) delle attività (nodi) con dipendenze (archi). Sa quando eseguire quale attività (aspettando che le dipendenze upstream siano completate prima di eseguire), inclusa la pianificazione (esecuzione periodica), la gestione degli errori (retry automatici, avvisi) e il logging.

Se consideriamo l'intero flusso di dati come una ricetta, allora ogni singolo compito è un sottopasso (tagliare, bollire, impiattare, ecc.). Airflow è come il capo cucina che conosce le ricette e tiene d'occhio tutti i diversi sottopassi che avvengono in ogni cucina, assicurandosi che tutto accada nell'ordine corretto e al momento giusto.

Senza l'orchestrazione dei flussi di lavoro, potresti avere tutti gli chef e gli ingredienti, ma non sono coordinati. Airflow fornisce quella coordinazione, permettendo la produzione del "piatto" desiderato (esecuzione del flusso di lavoro).

Perché i team di dati usano Airflow?

Airflow è diventato popolare tra gli ingegneri dei dati, gli ingegneri ML e i team DevOps perché risponde a molte esigenze dei pipeline di dati complessi. Ecco alcuni dei principali motivi per cui i team scelgono Airflow:

• Orchestrare flussi di lavoro complessi e dipendenze: Airflow è stato progettato per flussi di lavoro con un gran numero di passaggi interdipendenti. Airflow rende semplice dichiarare le dipendenze a monte e a valle, garantendo che i compiti vengano eseguiti nell'ordine corretto. Se il compito C dipende dai compiti A e B, il scheduler di Airflow è consapevole di queste relazioni e garantirà che A e B vengano eseguiti (eventualmente in parallelo) prima di eseguire C. Airflow può gestire dipendenze complesse, logica ramificata, ecc.
• Pianificazione Centralizzata (Cron Potenziato): Fornisce un pianificatore che consente di eseguire attività secondo un determinato programma. È possibile definire decine o centinaia di pipeline che attivano attività in momenti diversi (ogni ora, giornalmente, mensilmente) utilizzando il pianificatore di Airflow. Airflow comprende le relazioni tra le attività. Può pianificare e attivare le attività in modo intelligente. Il pianificatore centrale ti libera dall'aver bisogno di scrivere diversi script su server diversi.

• Scalabilità ed Esecuzione Distribuita: Airflow può scalare anche con i tuoi flussi di lavoro. Può essere configurato per eseguire attività in parallelo su più processi worker o macchine. Inoltre, offre diversi executor (discussi più avanti in questo articolo) che possono eseguire attività su sistemi distribuiti come Celery o Kubernetes attraverso un cluster di nodi. Questo significa che puoi parallelizzare compiti pesanti (elaborazione di big data) e avere più pipeline in esecuzione contemporaneamente.
• Monitoraggio, Registrazione e Avvisi: Viene fornito con un'interfaccia web ricca che offre visibilità in tempo reale sui tuoi flussi di lavoro. Puoi vedere quali attività hanno avuto successo e quali sono fallite, e visualizzare l'output dei log per ciascuna attività. Puoi persino attivare manualmente e/o riprovare le attività tramite l'interfaccia web.

Componenti chiave del flusso d'aria

Di seguito è riportata una breve panoramica dei principali elementi costitutivi di Apache Airflow. Usa questa tabella come riferimento rapido per capire cosa è ciascun componente e cosa fa.

Come Funziona Airflow (Architettura)

Ecco una panoramica passo-passo su come Airflow elabora un flusso di lavoro (DAG):

• Creazione del DAG: L'autore del DAG scrive uno script Python che definisce un DAG. Questo script include elementi come la pianificazione (ad esempio, “esegui quotidianamente alle 9:00”) o le impostazioni per il trigger manuale. Contiene anche le definizioni dei task (tramite operator o funzioni @task), così come le dipendenze tra i task (ad esempio, task1 >> task2 (task2 aspetta che task1 finisca)). Il file Python viene salvato nella cartella DAG di Airflow affinché possa essere rilevato da Airflow. Airflow leggerà (analizzerà) questo file periodicamente per caricare la definizione del DAG.
• Pianificazione e Parsing dei DAG: Airflow Scheduler è un processo che continua a girare in background. Controlla se ciascun DAG deve essere eseguito in base al suo programma o ai trigger. Ad esempio, un DAG con una pianificazione “giornaliera” può essere rilevato dal scheduler, che nota che una nuova esecuzione deve essere programmata a mezzanotte. Airflow Scheduler verifica anche se i task all'interno dei DAG sono pronti per essere eseguiti – il che significa che tutte le loro dipendenze upstream sono completate e il loro orario programmato (data logica) è arrivato.
• Esecuzione dei Compiti: Ogni compito da eseguire viene consegnato all'Esecutore dallo Scheduler. L'esecutore decide dove e come eseguire il compito. Può essere eseguito localmente o inviato a una coda di lavoratori. Per un'architettura distribuita, il compito verrà aggiunto a una coda e poi selezionato dai processi Worker che eseguono i compiti.
• Aggiornamento dello Stato e Gestione degli Errori: Ogni istanza di attività riporta il proprio stato (successo, fallimento, ecc.) al scheduler man mano che viene completata, e il scheduler aggiorna il suo database dei metadati con lo stato dell'attività.
  Il monitoraggio dello stato consente al scheduler di sapere quale azione intraprendere successivamente: ad esempio, se il Task A è stato completato con successo, sa che può pianificare il Task B, oppure se il Task A è fallito con tentativi ancora disponibili, pianifica un nuovo tentativo, o se è fallito senza tentativi rimanenti, contrassegna l'esecuzione del DAG come fallita. Se un'attività fallisce, Airflow dispone di una logica per ritentare automaticamente l'attività se configurata per i tentativi. Se continua a fallire dopo i tentativi, Airflow può inviare notifiche sui fallimenti (email o messaggio su Slack, ecc.). Tutto questo è configurabile come parte dei parametri del DAG o dell'attività (ad esempio, numero di tentativi, ritardo tra i tentativi, email da inviare in caso di allerta).
• Monitoraggio e Intervento: L'interfaccia web di Airflow fornisce un'interfaccia live del processo sopra descritto. Puoi vedere apparire l'esecuzione di un DAG (ad esempio, una nuova riga nella visualizzazione ad albero per un'esecuzione odierna) e osservare i cambiamenti di stato dei task man mano che procedono. La visualizzazione a grafo rappresenta i task del DAG e mostra quali sono stati completati o hanno generato errori.
• Pulizia e Prossima Esecuzione: Quando un'esecuzione di un DAG è completata (tutte le attività hanno avuto successo o sono state saltate/fallite), Airflow passa alla successiva esecuzione programmata. Tutti i metadati delle esecuzioni passate vengono comunque conservati nel database, e puoi sempre controllare la cronologia. Lo scheduler continuerà a attivare nuove esecuzioni secondo la pianificazione o tramite attivazione manuale. Questo processo viene ripetuto fino a quando il processo di Airflow non viene terminato.

Comprendere i DAG in Airflow

Un DAG in Airflow definisce il programma, le attività e le dipendenze necessarie per eseguire un flusso di lavoro. Un DAG non deve sapere cosa c'è all'interno di ogni attività; deve solo definire quando e in quale ordine devono essere eseguite.

Compiti, operatori e sensori

Le attività sono le unità di lavoro di base in Airflow. In Airflow, le attività sono istanze di operatori, che sono modelli per azioni predefinite. Airflow offre una varietà di operatori principali. Il BashOperator ti permette di eseguire comandi shell, e il PythonOperator ti consente di chiamare funzioni Python. Airflow fornisce anche un decoratore @task, che ti consente di trasformare una funzione Python normale in un'attività.

Oltre agli operatori, Airflow fornisce anche sensori. I sensori sono attività che aspettano che una condizione o un evento si verifichi prima di avere successo. Possono essere utilizzati per cose come aspettare che un file compaia o che una tabella venga popolata.

Dipendenze dei compiti

Per dichiarare le dipendenze tra le attività, si utilizzano gli operatori >> e << (consigliati) o metodi come set_upstream e set_downstream. È anche possibile concatenare le attività, creare dipendenze incrociate a valle e persino costruire dinamicamente elenchi di attività. Un DAG senza alcuna dipendenza sarebbe solo un insieme indipendente di attività.

Le migliori pratiche per la progettazione dei DAG

La tabella sottostante presenta alcune linee guida da tenere a mente quando si progettano DAG.

Esempio: Un DAG Airflow Semplice

Nulla solidifica meglio i concetti di un esempio. Passiamo attraverso un semplice codice DAG di Airflow e spieghiamo le sue parti. Questo esempio creerà un piccolo flusso di lavoro con due attività in sequenza:

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

# Define the DAG and its schedule
with DAG(
    dag_id="example_dag",
    description="A simple example DAG",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",  # runs daily
    catchup=False
) as dag:
    # Task 1: Print current date
    task1 = BashOperator(
        task_id="print_date",
        bash_command="date"
    )
    # Task 2: Echo a message
    task2 = BashOperator(
        task_id="echo_hello",
        bash_command="echo 'Hello, Airflow!'"
    )
    # Define dependency: task1 must run before task2
    task1 >> task2

Analizziamo cosa sta succedendo qui:

• importare le classi necessarie DAG (per definire il DAG) e BashOperator (per eseguire comandi shell/bash nei task). Importiamo anche datetime per specificare la data di inizio del DAG.
• Creiamo un oggetto DAG con un gestore di contesto (with DAG(…) as dag:). Forniamo al DAG un ID unico “example_dag” (deve essere unico in un ambiente Airflow), una descrizione (opzionale, per aiutare gli utenti a comprendere), una start_date e un schedule_interval. Qui, impostiamo un schedule_interval preset (@daily) integrato e start_date=datetime(2025,1,1). Abbiamo anche impostato catchup=False per evitare il backfilling di eventuali esecuzioni mancanti tra la start_date e oggi (vogliamo solo che venga eseguito da ora in poi quotidianamente). Questo intero blocco crea un DAG di cui Airflow sarà a conoscenza per l'esecuzione giornaliera all'inizio della giornata.
• Definiamo quindi due attività nel contesto del DAG:
• task1 è definito utilizzando BashOperator per eseguire un comando bash (date). task_id è “print_date” e verrà utilizzato per identificare questo task nell'interfaccia utente e nei log. Quando viene eseguito, questo task stamperà semplicemente la data/ora corrente (nei log del task).
• task2 anche utilizza un BashOperator con task_id=“echo_hello”, e command=echo ‘Hello, Airflow!’. Questo stamperà un bel messaggio. Questi task al momento sono solo definiti, non ancora collegati.
• L'ultima riga (task1 >> task2) specifica una dipendenza (flusso): task1 fluisce in task2. Questo significa che il task 1 verrà eseguito e completato con successo prima che il task 2 possa essere considerato per la pianificazione dell'esecuzione. Graficamente, nel grafo DAG, ci sarà una freccia che punta da task1 a task2.

Alcune cose da notare in questo esempio:

• Abbiamo usato solo BashOperator per semplicità, ma avremmo potuto usare un DAG con tipi di operatori diversi. Potremmo avere un PythonOperator per chiamare una funzione Python, o un SimpleHttpOperator per fare una richiesta a un'API, e così via.
• Il modello con 'DAG(…) as dag' è un idioma comune per creare un'istanza di DAG. Tutti i task che vengono creati al suo interno sono automaticamente associati a quel DAG.
• catchup=False è un parametro che viene utilizzato in molti DAG con una start_date nel passato, quando non vuoi che venga eseguito per tutte le date passate. Se avessimo lasciato catchup=True (il valore predefinito) e oggi fosse il 10 gennaio 2025, Airflow proverebbe a eseguire il DAG per il 1, 2, 3 gennaio… fino al 10 gennaio, recuperando quei giorni persi arretrati. Qui lo abbiamo disattivato per semplicità.
• Questo esempio utilizza il percorso di importazione relativo per BashOperator. In Airflow 2, molti operatori sono stati spostati sotto airflow.providers. Questo esempio è scritto assumendo che quei provider siano installati (BashOperator si trova nel cuore di Airflow stesso).

Questo semplice esempio può essere facilmente ampliato: potresti aggiungere attività aggiuntive (magari fare una chiamata API, poi caricare i risultati su un database, quindi inviare una notifica). Puoi anche aggiungere logiche più complesse (come ramificazioni: Airflow supporta le ramificazioni in stile if/else utilizzando, ad esempio, il BranchOperator).

Man mano che i tuoi DAG diventano più grandi, l'interfaccia utente e i log di Airflow diventano indispensabili per capire cosa sta succedendo. In questo esempio, il log di print_date conterrebbe la data di sistema stampata, e il log di echo_hello conterrebbe “Hello, Airflow!”. Se print_date fallisse (forse non riusciva a trovare il comando date, in uno scenario ipotetico), allora echo_hello non verrebbe mai eseguito (a causa della dipendenza), e l'esecuzione del DAG sarebbe contrassegnata come fallita. Potresti quindi controllare il log, risolvere il problema e rieseguire quel compito o l'intero DAG.

Casi d'uso comuni per Airflow

Di seguito è riportato un riepilogo strutturato dei casi d'uso comuni di Apache Airflow. La tabella evidenzia cosa comporta ciascun caso d'uso e come Airflow apporta valore in quel contesto.

Quando NON usare Airflow

Sebbene Airflow sia potente, non è lo strumento giusto per ogni situazione. Comprendere i suoi limiti ti aiuta a scegliere l'approccio di orchestrazione corretto:

• Carichi di lavoro in tempo reale o in streaming: Airflow è orientato a batch con un inizio e una fine ben definiti. Non è adatto per carichi di lavoro di lunga durata, basati su eventi o in streaming. Se stai acquisendo flussi continui di eventi (clic degli utenti, sensori IoT, ecc.) e hai bisogno di elaborarli con una latenza molto bassa in tempo reale, Airflow non è la scelta migliore.
• Attività ad alta frequenza o molte attività brevi: Questo è un altro caso in cui Airflow potrebbe non funzionare, o almeno non in modo efficiente. Le attività di Airflow hanno un certo overhead (nel tracciamento del database, ecc.), quindi se hai bisogno di attivare qualcosa ogni pochi secondi o hai migliaia di piccole attività, Airflow potrebbe non scalare bene.
• Flussi di lavoro puramente basati su eventi: Questo è simile al caso d'uso dello streaming, ma per qualcosa che dovrebbe essere attivato da un evento. "Ogni volta che accade un evento X, fai Y" è un caso d'uso perfettamente valido per Airflow, ma se l'unica cosa che dovrebbe attivarlo è un evento (ad esempio, "ogni volta che un file arriva in un bucket S3, fai X"), allora Airflow non è la soluzione più leggera in questo caso. Airflow generalmente utilizza sensori (ad esempio S3KeySensor) per "rilevare" la presenza di file/eventi. I sensori hanno degli svantaggi in quanto si basano sul polling. I problemi dei sensori includono il fatto di attivarsi su file già esistenti e di non essere necessariamente reattivi agli eventi (ad esempio, eseguono il polling a intervalli).

In breve, Airflow è adatto a flussi di lavoro periodici, orientati ai batch o complessi da definire e gestire. Potrebbe non essere una buona scelta se il tuo caso d'uso richiede elaborazioni in tempo reale o continue, lavori molto frequenti o flussi di lavoro molto semplici che non necessitano del sovraccarico di Airflow. In tal caso, potresti voler esplorare altre opzioni o semplificare la tua soluzione.

Alternative ad Airflow

Lo spazio dell'orchestrazione dei flussi di lavoro è esploso negli ultimi anni e, sebbene Airflow sia un leader, ci sono diversi strumenti alternativi che vale la pena conoscere:

Airflow non è l'unica opzione disponibile, e la scelta migliore dipende dal tuo caso d'uso. Luigi, Prefect e Dagster sono comunemente elencati come principali altre opzioni open-source nello stesso ambito (orchestratori di workflow basati su Python). Se trovi la UI datata di Airflow o altre sue limitazioni scomode, dovresti prendere in considerazione di valutare queste alternative. Prefect cerca di essere un Airflow "migliore" o "più semplice", Dagster cerca di essere un orchestratore più "strutturato" con enfasi sugli asset di dati, e Luigi è un predecessore più semplice di Airflow. D'altra parte, se il tuo caso d'uso diverge significativamente (streaming in tempo reale o completamente cloud-native, per esempio), potresti non usare Airflow affatto e considerare direttamente piattaforme di streaming o servizi di orchestrazione gestiti.

Considerazioni Finali

Apache Airflow è una piattaforma open-source per orchestrare flussi di lavoro computazionali complessi e pipeline di elaborazione dei dati. Originariamente sviluppata da Airbnb, ha acquisito una significativa popolarità nelle comunità di data engineering e machine learning. I componenti di Airflow includono un scheduler, un executor, dei worker, un database di metadata e un'interfaccia web. Questo consente ai team di eseguire e mantenere facilmente pipeline ETL, MLOps e infrastrutturali pronte per la produzione. L'astrazione DAG (Directed Acyclic Graph) di Airflow permette di mantenere i flussi di lavoro espliciti, testabili e manutenibili nel tempo. L'ecosistema della piattaforma, costituito da operatori e provider sviluppati dalla comunità, semplifica inoltre l'orchestrazione dei flussi di lavoro che dipendono da qualsiasi tecnologia o strumento.

Tuttavia, non è una soluzione universale. Funziona meglio con pipeline batch o micro-batch, e gli utenti devono sentirsi a proprio agio con Python. Per flussi di lavoro in streaming o ad alta frequenza, ci sono altre soluzioni più adatte di Airflow. Per data scientist o team che preferiscono scrivere pipeline dichiarative, potrebbe valere la pena considerare strumenti alternativi ad Airflow.

Airflow è ancora attivamente sviluppato e nuove funzionalità vengono aggiunte nel tempo, come la pianificazione basata su eventi, i dataset e gli asset. I concetti e le best practice introdotti in questa guida dovrebbero aiutarti, indipendentemente dal tuo background, a comprendere come funziona l'orchestrazione dei workflow e quando Airflow dovrebbe essere utilizzato (o meno). Si spera che questo possa servire come punto di partenza per esplorare ulteriormente questo ricco universo.

Riferimenti

• Apache Airflow
• Cos'è Airflow?
• Apache Airflow: casi d'uso, architettura e 6 consigli per il successo
• Introduzione per principianti ad Airflow con Docker — Pipeline di dati nel cloud (I)
• 8 alternative ad Airflow - Casi d'uso
• Panoramica dell'architettura
• Pianificatore

Grazie per aver imparato con la community di Linux-Console.net.