# Apache Airflow nel 2026: Orchestrazione di Pipeline, DAG e Domande per Colloqui Tecnici

> Guida completa ad Apache Airflow 3.2 nel 2026: creazione di DAG con Task SDK, dynamic task mapping, asset partitions, confronto con Prefect e Dagster, e domande frequenti nei colloqui tecnici.

- Published: 2026-06-13
- Updated: 2026-06-13
- Author: SharpSkill
- Tags: apache-airflow, data-engineering, pipeline-orchestration, dag, interview
- Reading time: 12 min

---

Nel panorama del data engineering moderno, Apache Airflow continua a rappresentare lo standard de facto per l'orchestrazione di pipeline di dati complesse. Con il rilascio della versione 3.2, il progetto ha compiuto un salto qualitativo significativo, introducendo funzionalità che rispondono alle esigenze delle organizzazioni enterprise e dei team di sviluppo distribuiti. La gestione di workflow complessi, la schedulazione basata su asset e il supporto nativo per operazioni asincrone posizionano Airflow come strumento indispensabile per chiunque lavori con dati su larga scala.

> **Novità di Apache Airflow 3.2**
>
> La versione 3.2 introduce il Task SDK rinnovato con importazioni semplificate da `airflow.sdk`, supporto nativo per task asincroni con `async/await`, asset partitions con schedulazione multi-asset e isolamento multi-team sperimentale per ambienti enterprise. Queste funzionalità rappresentano una maturazione significativa della piattaforma verso scenari di produzione sempre più complessi.

## Creazione di DAG con il Task SDK

Il Task SDK di Airflow 3.2 semplifica radicalmente la definizione di Directed Acyclic Graph (DAG), il costrutto fondamentale che descrive le dipendenze tra task. L'approccio basato su decoratori Python elimina gran parte del boilerplate tradizionale, permettendo agli sviluppatori di concentrarsi sulla logica di business.

Un esempio concreto illustra la creazione di una pipeline ETL per l'aggregazione giornaliera dei ricavi:

```python
# etl_daily_revenue.py
import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["finance", "etl"],
)
def etl_daily_revenue():
    @task()
    def extract_transactions() -> list[dict]:
        # Pulls raw transactions from the payments API
        import requests
        response = requests.get("https://api.internal/payments/daily")
        return response.json()["transactions"]

    @task(multiple_outputs=True)
    def transform(transactions: list[dict]) -> dict:
        # Aggregates by currency and computes totals
        totals = {}
        for tx in transactions:
            currency = tx["currency"]
            totals[currency] = totals.get(currency, 0) + tx["amount"]
        return {"totals": totals, "count": len(transactions)}

    @task()
    def load(totals: dict, count: int):
        # Inserts aggregated row into the warehouse
        from warehouse import insert_revenue_summary
        insert_revenue_summary(totals=totals, transaction_count=count)

    raw = extract_transactions()
    result = transform(raw)
    load(result["totals"], result["count"])

etl_daily_revenue()
```

Il decoratore `@dag` definisce i metadati della pipeline, inclusa la schedulazione cron (`@daily`), la data di inizio e i tag per l'organizzazione. Ogni funzione decorata con `@task()` diventa automaticamente un'unità eseguibile indipendente. L'opzione `multiple_outputs=True` permette di restituire dizionari i cui valori vengono automaticamente spacchettati come XCom separati, facilitando il passaggio di dati strutturati tra task.

La chiamata finale `etl_daily_revenue()` registra il DAG nel sistema. Questo pattern dichiarativo rende il codice leggibile e manutenibile, caratteristiche essenziali per team che gestiscono centinaia di pipeline.

## Dynamic Task Mapping per Elaborazioni Parallele

Una delle funzionalità più potenti introdotte nelle versioni recenti di Airflow è il dynamic task mapping, che permette di creare istanze multiple di un task a runtime basandosi sui dati effettivi. Questo elimina la necessità di definire staticamente il numero di task paralleli.

