DAGs Built from Config

Concepts covered: DAG parsing

14 loads that differ in the details

Mapping fits when the copies share a run and an owner and can sit on one row in the grid, which describes the partner feeds. The warehouse team's loads do not fit. The team extracts and then loads 14 source tables the same way, but the payments team owns 2 of them, the search team owns 5, each table has its own history to inspect, and when a source breaks the owning team wants to pause that 1 load and leave the other 13 running. That calls for 14 DAGs, each with its own dag_id, and nobody wants to maintain 14 near-identical files. The answer is 1 Python file that reads a config and produces a DAG object per entry.

- name: orders
  source: orders_db
  owner: payments
- name: listings
  source: listings_db
  owner: search

from pathlib import Path

import yaml
from airflow.sdk import DAG, task

CONFIG = Path(__file__).parent / "tables.yaml"


def build(entry):
    with DAG(dag_id=f"load_{entry['name']}", tags=[entry["owner"]]) as d:

        @task
        def extract():
            return f"SELECT * FROM {entry['source']}.{entry['name']}"

        @task
        def load(sql):
            print(sql)

        load(extract())
    return d


for entry in yaml.safe_load(CONFIG.read_text()):
    globals()[f"load_{entry['name']}"] = build(entry)

The processor imports this file like any other. The loop runs, build returns a DAG for each entry, and globals() places each under its own name at the top level of the module, which is where the processor collects DAG objects from. airflow dags list then shows load_orders and load_listings as 2 rows with the same file location. Each has its own pause switch, its own runs and its own grid; the shared file is invisible to anyone reading the UI. The entry dict is captured by the inner functions, so extract in load_orders builds its SQL from orders_db and nothing else, and adding a 3rd entry to the YAML produces a 3rd DAG on the next parse without touching the Python.

The price shows up in the parse loop

The intermediate lesson showed that the DAG processor re-imports every file on a cycle, 30 seconds apart by default, and that a file which raises on import is recorded as an import error. Config-driven generation multiplies both facts. The YAML read happens on every cycle, which costs nothing measurable for a local file of a few kilobytes. Replace it with a query against the warehouse's catalog and the parse loop now carries a network round trip on every cycle, the per-file timeout of 50 seconds becomes reachable on a slow night, and 1 timeout counts as a failed import of the whole file: none of the 14 DAGs takes a new definition until a parse of that file finishes.

An error has the same blast radius. One entry with a missing source key raises KeyError inside build. That fails the import, and the import errors page lists the file once. The 14 DAGs already registered stay in the list on their last good definitions and stop updating there: no fix to any of them lands, an entry added in the same edit never appears, and a worker that loads the file to run one of their tasks hits the same error. Validating the config before the loop doesn't unfreeze the other 13, but it turns a KeyError on a line inside build into a message that names the broken entry, which is the difference between a 1-minute fix and a 20-minute hunt.

REQUIRED = {"name", "source", "owner"}

entries = yaml.safe_load(CONFIG.read_text())
for entry in entries:
    missing = REQUIRED - entry.keys()
    if missing:
        raise ValueError(f"tables.yaml entry {entry!r} lacks {sorted(missing)}")

for entry in entries:
    globals()[f"load_{entry['name']}"] = build(entry)

The subtler cost is identity. A DAG's history lives under its dag_id. Change the naming scheme from load_{name} to warehouse_{name} and the processor registers 14 new DAGs with empty histories, while the 14 old ones are marked as no longer present in any file and keep their runs under the old names. Treat the id formula as a contract, and keep the config file in the same repository as the DAG file so that a change to one is reviewed beside the other.

One DAG, mapped tasks

• Copies share 1 run and 1 history
• Count decided inside the run, by data
• Pausing stops every copy
• 1 row in the DAG list

One file, many DAGs

• Each DAG has its own history and pause switch
• Count decided at parse time, by config
• A broken config entry freezes every DAG in the file on its last good definition
• 14 rows, 1 file location

About this lesson

This section is part of the DAG Anatomy: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.