Airflow — Cheat Sheet

Tag: airflow, orchestration, python, dag, data-pipeline, etl, scheduler

Livello: intermedio

TL;DR: Apache Airflow orchestra pipeline dati complesse come grafi aciclici diretti (DAG) in Python. Gestisce dipendenze di esecuzione, scheduling, retry e monitoraggio di workflow eterogenei.

Concetti chiave

  • DAG: Grafo Aciclico Diretto che definisce task, sequenza logica e relazioni di dipendenza senza cicli.
  • Task e Operator: L'Operator definisce la logica del blocco; il Task ne rappresenta l'istanza configurata nel DAG.
  • Scheduler e Executor: Lo Scheduler monitora le scadenze e assegna i task pronti all'Executor designato per l'esecuzione.
  • Tipologie di Executor: Supporta SequentialExecutor, LocalExecutor, CeleryExecutor e KubernetesExecutor per scalare orizzontalmente i worker.
  • Metadata Database: Database relazionale centrale che conserva lo stato di DAG, task, run, log e credenziali.
  • Task Lifecycle: I task transitano per stati definiti, tra cui none, scheduled, queued, running, success, up_for_retry o failed.
  • XCom: Canale di Cross-Communication per scambiare piccoli messaggi di testo o metadati JSON tra task diversi.
  • Sensors: Operatori dedicati che restano in attesa che una condizione esterna o un file si manifesti.
  • Catchup e Backfill: Capacità di eseguire a ritroso le istanze temporali di un DAG rimaste inevase.
  • Variables e Connections: Archivio protetto su DB per stringhe di configurazione e credenziali d'accesso a database o API esterne.

Domande tipiche e risposte pronte

  • Q: Come gestisci lo scambio di volumi di dati consistenti tra due task successivi?

R: Evito XCom per dati pesanti; salvo il dataset su object storage come S3 o GCS e trasmetto tramite XCom soltanto il path dell'oggetto.

  • Q: Qual è la differenza operativa tra un Operator e un Sensor?

R: Un Operator esegue attivamente un'elaborazione, mentre un Sensor resta in polling su un evento esterno prima di sbloccare i nodi a valle.

  • Q: Come eviti la saturazione degli worker slot quando molti sensori sono attivi?

R: Configuro il sensore con mode='reschedule' invece del default poke, consentendo al task di cedere lo slot tra un controllo e l'altro.

  • Q: Come scegli tra CeleryExecutor e KubernetesExecutor?

R: Scelgo Celery se ho worker fissi con overhead minimo; scelgo Kubernetes se ho task con dipendenze isolate o carichi altamente elastici.

Insidie e errori comuni

  • Top-level code nel file DAG: chiamate esterne o query pesanti nel file Python rallentano il parsing continuo dello scheduler.
  • Abuso di XCom: salvare interi DataFrame o payload binari su XCom degrada le performance del metadata database.
  • Sensori in poke mode prolungati: consumano tutti gli slot dei worker impedendo l'avvio degli altri task del cluster.
  • Uso di datetime.now() in start_date: l'uso di date dinamiche impedisce il calcolo deterministico delle esecuzioni passate e future.
  • Catchup involontario: mantenere catchup=True su un DAG storico genera centinaia di esecuzioni contemporanee non previste.

Snippet e comandi

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.filesystem import FileSensor

default_args = {
    "owner": "umberto",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="daily_orders_pipeline",
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule_interval="@daily",
    catchup=False,
    max_active_runs=1,
) as dag:
    wait_for_input = FileSensor(
        task_id="wait_for_input",
        filepath="/data/raw/orders.csv",
        mode="reschedule",
        poke_interval=60,
        timeout=1800,
    )

    def extract_and_clean(ti, **context):
        execution_date = context["ds"]
        return f"s3://datalake/cleaned/{execution_date}/orders.parquet"

    clean_data = PythonOperator(
        task_id="clean_data",
        python_callable=extract_and_clean,
    )

    wait_for_input >> clean_data

Bash

# Validazione sintattica del file DAG
python path/to/daily_orders_pipeline.py

# Test locale di un task senza avviare lo scheduler
airflow tasks test daily_orders_pipeline clean_data 2026-03-01

# Trigger manuale con passaggio di parametri
airflow dags trigger -c '{"backfill_flag": true}' daily_orders_pipeline

# Ispezione dello stato delle esecuzioni
airflow dags list-runs -d daily_orders_pipeline --state running

Da ripassare

[ ] TaskFlow API con decoratori @dag e @task per una gestione trasparente degli XCom

[ ] Implementazione e configurazione di Cluster Policies per enforcement degli standard

[ ] Differenze tra Data Interval, Execution Date e Logical Date in Airflow 2.x

[x] Differenza tra CeleryExecutor e KubernetesExecutor

[x] Utilizzo della modalità reschedule sui sensori per liberare slot worker

Riferimenti

Documentazione ufficiale Apache Airflow (airflow.apache.org)

Airflow Architecture and Core Concepts Guide

Best Practices for DAG Writing in Apache Airflow

Jarvis Interviews
Jarvis