Orchestrating with Airflow

Apache Airflow is an orchestrator: it holds a pipeline as a graph of tasks, runs them in dependency order on a schedule, retries the ones that fail, and keeps a record of every run. The retraining loop needs exactly that, and a cron line does not provide it.

Last updated: 19 Oct, 2026 · MLOps

Retraining strategies left the retrain needing something to start it every week. This lesson gives it one, as a small DAG over the two stages the training pipeline already has.

The clip's DAG extracts weather data, transforms it and loads it into Postgres; this DAG has two tasks that run the retraining pipeline, and the graph, the task and the dependency between them mean the same thing.

Reading a DAG as a graph of tasks

The video builds an extract, transform and load pipeline and defines the shape first. A DAG is a directed acyclic graph: directed because each arrow points from one task to the next, acyclic because the arrows never come back around to where they started. Each node is a task, one unit of work, and an arrow from one task to another says the second runs only after the first has succeeded. The pipeline here is the same shape with two nodes: pull the data, then fit the model.

Declaring the DAG and its schedule

The with DAG(...) block names the pipeline and says when it runs. catchup=False matters more than it looks: without it, a DAG switched on today with a start date in January would immediately queue every weekly run it missed, and a backlog of stale retrains helps nobody.

from datetime import datetime, timedelta

from airflow.sdk import DAG, task

# a missed week is skipped rather than queued up, so no stale retrain runs late
with DAG(
    dag_id="sp_retrain",
    start_date=datetime(2026, 1, 1),
    schedule=timedelta(days=7),
    catchup=False,
    default_args={"owner": "mlops", "retries": 1},
) as dag:
    ...   # the tasks are defined inside this block

Writing the two tasks

A function under @task becomes a task. The imports sit inside the functions rather than at the top of the file, because the scheduler parses this file constantly to find DAGs and a heavy import at module level slows every parse down.

    @task
    def ingest() -> int:
        # imported inside the task: the scheduler parses this file often and should stay light
        from sp.components.data_ingestion import DataIngestion
        from sp.config.configuration import ConfigurationManager

        cfg = ConfigurationManager()
        ingestion = DataIngestion(cfg.get_data_ingestion_config())
        frame = ingestion.download()
        ingestion.split(frame)
        return len(frame)

    @task
    def train(rows: int) -> dict:
        from sp.pipeline.train_pipeline import main

        return main()

    # calling one task with the other's result is what creates the dependency
    train(ingest())

That last line is the whole wiring. Passing ingest() into train() does not run anything; it tells Airflow that train depends on ingest, and the row count travels between them as the task's return value.

The file the scheduler reads

from datetime import datetime, timedelta

from airflow.sdk import DAG, task

# a missed week is skipped rather than queued up, so no stale retrain runs late
with DAG(
    dag_id="sp_retrain",
    start_date=datetime(2026, 1, 1),
    schedule=timedelta(days=7),
    catchup=False,
    default_args={"owner": "mlops", "retries": 1},
) as dag:

    @task
    def ingest() -> int:
        # imported inside the task: the scheduler parses this file often and should stay light
        from sp.components.data_ingestion import DataIngestion
        from sp.config.configuration import ConfigurationManager

        cfg = ConfigurationManager()
        ingestion = DataIngestion(cfg.get_data_ingestion_config())
        frame = ingestion.download()
        ingestion.split(frame)
        return len(frame)

    @task
    def train(rows: int) -> dict:
        from sp.pipeline.train_pipeline import main

        return main()

    train(ingest())

Parsing the DAG before trusting it

The first test of a DAG file is whether it imports at all. A typo or a missing package here shows up in the Airflow interface as a broken DAG with no tasks, which is a slow way to find a one-line error. Importing the file and reading the DAG object back is quicker, and it needs no scheduler.

$ cd dags
$ python -c "import sp_retrain as m; d = m.dag; print(d.dag_id, '|', d.schedule, '|', sorted(t.task_id for t in d.tasks), '|', d.catchup)"
sp_retrain | 7 days, 0:00:00 | ['ingest', 'train'] | False

The schedule comes back as a seven-day interval, both tasks are there, and catchup is off. The dependency is readable the same way: ingest has train downstream of it, and train has nothing downstream.

Running it on a real schedule

Parsing proves the file is valid. Running it on a schedule needs more than a Python interpreter: Airflow keeps the state of every run in a metadata database, and a scheduler process has to be alive to create those runs. The quick start documents three commands for a local install, and airflow standalone rolls them into one.

# from the Airflow quick start, shown as the docs give it, not run here
$ airflow db migrate
$ airflow api-server --port 8080
$ airflow scheduler

# or all of it in one process, for a local trial
$ airflow standalone

Those are the commands the Airflow 3 quick start gives at airflow.apache.org/docs/apache-airflow/stable/start.html, and the generated admin password lands in $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated. A one-off run needs less than that: airflow dags test sp_retrain executes a single DagRun in the foreground against the metadata database, with no scheduler process involved, which the documentation describes as a simulated execution flow similar to a LocalExecutor session.

Airflow 2 against Airflow 3

What an orchestrator buys over a cron line

• Dependencies are declared, not implied. The train task cannot start before ingest finishes, and that rule lives in one place rather than in the order of two cron entries.
• Retries are per task. A failed download retries on its own without refitting the model, which is what retries: 1 in the default arguments sets up.
• Every run has a record. Which runs happened, what each task returned, how long it took, where it failed: the interface the video walks through is a log of the pipeline's history.
• Backfills are a decision. catchup makes replaying missed windows a choice you state, rather than something that happens to you.

Related

• Previous: Retraining strategies
• Reference: Airflow DAGs documentation