Technical Tutorials

Airflow est depuis longtemps la référence de facto pour l'orchestration des flux de données. Cependant, à mesure que les volumes de données augmentent et que les sources deviennent plus fragmentées, les DAG statiques peinent souvent à assurer l'évolutivité et la maintenabilité. Dans Airflow 2.0 et les versions ultérieures, l'introduction du mappage de tâches et de la génération dynamique de DAG affinée offre des outils robustes pour construire des pipelines qui s'adaptent aux structures de données changeantes sans sacrifier les performances ni la clarté.

Pourquoi les DAG statiques échouent à grande échelle

Les DAG Airflow traditionnels sont définis au moment de l'analyse (parse time). Si votre pipeline doit traiter 100 fichiers, vous pourriez créer 100 tâches. Cette approche conduit à :

Gonflement du code : Des fichiers DAG massifs avec des définitions de tâches répétitives.
Temps d'analyse longs : Airflow met trop de temps à analyser des DAG complexes, ce qui ralentit le planificateur.
Manque de flexibilité : Vous ne pouvez pas gérer un nombre inconnu de fichiers sans coder en dur des limites ou utiliser des boucles complexes.

La solution consiste à passer de « nombreuses tâches » à « une tâche mappée » qui gère plusieurs éléments.

Introduction au mappage de tâches

Le mappage de tâches dans Airflow 2.0 vous permet de créer plusieurs instances de tâches à partir d'une seule définition de tâche. Lorsque vous utilisez la méthode x.map(), Airflow crée une exécution séparée pour chaque élément de l'itérable fourni (liste, tuple ou générateur).

Syntaxe de base

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    dag_id="task_mapping_example",
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:

    def process_file(file_name):
        print(f"Processing {file_name}")
        # Ajoutez votre logique de traitement ici
        return file_name

    # Définir une liste de fichiers à traiter
    files = ["file1.csv", "file2.csv", "file3.csv", "file4.csv", "file5.csv"]

    # Mapper la tâche sur la liste de fichiers
    map_task = PythonOperator(
        task_id="process_file_task",
        python_callable=process_file,
        op_kwargs={"file_name": files}  # Note : Dans Airflow 2.3+, utilisez 'op_args' ou 'op_kwargs' avec prudence
    )
    
    # En réalité, la méthode standard pour mapper une liste d'éléments vers un argument de fonction :
    # Si process_file prend un argument et que nous voulons passer chaque élément :
    
    # Implémentation corrigée pour Airflow 2.0+ :
    def process_item(item):
        print(f"Processing: {item}")
        return item

    # Nous devons passer la liste comme argument à mapper
    # L'utilisation de op_args n'est pas directement supportée pour mapper les éléments d'une liste dans toutes les versions.
    # Bonne pratique : Utiliser un wrapper ou passer la liste en op_kwargs et la dépaqueter ?
    # En fait, Airflow 2.3+ recommande :
    
    mapped_task = PythonOperator(
        task_id="process_items",
        python_callable=process_item,
        op_args=files  # Ce n'est PAS ainsi que le mappage fonctionne directement.
    )

Correction et bonne pratique : Dans Airflow, vous ne passez pas simplement une liste à op_args pour la mapper. Vous devez utiliser la méthode .x.map() sur l'instance de l'opérateur. Cependant, PythonOperator n'a pas de méthode .x.map() directe de la même manière que BashOperator pour des chaînes simples. Pour PythonOperator, le schéma courant consiste à passer la liste entière à la fonction et à gérer l'itération à l'intérieur, OU à utiliser un sous-classement générique de BaseOperator, OU plus couramment, à utiliser op_kwargs où la valeur est la liste, mais l'appelable doit accepter cette liste.

En fait, la méthode la plus robuste pour démontrer le mappage de tâches pour une liste d'éléments dans un PythonOperator est de passer la liste comme argument et de laisser Airflow la découper. Mais attention, le mappage de PythonOperator fonctionne en passant chaque élément de la liste mappée à l'appelable.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def process_file(file_name):
    print(f"Processing: {file_name}")
    return f"Processed {file_name}"