```python
# etl_multi_region.py
import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@hourly",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["regional", "etl"],
)
def etl_multi_region():
    @task()
    def get_regions() -> list[str]:
        # Returns the active regions from config
        return ["us-east-1", "eu-west-1", "ap-southeast-1"]

    @task()
    def process_region(region: str) -> dict:
        # Processes events for a single region
        from data_processor import aggregate_events
        return aggregate_events(region)

    @task()
    def merge_results(results: list[dict]):
        # Combines all regional outputs into a single report
        from reporter import publish_global_report
        publish_global_report(results)

    regions = get_regions()
    # .expand() creates one mapped task instance per region
    processed = process_region.expand(region=regions)
    merge_results(processed)

etl_multi_region()
```

Il metodo `.expand()` rappresenta il cuore del dynamic task mapping. Quando `get_regions()` restituisce tre regioni, Airflow crea automaticamente tre istanze separate di `process_region`, ognuna con il proprio parametro `region`. Queste istanze vengono eseguite in parallelo secondo le risorse disponibili nel cluster.

> **Limiti dei Mapped Task**
>
> Per impostazione predefinita, Airflow limita a 1024 il numero massimo di mapped task instances per singolo task. Questo valore è configurabile tramite il parametro `max_map_length` nel file di configurazione. Per elaborazioni che richiedono migliaia di istanze parallele, è consigliabile implementare un batching logico a livello applicativo.

Il task `merge_results` riceve automaticamente una lista contenente tutti i risultati delle istanze mappate, permettendo aggregazioni finali senza logica di coordinamento esplicita.

## Asset Partitions e Schedulazione Multi-Asset

La schedulazione basata su asset rappresenta un cambio di paradigma rispetto alla tradizionale schedulazione temporale. Invece di eseguire pipeline a orari predefiniti, i DAG possono attivarsi quando determinati asset (dataset, file, tabelle) vengono aggiornati. Airflow 3.2 estende questo concetto con le partizioni.

```python
# downstream_analytics.py
import pendulum
from airflow.sdk import dag, task
from airflow.timetables.assets import CronPartitionTimetable

# Define partitioned assets with hourly granularity
nba_stats = Asset("s3://datalake/nba/hourly/", partitions=CronPartitionTimetable("0 * * * *"))
epl_stats = Asset("s3://datalake/epl/hourly/", partitions=CronPartitionTimetable("0 * * * *"))
nfl_stats = Asset("s3://datalake/nfl/hourly/", partitions=CronPartitionTimetable("0 * * * *"))

@dag(
    # Triggers only when all three assets have matching partition
    schedule=(nba_stats & epl_stats & nfl_stats),
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def unified_sports_analytics():
    @task()
    def aggregate_all_leagues(**context):
        partition = context["partition"]
        # Process only the specific hourly slice across all leagues
        from analytics import build_cross_league_report
        build_cross_league_report(partition=partition)

    aggregate_all_leagues()

unified_sports_analytics()
```

L'operatore `&` definisce una dipendenza congiunta: il DAG si attiva solo quando tutte e tre le sorgenti di dati sportivi hanno una partizione oraria corrispondente disponibile. Questo garantisce che i report cross-league vengano generati esclusivamente quando i dati sono completi e coerenti.

Il `CronPartitionTimetable` con pattern `"0 * * * *"` crea partizioni orarie. Quando un producer upstream marca una partizione come completata, Airflow valuta se le condizioni di scheduling del consumer sono soddisfatte.

## Task Asincroni Nativi

Una delle aggiunte più significative di Airflow 3.2 è il supporto nativo per task asincroni. Operazioni I/O-bound come download di file, chiamate API multiple o interazioni con database possono ora sfruttare `async/await` per una concorrenza efficiente senza threading esplicito.

```python
# async_file_download.py
import asyncio
import aiohttp
from airflow.sdk import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["async", "download"],
)
def async_file_download():
    @task()
    async def download_files():
        urls = [f"https://data.source/files/{i}.csv" for i in range(500)]
        async with aiohttp.ClientSession() as session:
            # Downloads 500 files concurrently instead of sequentially
            tasks = [fetch_and_save(session, url) for url in urls]
            results = await asyncio.gather(*tasks)
        return {"downloaded": len(results)}

    download_files()

async def fetch_and_save(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as resp:
        content = await resp.read()
        filename = url.split("/")[-1]
        with open(f"/data/downloads/{filename}", "wb") as f:
            f.write(content)
        return filename

async_file_download()
```

