L'API TaskFlow

Créer des pipelines de données avec Airflow

Volker Janz

Senior Developer Advocate at Astronomer

L'approche classique

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'approche TaskFlow

from airflow.sdk import dag, task


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


    data = extract_data()
    print_summary(data)

Pourquoi TaskFlow ?

Moins de passe-partout : des décorateurs remplacent les instances d'opérateurs
Dépendances implicites : les valeurs de retour relient les tâches automatiquement
Lisible : un Dag se lit comme un script Python
Les opérateurs classiques restent offerts pour les intégrations de fournisseurs

API TaskFlow - visuel

Interface Airflow : vue Grille

Vue Grille d'Airflow

Chaque colonne est une exécution de Dag
Chaque rangée est une tâche
Les couleurs indiquent l'état : vert = succès, rouge = échec
Cliquez un carré pour voir les journaux et détails

Interface Airflow : vue Graphe

Vue Graphe d'Airflow

Se concentre sur une seule exécution de Dag
Met en évidence les dépendances et l'exécution parallèle
Ajustez les détails dans les paramètres

Interface Airflow : gestion des versions de Dag

Indicateur de version du Dag dans l'interface Airflow

Suit automatiquement les changements structurels
Chaque exécution est liée à la version active à ce moment
Trouvez les versions dans le panneau Détails du Dag

Passons à la pratique !

Créer des pipelines de données avec Airflow