Airflow Dynamic DAGs: One Ticker Shouldn't Need a PR

The tell-tale smell of an Airflow setup that's outgrown itself: adding a new stock ticker to the pipeline requires a code review, a deploy, and a prayer that nothing else broke.

That's what happens when DAGs are written by copy-paste. The fix is to stop writing DAGs per entity and start generating them from config — or better, map tasks over the entities at runtime.

Here are both mechanisms, when to use which, and the validation discipline that keeps generated DAGs from taking the whole scheduler down.

1. Two Kinds of Dynamic

People say "dynamic DAGs" and mean one of two completely different mechanisms.

Generated DAGs: a Python factory reads config — a file, a DB table, an API, or environment variables — and emits one DAG object per entry, every time the scheduler parses the module. The grid view shows N separate DAGs.

Mapped tasks: a single DAG whose task fans out at run time. The DAG file stays static; expand() turns one task into N task instances, each with its own map_index. The grid view shows one DAG with N task instances.

Both remove the copy-paste pattern. They differ in where the fan-out happens: at parse time vs at run time.

2. Config-Driven DAG Factories

The factory pattern: one module that builds DAGs from a config source. Here's the ticker version — 30 DAG files replaced by one function.

The factory reads the config, validates every row, and returns one DAG per entry. The scheduler calls the module on every parse loop, so a new ticker appears within one parse cycle — no deploy, no PR.

The critical discipline: the factory must be deterministic and fast. No API calls at module scope, no database reads. Config resolution that takes 10 seconds turns your whole scheduler into a crawl.

That rule has a name in the official docs: never read Airflow Variables at import time. Variable.get() opens a connection to the metadata DB on every parse loop, so DB-sourced config should be exported to a file in the DAG folder first. Two registration details: @dag-decorated factories auto-register each generated DAG (set auto_register=False to opt out), while plain DAG() objects need the globals() assignment shown below. A third flavor worth knowing: generate DAG files at CI time from a template plus JSON configs, so each DAG ships as a normal, self-contained file with zero factory code in the scheduler.

import json
from datetime import datetime
from pathlib import Path
from airflow import DAG
from airflow.decorators import task
from airflow.utils.dates import days_ago

CONFIG_PATH = Path(__file__).parent / "config" / "tickers.json"


def load_tickers():
    # Config is cheap to read; never fetch it from an API at parse time
    return json.loads(CONFIG_PATH.read_text())["tickers"]


def build_ticker_dags():
    dags = []
    for row in load_tickers():
        # Validate: a missing key fails here, at parse time, with a clear error
        ticker = row["symbol"]
        table = row["target_table"]

        dag = DAG(
            dag_id=f"ticker_{ticker.lower()}",
            start_date=datetime(2026, 1, 1),
            schedule="@daily",
            catchup=False,
        )

        @task(dag=dag)
        def fetch_and_load():
            print(f"Loading {ticker} into {table}")

        fetch_and_load()
        dags.append(dag)
    return dags


for dag in build_ticker_dags():
    globals()[dag.dag_id] = dag

3. Task Mapping with expand()

When every ticker runs on the same schedule and does the same work, you don't need N DAGs — you need one DAG whose task runs N times.

expand() takes the input lists and creates one task instance per row. Each instance is a first-class citizen: its own logs, its own retry state, its own entry in the grid view. The map_index distinguishes them.

Mapping shines when the fan-out is per run — a list of files that landed, a batch of IDs, a set of partitions. Dynamic task mapping has existed since Airflow 2.3, and mapped inputs can come from literal lists or from an upstream task's output.

from datetime import datetime
from airflow import DAG
from airflow.decorators import task


@task
# One instance per ticker; the input row arrives in each instance's kwargs
def load_ticker(symbol: str, target_table: str):
    print(f"Loading {symbol} into {target_table}")


with DAG(
    dag_id="market_load_daily",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    load_ticker.expand(
        symbol=["AMZN", "AAPL", "GOOG", "MSFT"],
        target_table=["amzn", "aapl", "goog", "msft"],
    )

4. map_index and Downstream Joins

Every mapped task instance carries a map_index — the position of its input in the expanded lists. It shows in the grid view, in the logs, and in the task metadata.

Downstream tasks that receive the results of a mapped task get the full mapped output — a list of N results, one per instance. If a downstream task needs to process each result, it should itself be mapped over the upstream output. Airflow preserves the correlation: downstream instance i always consumes upstream instance i's output.

The gotcha: mixing mapped and unmapped downstream tasks. An unmapped task reading a mapped output gets a list; a mapped task reading a mapped output gets element-wise pairing.

from datetime import datetime
from airflow import DAG
from airflow.decorators import task


@task
# Each instance gets one ticker's response

def fetch_ticker(symbol: str):
    return {"symbol": symbol, "rows_loaded": len(symbol) * 1000}


@task
# Mapped over the fetch output: instance i pairs with fetch instance i

def summarize(load_result: dict):
    print(f"{load_result['symbol']}: {load_result['rows_loaded']} rows")


with DAG(
    dag_id="market_summary",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    results = fetch_ticker.expand(symbol=["AMZN", "AAPL", "GOOG"])
    summarize.expand(load_result=results)

5. Validating and Testing Generated DAGs

Generated DAGs change the failure surface. A single bad config row now threatens the whole fleet — which is why validation and testing are non-negotiable.

Three gates: (1) config schema validation before generation — a required-key check fails the module loudly; (2) a unit test that loads the factory module and asserts the right DAG count; (3) a CI step that parses every generated DAG with the production DagBag and fails the build on any import error.

The test is simple and catches the real classes of failure: wrong task count, missing fields, generated DAGs that don't import.

import json
import pytest
from dags.ticker_factory import build_ticker_dags


def test_factory_generates_one_dag_per_ticker(tmp_path):
    (tmp_path / "tickers.json").write_text(
        json.dumps({"tickers": [{"symbol": "AMZN", "target_table": "amzn"}]})
    )
    dags = build_ticker_dags()
    assert len(dags) == 1
    assert dags[0].dag_id == "ticker_amzn"


def test_factory_rejects_missing_keys(tmp_path):
    (tmp_path / "tickers.json").write_text(json.dumps({"tickers": [{"symbol": "AMZN"}]}))
    with pytest.raises(KeyError):
        build_ticker_dags()

6. The Anti-Patterns

Dynamic doesn't mean unmanageable. Three patterns turn dynamic DAGs into production incidents.

Anti-pattern one: slow generation. API calls or DB reads in the factory run on every scheduler parse. At 100 generated DAGs with a 2-second fetch each, the scheduler spends 200 seconds per loop — parsing everything else crawls.

Anti-pattern two: generation at DAG-run time. Building DAGs inside a task re-creates objects Airflow has already scheduled. If you need runtime variance, that's what mapping is for.

Anti-pattern three: unbounded generated DAG counts. A config table that grows to 5,000 tickers generates 5,000 DAGs — the UI becomes unusable and parsing saturates. At that scale, mapping is the answer, not generation.