Questo esempio scarica 500 file CSV in modo concorrente. Utilizzando `asyncio.gather()`, tutte le richieste HTTP vengono lanciate simultaneamente, riducendo drasticamente il tempo totale rispetto a un approccio sequenziale. La funzione `fetch_and_save` gestisce il download e la persistenza di ogni singolo file.

Il supporto async elimina la necessità di workaround come pool di thread esterni o operatori custom per scenari ad alta concorrenza I/O, semplificando significativamente il codice delle pipeline.

## Architettura e Componenti Fondamentali

L'architettura di Airflow si basa su componenti distribuiti che collaborano per garantire affidabilità e scalabilità. Lo **Scheduler** rappresenta il cervello del sistema: monitora continuamente i DAG, determina quali task sono pronti per l'esecuzione e li inserisce nella coda. In Airflow 3.x, lo scheduler supporta configurazioni ad alta disponibilità con multiple istanze attive.

I **Worker** eseguono i task effettivi. La scelta dell'executor determina come i worker vengono orchestrati: il CeleryExecutor distribuisce i task su un cluster di worker tramite message broker come Redis o RabbitMQ, mentre il KubernetesExecutor crea pod dedicati per ogni task, garantendo isolamento e scalabilità elastica.

> **Scelta dell'Executor in Produzione**
>
> Il LocalExecutor, pur essendo comodo per sviluppo e test, non è adatto ad ambienti di produzione con carichi significativi. Per deployment enterprise, il KubernetesExecutor offre il miglior compromesso tra isolamento, scalabilità e gestione delle risorse. Il CeleryExecutor rimane una scelta valida per organizzazioni con infrastruttura Celery esistente.

Il **Metadata Database** (tipicamente PostgreSQL) persiste lo stato di ogni DAG run e task instance, abilitando retry, monitoraggio storico e debugging. La **Web UI** fornisce visibilità completa sullo stato delle pipeline, con grafici delle dipendenze, log in tempo reale e strumenti di troubleshooting.

## Confronto tra Airflow, Prefect e Dagster nel 2026

Il mercato dell'orchestrazione di pipeline nel 2026 presenta tre attori principali, ognuno con filosofie e punti di forza distinti.

| Feature | Airflow 3.2 | Prefect 3.x | Dagster 1.9 |
|---|---|---|---|
| **DAG definition** | Python decorators (`airflow.sdk`) | Python decorators (`@flow`, `@task`) | Python decorators (`@asset`, `@op`) |
| **Scheduling** | Cron, asset-aware, partition-aware | Cron, event-driven | Cron, sensor-based, asset-aware |
| **Execution model** | Centralized scheduler + distributed workers | Hybrid (server + work pools) | Centralized dagster-daemon |
| **Dynamic tasks** | `.expand()` mapped tasks | Native Python loops | Dynamic partitions |
| **Async support** | Native in 3.2 | Native since 2.0 | Async I/O ops |
| **Multi-team isolation** | Built-in (3.2 experimental) | Workspace-based (Cloud) | Branch deployments |
| **Community size** | Largest (35k+ GitHub stars) | Growing (18k+ stars) | Growing (12k+ stars) |
| **Best fit** | Complex, multi-team pipelines with established infrastructure | Smaller teams wanting fast iteration | Data asset-centric organizations |

Airflow eccelle in scenari enterprise con pipeline complesse e team multipli che necessitano di governance centralizzata. La sua comunità estesa garantisce disponibilità di risorse, integrazioni e supporto. Prefect si distingue per la velocità di iterazione e un'esperienza sviluppatore moderna, risultando ideale per team agili che privilegiano la semplicità. Dagster, con il suo approccio asset-centric, si adatta particolarmente a organizzazioni dove il lineage dei dati e la qualità sono priorità fondamentali.

## Domande Frequenti nei Colloqui Tecnici su Airflow

