Airflow DAG Authoring: Dynamic DAGs and Task Mapping

Two kinds of dynamic

Dynamic DAG generation produces many DAGs from configuration - a loop over a YAML file calling a factory function, each iteration returning a DAG with a unique dag_id. Dynamic task mapping produces a variable number of task instances at runtime from .expand() or .partial().

Mapping syntax: extract.expand(filename=['a.csv','b.csv']) creates one mapped instance per value. op.partial(static_arg=1).expand(dynamic_arg=xcom_value) fixes some arguments and maps others. Mapped tasks expose {{ ti.map_index }} and can be aggregated downstream; the downstream task receives a list of XComs when pulling from a mapped task.

Traps candidates hit

• Generating DAGs with globals()[name] = dag works but pollutes the module namespace and can collide; prefer returning DAGs from a function and registering them.
• Mapping over an XCom requires the producer to run first; the number of mapped instances is unknown at parse time.
• expand cannot be combined with positional args for the same parameter; use partial for fixed values.
• max_mapping_length caps the number of mapped instances; exceeding it fails the task.
• Dynamic DAG generation at parse time multiplies scheduler work; keep loops cheap and cache file reads.
• Mapped tasks and trigger rules interact: an empty mapping list skips downstream unless handled.

Rule of thumb

If the set of DAGs is known from config, generate DAGs. If the number of task instances depends on runtime data, use .expand().

Worked example

process = PythonOperator.partial(task_id='process', op_kwargs={'mode':'fast'}).expand(op_args=files) where files comes from an upstream XCom list. Each file becomes one mapped instance, and process.output aggregates results for the next task.