with DAG(
    dag_id="dynamic_processing",
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
) as dag:

    # Simuler une liste dynamique de fichiers (par ex., depuis une requête DB ou un bucket S3)
    files = [f"file_{i}.csv" for i in range(1, 101)]

    # Définir la tâche
    process_task = PythonOperator(
        task_id="process_files",
        python_callable=process_file
    )

    # Mapper la tâche. Cela créera 100 instances de tâches.
    # La liste `files` est passée comme argument à mapper.
    process_task.x.map(process_task, files)

Note importante : La syntaxe .x.map() ci-dessus est légèrement simplifiée pour la clarté. En pratique, vous utilisez souvent op_kwargs pour spécifier quel paramètre reçoit les éléments mappés. Si votre fonction prend plusieurs arguments, vous devez spécifier l'index de l'argument à mapper.

def process_file(file_name, log_level="INFO"):
    print(f"[{log_level}] Processing: {file_name}")
    return f"Processed {file_name}"

# ... à l'intérieur du DAG ...
process_task = PythonOperator(
    task_id="process_files",
    python_callable=process_file,
    op_kwargs={"log_level": "DEBUG"}  # Argument par mot-clé fixe
)

# Mapper sur le premier argument (file_name)
# index=0 fait référence au premier argument positionnel de l'appelable
process_task.x.map(process_task, files, index=0)

Gérer les données dynamiques : Générateurs et Contexte

Souvent, vous ne connaissez pas la liste des éléments au moment de la définition du DAG. C'est là que la génération dynamique de DAG et les générateurs brillent. Au lieu de construire une liste massive en mémoire, utilisez un générateur pour produire les éléments un par un. C'est crucial pour les pipelines à grande échelle où le nombre d'éléments peut atteindre des milliers.

import boto3

def get_s3_files(bucket, prefix):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            yield obj['Key']

# À l'intérieur de la définition du DAG :
# Note : Vous ne pouvez pas facilement utiliser un générateur qui accède à S3 au moment de l'analyse si la
# structure du bucket change dynamiquement à chaque exécution.
# Bonne pratique : Utiliser une tâche pour récupérer la liste, puis mapper en fonction de cette sortie.

Meilleures pratiques pour l'évolutivité

Limiter le nombre de tâches : Bien que le mappage de tâches résolve le problème de code « nombreuses tâches », avoir 10 000 instances de tâches dans une seule exécution peut surcharger le planificateur et la base de données Airflow. Envisagez de regrouper les éléments en lots si le dépasse 500-1000.
Utiliser le traitement par lots : Si chaque élément est petit, regroupez les éléments en lots (par ex., 50 fichiers par tâche) pour réduire la surcharge liée à l'initialisation des tâches et à la journalisation.
Surveiller les performances : Suivez le nombre d'instances de tâches générées par exécution. Utilisez des métriques comme airflow_dagrun_taskinstance_run_duration pour identifier les goulets d'étranglement.
Gestion des erreurs : Par défaut, si une tâche mappée échoue, l'exécution entière est marquée comme échouée. Utilisez partial_mapping=True (disponible dans les versions plus récentes) ou une logique personnalisée si vous devez tolérer des échecs individuels sans arrêter tout le pipeline.

Conclusion

Airflow 2.0+ fournit des fonctionnalités puissantes pour construire des pipelines de données qui s'adaptent à vos besoins métier. En exploitant le mappage de tâches, vous pouvez passer de DAG fragiles et codés en dur à des flux de travail flexibles et dynamiques qui gèrent efficacement les volumes de données variables. N'oubliez pas de surveiller les nombres d'instances de tâches et d'envisager le traitement par lots pour les très grands jeux de données afin de vous assurer que votre planificateur Airflow reste réactif. Lorsque vous concevez votre prochain pipeline, demandez-vous : « Est-ce que cela peut être mappé ? » Si la réponse est oui, embrassez la puissance de la génération dynamique de tâches.