I colloqui per posizioni di data engineering includono frequentemente domande su Airflow. Ecco le aree tematiche più comuni e gli elementi chiave da padroneggiare.

**Concetti Fondamentali**: I candidati devono saper spiegare la differenza tra DAG, task e operator. Un DAG definisce le dipendenze, un task è un'istanza di esecuzione, un operator è il template che definisce cosa fa il task. La comprensione del ciclo di vita di un task (queued, running, success, failed, up_for_retry) è essenziale.

**XCom e Passaggio di Dati**: Gli XCom (cross-communication) permettono ai task di scambiarsi piccole quantità di dati. Per dataset più grandi, le best practice suggeriscono l'utilizzo di storage esterno (S3, GCS) con riferimenti passati via XCom. Airflow 3.x introduce limiti configurabili sulla dimensione degli XCom per prevenire problemi di performance del metadata database.

**Sensor e Trigger**: I sensor sono operator speciali che attendono condizioni esterne (file esistente, partizione disponibile, API pronta). La domanda tipica riguarda la differenza tra mode "poke" (controllo continuo) e mode "reschedule" (rilascio del worker slot tra i controlli).

**Backfill e Catchup**: Il parametro `catchup=True` esegue tutte le run mancate dalla `start_date`. Comprendere quando abilitarlo o disabilitarlo è cruciale per evitare esecuzioni indesiderate o mancate.

**Branching e Condizionali**: Il `BranchPythonOperator` permette esecuzioni condizionali basate sulla logica runtime. I candidati dovrebbero saper spiegare come gestire i task downstream quando alcuni rami non vengono eseguiti.

Per approfondire la preparazione ai colloqui tecnici su Airflow, la piattaforma [SharpSkill offre moduli dedicati](/technologies/data-engineering/interview-questions/airflow-fundamentals) con domande pratiche, quiz interattivi e simulazioni di colloquio focalizzate sull'orchestrazione di pipeline.

## Best Practice per Ambienti di Produzione

La gestione di Airflow in produzione richiede attenzione a diversi aspetti critici. L'**idempotenza** dei task garantisce che esecuzioni ripetute producano risultati consistenti, requisito fondamentale per gestire retry automatici senza corruzione dei dati.

La **modularizzazione dei DAG** attraverso factory pattern e configurazioni esterne facilita la manutenzione. Invece di duplicare codice, è preferibile creare funzioni generatrici di DAG parametrizzate.

Il **monitoraggio proattivo** tramite integrazione con sistemi come Prometheus, Datadog o CloudWatch permette di identificare degradazioni prima che impattino i processi di business. Metriche chiave includono task duration, queue depth e failure rate.

La **gestione delle credenziali** attraverso Connections e Variables crittografate, preferibilmente integrate con secret manager esterni (HashiCorp Vault, AWS Secrets Manager), elimina il rischio di esporre informazioni sensibili nel codice.

## Conclusione

Apache Airflow 3.2 consolida la posizione della piattaforma come riferimento per l'orchestrazione di pipeline dati enterprise. I punti chiave da ricordare:

- Il **Task SDK** con decoratori `@dag` e `@task` semplifica drasticamente la definizione di workflow
- Il **dynamic task mapping** con `.expand()` permette parallelismo runtime-driven senza configurazione statica
- Gli **asset partitions** abilitano schedulazione data-driven con allineamento automatico delle partizioni
- Il **supporto async nativo** ottimizza operazioni I/O-bound con concorrenza efficiente
- L'**architettura distribuita** con scheduler HA e executor scalabili supporta carichi enterprise
- Il confronto con **Prefect e Dagster** evidenzia come Airflow eccella per complessità e governance multi-team
- Le **domande di colloquio** si concentrano su XCom, sensor, backfill e branching
- Le **best practice** includono idempotenza, modularizzazione, monitoraggio e gestione sicura delle credenziali

---

Source: SharpSkill (https://sharpskill.dev), tech interview preparation for your real stack.
HTML version of this page: https://sharpskill.dev/it/blog/data-engineering/apache-airflow-pipeline-orchestration-dags-interview-2026