Dynamic DAGs & Dynamic Task Mapping

Two different techniques are both called dynamic, and people mix them up:

So ask one question first: when do you know how many? If each item needs its own schedule, generate DAGs. If the count only appears during the run, map tasks.

If it is always three things, write three tasks. They are easier to read and to search for.

Suppose each shop table needs its own copy DAG, with its own schedule. A factory is a function that builds one DAG. You call it once for each table:

TABLES = ["orders", "customers"]

def build(table):
    with DAG(
        dag_id=f"copy_{table}",
        schedule="@daily",
        start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
        catchup=False,
        tags=["generated"],
    ) as d:
        EmptyOperator(task_id="copy")
    return d

for table in sorted(TABLES):
    build(table)

This file makes two DAGs, copy_orders and copy_customers. Airflow registers a DAG made with "with DAG(...)" or with @dag by itself.

A plain DAG(...) object, made without "with", was not found in our test. Old code adds it to globals() for that reason.

The factory runs on every parse. So keep the config in a file next to the DAG, not in a database or an API. A slow source makes every parse slow, and if it fails, all the DAGs disappear.

Mapping makes one task instance for each item in a list. Each instance has its own log, state and retries.

Compare one task with a loop inside. If file 2 of 3 fails, the whole task fails, and a retry loads all three again. With mapping, only that instance fails and retries.

expand() takes the values that change. partial() takes the values that stay the same:

@task
def find_files() -> list[str]:
    return ["orders_1.csv", "orders_2.csv", "orders_3.csv"]

@task
def load_file(path: str, target: str) -> int:
    print(f"loading {path} into {target}")
    return 2

counts = load_file.partial(target="raw_orders").expand(path=find_files())

Give expand() more than one argument, and you get every combination:

expand() takes keyword arguments only. f.expand([1, 2]) fails with a TypeError when the file is parsed.

Mapping is most useful when an upstream task makes the list. Nobody knows how many files will arrive until find_files runs.

If the list is fixed in the code, you could write a loop that makes normal tasks. That is often clearer.

A task after the mapped task can collect all the results. This is called the reduce step:

@task
def total(counts):
    print(f"counts type: {type(counts).__name__}")
    print(f"total rows: {sum(counts)}")

total(counts)

Before the upstream task runs, the Graph view shows a mapped task as one box. The instances appear when the list exists.

A list can have at most 1024 items by default. The setting is max_map_length in the core section.

In our test, a task returned 1025 items. That task failed, not the mapped task, which became upstream_failed. Its log said only: Server returned error.

So if a list can grow, check its length in the task that makes it, and fail with a clear message.

Templates need care too. A literal list was rendered, but values from an upstream task were not:

The second task still succeeded, with the braces in its output. So fill in dates in the task that makes the list.

In the Grid view, a mapped task shows its instances by number: 0, 1, 2 and so on. Numbers say little, so give each instance a name with map_index_template:

@task(map_index_template="{{ file_name }}")
def load_file(path: str):
    context = get_current_context()
    context["file_name"] = path
    print(f"loading {path}")

Now a failed instance shows its file name. You can open its log, and clear only that instance. The others keep their success.

Generated, mapped, or neither:

1. Known when the file is parsed, with its own schedule: generate DAGs.
2. Known only during the run: map tasks with expand().
3. Always the same few things: write them out.
4. Two lists in expand() give every combination. expand_kwargs gives pairs.
5. Check the list length, and fill in templates, before mapping.

Check yourself: expand(table=["orders", "customers"], region=["eu", "us"]). How many instances? (Answer: 4.)