Un DAG Airflow est du code Python, et la tentation est grande de mettre ce code à profit pour générer de la structure : une boucle qui crée dix tâches, un groupe de tâches par source de données, une topologie qui épouse la configuration. C’est ce que l’on appelle un “DAG dynamique”. Derrière ce terme unique se cachent en réalité deux mécanismes très différents, dont le choix conditionne la robustesse de tout ce qui suit.
Deux façons de générer la structure
La confusion entre les deux familles de DAGs dynamiques est une source classique de mauvaises décisions d’architecture. Il faut les distinguer nettement.
La génération au parse-time.
Du code Python, typiquement une boucle for, crée N tâches au moment où l’orchestrateur lit le fichier. La topologie est figée dès le parsing.
# Parse-time: the loop runs every time the DAG file is parsed
for source in get_sources():
build_task_group(source)Le piège est dans get_sources: cette liste doit être disponible au parsing, donc relue à chaque évaluation du fichier. Cela signifie que si vous souhaitez passer des sources différentes à l'exécution vous ne pouvez pas. Ce mode reste légitime pour une configuration statique et locale, mais dès que l’entrée est “vivante”, c’est un anti-pattern.
Le dynamic task mapping.
Introduit sous Airflow 2.3, il déporte la décision à l'exécution. Une tâche amont calcule la liste pendant l’exécution, la pousse en XCom, et Airflow déplie autant d’instances mappées qu’il y a d’éléments. La structure du DAG reste stable, qu'il s'agisse d'une tâche ou d'un groupe de tâches mappé, et la ramification se décide à l’exécution.
Avec notre exemple précédent :
# Runtime: the list is computed inside a task, then mapped
@task
def get_sources():
return [...] # any logic: file, DB, API...
process.expand(source=get_sources())L’interface utilisateur affiche proprement les instances sous forme de tâches mappées et surtout l'analyseur de DAG (en charge du parsing) ne paie jamais le coût de get_sources : il ne voit qu’une seule tâche mappée, peu importe qu’elle se déplie en trois ou trois cents instances à l’exécution.
Dès qu'un cas est piloté par un input susceptible de changer (configuration, contenu d'un bucket, résultat d'une exécution), c'est le mapping que l'on veut. Ainsi, si la liste qui pilote la génération n’est pas une constante figée, elle doit être calculée dans une tâche, pas au parse-time.
Mapper une tâche, puis un groupe de tâches
Le mapping d’une tâche simple est le cas de base : expand reçoit un argument à faire varier, et Airflow crée une instance par valeur.
@task
def process(source: str):
...
process.expand(source=["a", "b", "c"]) # 3 mapped instancesexpand peut recevoir plusieurs paramètres, mais il en fait alors le produit cartésien : une instance par combinaison possible. Par exemple, expand(x=[1, 2], y=[3, 4]) donne quatre instances. Quand on veut au contraire associer explicitement les arguments, on passe à expand_kwargs. Il prend en paramètres une liste de dictionnaires : un par instance, dont les clés correspondent aux arguments de la tâche. Chaque instance reçoit alors un jeu de valeurs complet, sans combinatoire.
Point intéressant : ce mapping peut aussi s'appliquer à un groupe de tâches entier. Décoré avec @task_group, un groupe se mappe exactement comme une tâche. Chaque instance déroule alors l’intégralité du sous-pipeline du groupe, en parallèle des autres.
@task_group
def pipeline(source: str, target: str):
extract(source) >> load(target)
pipeline.expand_kwargs([
{"source": "a", "target": "x"},
{"source": "b", "target": "y"},
])C'est cette dernière forme qui constitue le pattern réutilisable : expand_kwargs sur un groupe de tâches, alimenté par une tâche qui calcule la configuration à l'exécution. Voyons-le sur un cas concret.
Le cas concret : une ingestion pilotée par configuration
Prenons un besoin typique : une même séquence d’étapes – extraire, contrôler, charger – répétée à l’identique sur une vingtaine de datasets. La logique ne change pas, seuls les paramètres et le nombre de datasets à traiter bougent : table source, table cible, clé de déduplication.
Dupliquer un DAG par dataset, c’est autant de fichiers à maintenir que de datasets à traiter. Une boucle au parse-time, c’est le problème vu plus haut. Le bon réflexe : un seul DAG, un seul groupe de tâches extract → check → load, déplié à l'exécution à partir d’une configuration. Ainsi, ajouter un dataset devient ajouter une simple entrée dans la configuration exposée par l'API, sans duplication de code.
Nous omettons volontairement les imports et les arguments obligatoires du DAG pour simplifier la lecture.
D'abord, nous définissons un simple DAG avec un unique groupe de tâches, instancié une fois avec les valeurs données en paramètres. Ce DAG enchaîne simplement 3 tâches, ingestion, vérification, chargement.
with DAG(
dag_id="dynamic_dataset_ingestion",
params={
"source_table": None,
"target_table": None,
"primary_key": None,
},
) as dag:
@task_group(group_id="ingest")
def ingest_dataset(source_table: str, target_table: str, primary_key: str):
@task
def extract(source_table: str) -> int:
print(f"Extracting rows from {source_table}")
return 42
@task
def check(row_count: int, target_table: str) -> int:
if row_count == 0:
raise ValueError(f"No row extracted for {target_table}")
print(f"{row_count} rows passed quality checks for {target_table}")
return row_count
@task
def load(row_count: int, target_table: str, primary_key: str) -> None:
print(
f"Upserting {row_count} rows into {target_table} "
f"(key: {primary_key})"
)
row_count = extract(source_table)
checked = check(row_count, target_table)
load(checked, target_table, primary_key)
ingest_dataset(
source_table="{{ params.source_table }}",
target_table="{{ params.target_table }}",
primary_key="{{ params.primary_key}}"
)Ce DAG fonctionne parfaitement tout seul et pourrait se suffire à lui-même. Mais supposons maintenant que l'on veuille effectuer l'opération sur trente datasets. On pourrait lancer le DAG manuellement trente fois avec des paramètres différents. Sinon, il suffit d'ajouter une tâche qui appelle une API retournant la liste des datasets, et instancie autant de groupes de tâches qu'il y a d'entrées fournies. Cette tâche pourrait tout aussi bien lire un fichier JSON embarqué avec le DAG, ou récupérer une configuration depuis un bucket S3 : peu importe la provenance, seul compte le fait que la liste soit résolue à l'exécution.
API_URL = os.environ.get("DATASETS_API_URL", "https://config.example.internal/datasets")
with DAG(...)
@task
def load_datasets() -> list[dict]:
"""Return the list of datasets to process."""
response = requests.get(API_URL, params={"env": ENV}, timeout=10)
response.raise_for_status()
return response.json()
@task_group
...
# Au lieu d'instancier un groupe de tâches avec les paramètres passés dans le DAG, on fournit la liste retournée par notre nouvelle tâche.
ingest_dataset.expand_kwargs(load_datasets())Enfin, afin de pouvoir surcharger le retour de l’API et ne lancer qu'une seule instance du groupe de tâches, on demande à la fonction load_datasets de retourner les entrées du DAG si elles sont fournies, le résultat de l'appel API sinon. On déplace la fonction en dehors et on crée une nouvelle tâche qui l'appelle avec les paramètres du DAG.
def load_datasets(source_table, target_table, primary_key) -> list[dict]:
if source_table and target_table and primary_key:
return [
{
"source_table": source_table,
"target_table": target_table,
"primary_key": primary_key,
}
]
response = requests.get(API_URL, params={"env": ENV}, timeout=10)
response.raise_for_status()
return response.json()
with DAG(...)
@task
def determine_datasets(params: dict) -> list[dict]:
return load_datasets(
params["source_table"],
params["target_table"],
params["primary_key"],
)
... # Les autres tâches restent les mêmes, on modifie juste le dernier chaînage pour appeler notre nouvelle tâche.
ingest_dataset.expand_kwargs(determine_datasets())Le bloc params au niveau du DAG répond à un besoin opérationnel courant : rejouer un seul dataset sans toucher à la configuration, par exemple après un incident sur une table précise. determine_datasets lit les paramètres depuis le contexte ; s’ils sont tous renseignés, elle court-circuite l’appel API et ne renvoie qu’un dataset. Le DAG ne déplie alors qu’un seul groupe de tâches. Le reste du temps, déclenché sans paramètres, il traite l’intégralité du catalogue. Même code, deux comportements sans branchement conditionnel dans la topologie. determine_datasets renvoie une XComArg – une référence vers le résultat de la tâche, pas la valeur, qui n’existera qu’à l'exécution.
L’intérêt est concentré dans la tâche determine_datasets : c’est elle qui porte la logique de la résolution de la configuration. A la manière de n’importe quelle tâche, on peut y mettre ce que l’on veut : appel à une API, lecture d’un fichier, etc. Et contrairement à une boucle for dans le code, ce travail ne sera facturé qu'à l'exécution, pas à chaque parsing.
Limitations
Le dynamic task mapping est puissant, mais il vient avec un lot de contraintes structurelles qu’il vaut mieux connaître avant de bâtir dessus.
Pas de mapping imbriqué dans un task group mappé
C'est sans doute la contrainte la plus structurante pour notre pattern : on ne peut pas faire d'expand sur une tâche à l'intérieur d'un groupe de tâches lui-même mappé. Concrètement, si votre groupe ingest traite un dataset, vous ne pouvez pas, dans ce même groupe mappé, re-déplier une tâche par partition de ce dataset. C'est un choix délibéré des mainteneurs, motivé par la complexité que cela ajouterait à l'interface utilisateur. Si vous avez besoin de deux niveaux de ramification, il faut découper en deux DAGs (avec un déclenchement parent → enfant) ou repenser le découpage.
L'ordre d'expansion n'est pas garanti
Quand plusieurs instances sont mappées, l'ordre n'est pas déterministe. Toute logique qui supposerait que l'instance 0 traite tel élément est fragile. Si vous avez besoin d'identifier les instances de façon stable, utilisez map_index_template pour leur donner un nom basé sur leur input plutôt que de vous reposer sur l'index entier par défaut.
En résumé
Tout part d'un principe simple : dès que la liste qui pilote la génération n'est pas une constante figée, elle doit être calculée dans une tâche et dépliée à l'exécution via expand / expand_kwargs, plutôt que construite au parse-time. Appliqué à un groupe de tâches, ce pattern remplace N DAGs dupliqués, ou une boucle fragile, par un seul DAG piloté par la configuration. On y gagne en maintenabilité (onboarder un dataset = une entrée de plus retournée par l’API), en sobriété pour l'orchestrateur (la liste se calcule dans une tâche, pas à chaque parsing) et en flexibilité opérationnelle (le mode override pour rejouer un dataset isolé).
Aller plus loin
Le pattern repose pour l'instant sur un endpoint unique. C'est volontairement simple, mais cela suppose que la configuration de référence couvre tous les besoins : impossible de rejouer un lot arbitraire de datasets sans le faire exister côté API. Une évolution naturelle consiste à accepter un fichier de configuration en entrée du DAG.
L'idée : ajouter un paramètre qui désigne un fichier de configuration (un chemin local, ou un objet dans un bucket), et laisser determine_datasets le résoudre à l'exécution. Un fichier explicitement fourni l'emporte, sinon on retombe sur l'API. Le bloc params devient alors le point d'entrée d'une configuration complète plutôt que des trois colonnes décrivant un dataset isolé. La mécanique derrière ne bouge pas : expand_kwargs déplie toujours le groupe une fois par entrée, seule la provenance de la liste change.
L'intérêt est double. D'abord, on peut rejouer une configuration arbitraire sans toucher ni au dépôt ni à l'API : pratique pour un backfill ponctuel sur un sous-ensemble de datasets, ou pour valider une nouvelle configuration avant de la publier. Ensuite, on obtient une cascade à trois niveaux : les trois params pour un dataset isolé, un fichier pour un lot ad hoc, l'API pour le nominal. Exactement la même logique que le mode override, appliquée à chaque étage.
Code final
"""
Dynamic, config-driven ingestion DAG (Airflow 3).
Resolve datasets from params, then an explicit config file, then the API – this DAG builds one task group running a small
extract -> check -> load pipeline. The number of task groups is decided at
runtime from the configuration, so onboarding a new dataset is a config change,
not a code change.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
import pendulum
import requests
from airflow.sdk import DAG, task, task_group
ENV = os.environ.get("DATA_ENV", "dev")
API_URL = os.environ.get("DATASETS_API_URL", "https://config.example.internal/datasets")
def load_datasets(source_table, target_table, primary_key, config_file) -> list[dict]:
"""Return the list of datasets to process.
Resolution order: ad-hoc dataset from params, then an explicitly provided
config file, then the datasets API as the default source.
"""
if source_table and target_table and primary_key:
return [
{
"source_table": source_table,
"target_table": target_table,
"primary_key": primary_key,
}
]
if config_file:
content = Path(config_file).read_text().replace("{{ENV}}", ENV)
return json.loads(content)
response = requests.get(API_URL, params={"env": ENV}, timeout=10)
response.raise_for_status()
return response.json()
with DAG(
dag_id="dynamic_dataset_ingestion",
schedule=None,
start_date=pendulum.datetime(2025, 1, 1, tz="Europe/Paris"),
catchup=False,
tags=["blog", "dynamic-dag"],
params={
"source_table": None,
"target_table": None,
"primary_key": None,
“config_file”: None
},
) as dag:
@task
def determine_datasets(params: dict) -> list[dict]:
"""Resolve which datasets to process from params or the JSON config."""
return load_datasets(
params["source_table"],
params["target_table"],
params["primary_key"],
params[“config_file”]
)
@task_group(group_id="ingest")
def ingest_dataset(source_table: str, target_table: str, primary_key: str):
@task
def extract(source_table: str) -> int:
print(f"Extracting rows from {source_table}")
# Real extraction logic goes here. Returns a row count for the demo.
return 42
@task
def check(row_count: int, target_table: str) -> int:
if row_count == 0:
raise ValueError(f"No row extracted for {target_table}")
print(f"{row_count} rows passed quality checks for {target_table}")
return row_count
@task
def load(row_count: int, target_table: str, primary_key: str) -> None:
print(
f"Upserting {row_count} rows into {target_table} "
f"(key: {primary_key})"
)
row_count = extract(source_table)
checked = check(row_count, target_table)
load(checked, target_table, primary_key)
# One task group instance is fanned out at runtime per dataset returned by determine_datasets(). Each dict's keys map to the group's arguments.
ingest_dataset.expand_kwargs(determine_datasets())Sources
- Dynamic Task Mapping — Documentation officielle Apache Airflow : référence complète sur expand, expand_kwargs, partial, le mapping de groupes de tâches et les limitations (max_map_length, map_index_template).
- Best Practices — Documentation officielle Apache Airflow : section « Top level Python Code » sur les dangers du code exécuté au parsing (appels réseau, accès base).
- Create dynamic Airflow tasks — Astronomer : guide pratique avec exemples, dont le rendu des tâches mappées dans l'interface.
- DAG writing best practices in Apache Airflow — Astronomer : approfondissement sur le top-level code et son impact sur le parsing.
