Airflow Dynamic DAGs

Dynamic DAGs generate tasks or entire DAG structures automatically from data — a list of files, a database query, or a configuration file. Instead of writing 50 tasks by hand, you write a loop that creates them for you.

Why Dynamic DAGs?

Scenario: You process monthly sales reports for 12 regions.

Manual approach (bad): Write 12 separate task definitions
Dynamic approach (good): Write a loop that creates 12 tasks automatically

Static (manual):
  task_region_north = PythonOperator(task_id="process_north", ...)
  task_region_south = PythonOperator(task_id="process_south", ...)
  task_region_east  = PythonOperator(task_id="process_east", ...)
  ... (12 total)

Dynamic (loop):
  regions = ["north", "south", "east", "west", ...]
  for region in regions:
      PythonOperator(task_id=f"process_{region}", ...)

Pattern 1: Dynamic Tasks From a List

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

regions = ["north", "south", "east", "west", "central"]

def process_region(region_name):
    print(f"Processing sales for region: {region_name}")

with DAG("regional_sales", start_date=datetime(2024, 1, 1),
         schedule="@monthly", catchup=False) as dag:

    tasks = []
    for region in regions:
        task = PythonOperator(
            task_id=f"process_{region}",
            python_callable=process_region,
            op_kwargs={"region_name": region},
        )
        tasks.append(task)

    # All regional tasks run in parallel
    # (no dependencies between them — each is independent)

DAG Graph:
[process_north]
[process_south]   ← all run at the same time
[process_east]
[process_west]
[process_central]

Pattern 2: Dynamic Tasks With a Shared Finish Task

from airflow.operators.empty import EmptyOperator

with DAG("regional_with_summary", start_date=datetime(2024, 1, 1),
         schedule="@monthly", catchup=False) as dag:

    start   = EmptyOperator(task_id="start")
    summary = EmptyOperator(task_id="generate_summary")

    for region in regions:
        task = PythonOperator(
            task_id=f"process_{region}",
            python_callable=process_region,
            op_kwargs={"region_name": region},
        )
        start >> task >> summary

DAG Graph:
              [start]
             /    |   \   \   \
[process_north] [process_south] [process_east] [process_west] [process_central]
             \   |   /   /   /
          [generate_summary]

Pattern 3: Dynamic Tasks From a Config File

Reading task list from a JSON file makes the DAG configurable without code changes:

# config/pipelines.json
[
  {"name": "inventory",  "table": "inv_data",  "threshold": 100},
  {"name": "orders",     "table": "ord_data",  "threshold": 500},
  {"name": "customers",  "table": "cust_data", "threshold": 50}
]

import json
from pathlib import Path

config_path = Path(__file__).parent / "config" / "pipelines.json"
pipelines = json.loads(config_path.read_text())

def validate_pipeline(name, table, threshold):
    print(f"Checking {table}: threshold {threshold}")

with DAG("config_driven_dag", start_date=datetime(2024, 1, 1),
         schedule="@daily", catchup=False) as dag:

    for pipe in pipelines:
        PythonOperator(
            task_id=f"validate_{pipe['name']}",
            python_callable=validate_pipeline,
            op_kwargs=pipe,
        )

Update the JSON file to add or remove pipelines. The DAG generates the matching tasks automatically on the next parse.

Pattern 4: Dynamic DAGs From a Single DAG Factory File

You can generate multiple complete DAGs from one Python file:

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

clients = [
    {"id": "acme",    "schedule": "@daily"},
    {"id": "globex",  "schedule": "@hourly"},
    {"id": "initech", "schedule": "@weekly"},
]

def make_etl_dag(client_id, schedule):
    def run_etl():
        print(f"Running ETL for client: {client_id}")

    with DAG(
        dag_id=f"etl_{client_id}",
        start_date=datetime(2024, 1, 1),
        schedule=schedule,
        catchup=False,
        tags=["client", client_id],
    ) as dag:
        PythonOperator(task_id="run_etl", python_callable=run_etl)
    return dag

# Generate one DAG per client — all registered in Airflow
for client in clients:
    globals()[f"etl_{client['id']}"] = make_etl_dag(
        client["id"], client["schedule"]
    )

Result: Three separate DAGs appear in the Airflow UI:
  etl_acme     → runs @daily
  etl_globex   → runs @hourly
  etl_initech  → runs @weekly

The globals() assignment registers each DAG in the module namespace so Airflow's parser discovers it.

Dynamic Task Mapping (Airflow 2.3+)

Dynamic Task Mapping is the modern, built-in way to create dynamic tasks. You map an operator over a list — Airflow creates one task instance per item at runtime, not at parse time.

from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2024, 1, 1), schedule="@daily", catchup=False)
def mapped_pipeline():

    @task
    def get_files():
        # Returns a list determined at runtime (from S3, DB, API, etc.)
        return ["jan_sales.csv", "feb_sales.csv", "mar_sales.csv"]

    @task
    def process_file(filename: str):
        print(f"Processing: {filename}")

    files = get_files()
    process_file.expand(filename=files)  # ← creates one task per file

mapped_pipeline()

At runtime, if get_files() returns 3 items:
  [get_files]
       |
   ┌───┴───┐
   ↓       ↓       ↓
[process jan_sales.csv]
[process feb_sales.csv]
[process mar_sales.csv]

If it returns 10 items next month, 10 tasks are created automatically.

Key Rules for Dynamic DAGs