L'API TaskFlow

Creare data pipeline con Airflow

Volker Janz

Senior Developer Advocate at Astronomer

L'approccio classico

def extract_data():
    return {"users": 150, "events": 4200}


with DAG("etl_pipeline") as dag:
    t1 = PythonOperator(task_id="extract",
                        python_callable=extract_data)
    t2 = PythonOperator(task_id="summary",
                        python_callable=print_summary)

    t1 >> t2

L'approccio TaskFlow

from airflow.sdk import dag, task


@dag
def etl_pipeline():
    @task
    def extract_data():
        return {"users": 150}


    data = extract_data()
    print_summary(data)

Perché TaskFlow?

Meno boilerplate, i decorator sostituiscono gli operatori
Dipendenze implicite, i valori di ritorno collegano i task in automatico
Leggibile, il Dag sembra uno script Python
Operatori classici ancora disponibili per le integrazioni dei provider

TaskFlow API - visual

Airflow UI: Grid view

Vista Grid di Airflow

Ogni colonna è una Dag run
Ogni riga è un task
I colori mostrano lo stato: verde = successo, rosso = errore
Clicca un quadrato per vedere log e dettagli

Airflow UI: Graph view

Vista Graph di Airflow

Si concentra su una singola Dag run
Mostra dipendenze ed esecuzione in parallelo
Regola i dettagli nelle impostazioni

Airflow UI: versioning dei Dag

Indicatore versione Dag nell'interfaccia Airflow

Traccia le modifiche strutturali automaticamente
Ogni run è collegata alla versione attiva in quel momento
Trova le versioni nel pannello Dettagli del Dag

Esercitiamoci!

Creare data pipeline con Airflow