Airflow Dynamic DAGs: One Config Powers Every Ticker

A new ticker meant a new file, a review, and a deploy. Twelve tickers meant twelve copies of the same bug.

You'll replace copies with a factory. One config lists tickers, one function builds DAGs, mapping fans out tasks.

We cover generated DAGs versus mapped tasks, validation, and the anti-patterns that turn factories into fog. Adding a ticker becomes a config edit.

Config changes. Code rests.

Two Kinds of Dynamic: Generated DAGs and Mapped Tasks

Generated DAGs vary DAG count from config: one ticker becomes one DAG. Mapped tasks vary task count inside one DAG via expand.

Use factories when owners or schedules differ per entity. Use mapping when one DAG handles many homogeneous items like tickers in one batch.

Config-Driven DAG Factories: The Ticker Pattern

A YAML list of tickers feeds a loop that builds one @dag per entry with derived table names and tags. No per-ticker files remain.

Validate slugs, enforce unique dag_ids, and keep the factory file side-effect free. Adding NVDA is one YAML line plus CI green.

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

with open("dags/tickers.yaml") as f:
    TICKERS = yaml.safe_load(f)["tickers"]

def make_ticker_dag(symbol: str):
    @dag(dag_id=f"ticker_{symbol.lower()}_daily", schedule="@daily",
         start_date=datetime(2026, 1, 1), catchup=False, tags=["equities"])
    def ticker_pipeline():
        @task
        def fetch():
            return {"symbol": symbol}

        @task
        def store(payload: dict):
            return {"stored": payload["symbol"]}

        store(fetch())
    return ticker_pipeline()

for _symbol in TICKERS:
    globals()[f"ticker_{_symbol.lower()}_daily"] = make_ticker_dag(_symbol)

Task Mapping With Expand

expand() fans one task over a ticker list inside a single DAG. One batch DAG handles 50 tickers with per-ticker map indexes.

Failed tickers retry alone while siblings succeed. The batch stays green-ish with clear per-ticker lineage instead of all-or-nothing.

Split args with partial for constants and expand for mapped ones: add.partial(y=10).expand(x=[1,2,3]) fans three tasks sharing y. Classic operators use the same verbs as classmethods, but task_id, pool, queue, and most BaseOperator args can't map and must ride in partial. expand takes keyword args only, and cross-product is the default: two mapped kwargs multiply instances. Reach for expand_kwargs with a list of dicts for explicit sets, or zip / XComArg.zip with fillvalue to pair iterables positionally. Combine upstream lists with .concat, reshape them pre-map with .map (raising AirflowSkipException inside skips just that instance), and map whole @task_group bodies when a multi-step unit repeats.

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

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["equities"])
def ticker_batch():
    @task
    def load_ticker(symbol: str):
        return {"symbol": symbol, "rows": 1500}

    @task(trigger_rule="none_failed")
    def join_loads(results: list):
        return {"tickers": len(results)}

    symbols = ["AAPL", "NVDA", "MSFT", "TSLA", "AMZN"]
    join_loads(load_ticker.expand(symbol=symbols))

ticker_batch()
# add.partial(y=10).expand(x=[1, 2, 3])  # constants in partial, mapped in expand
# expand_kwargs([{...}, {...}]) for explicit sets; .zip(fillvalue=...) pairs XComArgs
# max_active_tis_per_dagrun=8 throttles the fan-out; max_map_length caps at 1024

Map Index and Downstream Joins

Each mapped task carries a map_index identifying its ticker. Downstream joins aggregate the list of results with none_failed rules.

Inspect per-ticker logs via map_index in Grid view. Joins must tolerate partial skips when one ticker has no data for the interval.

Cap fan-out deliberately: max_map_length (1024 default) bounds instances per task, while max_active_tis_per_dagrun and max_active_tis_per_dag throttle how many run at once. An upstream returning [] maps zero instances: the task skips and downstream follows trigger rules, which by default skips too, so set the join rule explicitly. Read mapped XComs by index with ti.xcom_pull(task_ids=[...], map_indexes=[2]), and never set TriggerRule.ALWAYS on task-generated mapping; the scheduler rejects it at parse time. Repeated mapping chains mapped outputs into the next expand, and Grid view groups instances under task_id[n] with per-index logs.

Validation and Testing of Generated DAGs

Test factories with DagBag: assert every ticker yields a DAG, ids are unique, and each DAG has no import error.

Run airflow dags test on two representative tickers in CI. Config schema validation with pydantic catches bad slugs before the scheduler does.

from airflow.models import DagBag

def test_factory_builds_all_tickers():
    bag = DagBag(dag_folder="dags/", include_examples=False)
    assert not bag.import_errors, bag.import_errors
    ids = [d for d in bag.dags if d.startswith("ticker_")]
    assert len(ids) >= 5
    assert len(ids) == len(set(ids))

def test_ticker_dag_shape():
    bag = DagBag(dag_folder="dags/", include_examples=False)
    dag = bag.get_dag("ticker_nvda_daily")
    assert dag is not None
    assert len(dag.tasks) == 2

The Generated-DAG Anti-Patterns

Do not generate DAGs from live API calls at parse time; the scheduler will hammer the API every 30 seconds. Snapshot the list to a file.

Do not branch per ticker inside factories with different SLAs crammed together. Split factories by owner when schedules diverge.

A third trap: divergent SLAs inside one mapped batch. Mapping assumes homogeneous work; when tickers need different owners or deadlines, the factory pattern with separate DAGs wins back. Validate that too: assert per-entity schedule and owner fields in config tests before the scheduler ever sees them.