How can I dynamically generate Airflow DAG tasks and dependencies using dag_run.conf?

Below is one accepted solution that uses Airflow’s dynamic task mapping (introduced in Airflow 2.3) to “defer” the creation of tasks until runtime (when dag_run.conf is available). The key idea is that since a DAG’s structure is parsed long before any dag_run is triggered, you can’t build your tasks “on the fly” in the DAG file itself. Instead you first create a task that extracts the desired configuration (in this case, your elements list) and then “map” downstream tasks over that list. (If you have nested groups, you may need to flatten or use multiple mappings.)

Below is a working example:

---

PYTHONCopied
from airflow import DAG
from airflow.decorators import task
from airflow.utils.context import get_current_context
from datetime import datetime

# Task to extract configuration from dag_run.conf
@task
def get_elements():
    """
    Returns a list of elements from the dag_run conf.
    Expected config, for example, is:
    {
        "elements": [["a", "b"], ["c", "d"], ... ]
    }
    If your structure is nested and you want to process individual elements,
    you may need to flatten it here.
    """
    context = get_current_context()
    dag_run = context.get("dag_run")
    if not dag_run or not dag_run.conf:
        raise ValueError("dag_run.conf is missing, make sure to trigger the DAG with a conf.")

    elements = dag_run.conf.get("elements")
    if not elements:
        raise ValueError("No 'elements' key found in dag_run.conf.")

    # For demonstration, let’s flatten the nested list.
    # If your configuration is already flat, remove or modify this flattening.
    flat_elements = [item for group in elements for item in group]
    return flat_elements

# Task to process an individual element
@task
def process_element(element: str) -> str:
    print(f"Processing element: {element}")
    return f"Processed {element}"

with DAG(
    dag_id="process_elements_dag",
    description="Process elements using dynamic task mapping based on dag_run.conf",
    schedule_interval=None,
    start_date=datetime(2025, 1, 28),
    catchup=False,
) as dag:
    
    # Get the list of individual elements from the conf.
    elements = get_elements()

    # Dynamically generate tasks – one per element.
    processed_results = process_element.expand(element=elements)

---

How It Works

1. Configuration Extraction (@task get_elements):
   This task uses Airflow’s runtime context (via get_current_context) to access the dag_run.conf. It then extracts the list of elements from the provided configuration. (In the sample, the nested list is flattened; adjust this as needed for your use case.)

2. Dynamic Task Mapping (process_element.expand):
   Rather than hardcoding a loop over a static list, the task mapping (the .expand call) “maps” the process_element task over the list of elements returned by get_elements(). This happens at runtime, so the DAG structure adapts to the provided configuration.

3. DAG Structure vs. Runtime:
   Because the DAG is defined and parsed before runtime, you cannot build the tasks “directly” from dag_run.conf. Dynamic task mapping (and the get_elements task) provides the canonical solution for deferring task creation until the run is triggered.

How to Trigger the DAG

When triggering the DAG (for example, via the Airflow UI or the CLI), pass the configuration JSON so that the get_elements task can read it. For instance, using the UI, click “Trigger DAG” and add:

JSONCopied
{
    "elements": [
        [["a", "b"], ["c", "d"], ["e", "f"], ["g", "h"], ["i", "j"]],
        [["k", "l"], ["m", "n"], ["o", "p"], ["q", "r"]],
        [["s", "t"], ["u", "v"], ["w", "x"]],
        [["y", "z"]]
    ]
}

In this example the get_elements task flattens the list to process individual elements. If you need to preserve the structure (for setting up dependencies between groups), you may need to design a two‐step mapping process or perform the grouping inside a single task that triggers mapped tasks sequentially.

This solution (using dynamic task mapping) is from reputable sources and has been successfully used in production when a DAG must vary its behavior based on dag_run.conf.