Schedule Retraining Pipelines with Airflow

Your model was stellar in staging — but in production, the world moved on, data drifted, and your carefully trained model now makes stale predictions. Manually retraining is a nightmare: you forget, you skip, and your ML pipeline becomes a liability instead of an asset. This lesson solves that pain by teaching you how to schedule retraining pipelines with Airflow, the industry-standard workflow orchestrator that automates model refreshes on a cadence you control. By the end, you'll have a reliable, hands-off system that keeps your models fresh without constant babysitting.

The problem this lesson solves

Model degradation is silent and costly. Your production model was trained on last quarter's customer behavior, but churn patterns have shifted, new features launched, and the market responded — your predictions are now drifting. Without a retraining schedule, you're flying blind, and manual retraining is a leaky sieve: someone forgets, a data pipeline breaks silently, and before you know it, your model's accuracy has tanked and business decisions are based on stale insights.

The operational pain is real:

• Manual retraining is error-prone — it relies on human memory and discipline.
• Data dependencies are complex — retraining needs fresh data, validation, and evaluation.
• No audit trail — when a model fails, you can't trace when it was last trained or why.
• Scaling is impossible — you can't hand-run dozens of models weekly.

Airflow solves this by giving you a programmatic, scheduled, and observable way to retrain models automatically. Instead of a cron job that runs a single script, you build a DAG (Directed Acyclic Graph) that orchestrates every step — from data extraction to model deployment — and runs it on a schedule you define, with retries, monitoring, and logging built in.

Core concept / mental model

Think of Airflow as a conductor for a symphony of tasks. Each task (data fetch, preprocessing, training, evaluation, deployment) is a musician. The DAG is the sheet music that defines the order, dependencies, and timing. The scheduler is the conductor who ensures each musician starts exactly when they should, waits for the previous one to finish, and handles mistakes by resetting a section if needed.

In Airflow terms:

• DAG: A Python script that defines the pipeline structure — tasks and their dependencies. It's the blueprint.
• Task: A single unit of work, like fetch_data or train_model. Each task is an instance of an operator (e.g., PythonOperator, BashOperator).
• Schedule: The cadence at which the DAG runs — daily, weekly, hourly — defined by schedule_interval (a cron expression or preset).
• Scheduler: The Airflow component that watches DAGs and triggers task instances according to the schedule and dependencies.
• Executor: The worker that actually runs the tasks (local, Celery, Kubernetes).

A key mental shift: you're not writing a script; you're defining a workflow. The DAG describes what to do and in what order, and Airflow handles when and how.

How it works step by step

Scheduling a retraining pipeline with Airflow follows a logical sequence that moves from data to deployed model:

1. Define the DAG in Python — Create a .py file in your Airflow dags_folder. This file describes the pipeline structure, schedule, and tasks.
2. Set the schedule — Use schedule_interval with a cron expression (e.g., '0 2 * * 1' for every Monday at 2 AM) or a preset like @daily. Airflow will create DAG runs at those times.
3. Define tasks with operators — For each step, instantiate an operator. PythonOperator is common for calling Python functions; BashOperator for shell commands.
4. Set dependencies — Use bitshift operators (>>) to chain tasks: fetch_data >> preprocess >> train >> evaluate >> deploy. This creates a DAG structure that the scheduler respects.
5. Configure retries and timeouts — Add retries and retry_delay to tasks so transient failures (e.g., a flaky API) don't crash the whole pipeline.
6. Trigger the DAG — After placing the file in the DAGs folder, Airflow's scheduler picks it up and starts executing runs per the schedule. You can also trigger manually from the UI.
7. Monitor and debug — Use the Airflow UI to view task logs, status (success/failed), and the DAG graph. When a retrain fails, you see exactly which step broke.

The cause-and-effect chain: schedule -> scheduler triggers DAG run -> tasks execute in order -> dependencies respected -> retries handle flakiness -> model gets refreshed automatically.

Hands-on walkthrough

Let's build a complete retraining pipeline DAG. We'll assume a simple Python function that fetches new data, trains a scikit-learn model, and saves it.

Step 1: Environment setup

Install Airflow in a standalone mode (or use Docker if preferred):

pip install apache-airflow
airflow standalone

The airflow standalone command initializes a SQLite database, creates an admin user, and starts the webserver and scheduler — ideal for learning.

Step 2: Write the DAG file

Create retrain_model_dag.py in your dags folder (default: ~/airflow/dags):

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
import pandas as pd
from sklearn.linear_model import LogisticRegression
import joblib

# Default args applied to all tasks
default_args = {
    'owner': 'ml_team',
    'depends_on_past': False,
    'start_date': datetime(2025, 1, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

# Define the DAG
dag = DAG(
    'model_retraining_pipeline',
    default_args=default_args,
    description='Weekly model retraining pipeline',
    schedule_interval='0 2 * * 1',  # Every Monday at 2 AM
    catchup=False,
    tags=['ml', 'retraining'],
)

# Task 1: Fetch new training data
def fetch_data():
    # Simulate loading data from a source (e.g., database or S3)
    df = pd.read_csv('s3://bucket/new_training_data.csv')
    df.to_csv('/tmp/raw_data.csv', index=False)
    print(f'Fetched {len(df)} rows')

fetch_data_task = PythonOperator(
    task_id='fetch_data',
    python_callable=fetch_data,
    dag=dag,
)

# Task 2: Preprocess data
def preprocess_data():
    df = pd.read_csv('/tmp/raw_data.csv')
    # Example preprocessing: drop missing values, encode, etc.
    df = df.dropna()
    # Assume we split features/labels
    df.to_csv('/tmp/processed_data.csv', index=False)
    print(f'Preprocessed to {len(df)} rows')

preprocess_task = PythonOperator(
    task_id='preprocess_data',
    python_callable=preprocess_data,
    dag=dag,
)

# Task 3: Train the model
def train_model():
    df = pd.read_csv('/tmp/processed_data.csv')
    # Assume last column is target
    X = df.iloc[:, :-1]
    y = df.iloc[:, -1]
    model = LogisticRegression(max_iter=1000)
    model.fit(X, y)
    joblib.dump(model, '/tmp/model.pkl')
    print('Model trained and saved')

train_task = PythonOperator(
    task_id='train_model',
    python_callable=train_model,
    dag=dag,
)

# Task 4: Evaluate the model
from sklearn.metrics import accuracy_score

def evaluate_model():
    model = joblib.load('/tmp/model.pkl')
    df = pd.read_csv('/tmp/processed_data.csv')
    X = df.iloc[:, :-1]
    y = df.iloc[:, -1]
    preds = model.predict(X)
    acc = accuracy_score(y, preds)
    print(f'Accuracy: {acc:.3f}')
    # Could raise exception if acc < 0.9 to fail the DAG

evaluate_task = PythonOperator(
    task_id='evaluate_model',
    python_callable=evaluate_model,
    dag=dag,
)

# Task 5: Deploy to production (simulate)
def deploy_model():
    # Example: copy model to a serving bucket
    print('Deploying model to production endpoint...')

deploy_task = PythonOperator(
    task_id='deploy_model',
    python_callable=deploy_model,
    dag=dag,
)

# Set dependencies
fetch_data_task >> preprocess_task >> train_task >> evaluate_task >> deploy_task

Step 3: Run and observe

Trigger the DAG manually from the UI (or use CLI: airflow dags trigger model_retraining_pipeline). Check the Graph view to see the task dependencies, and click each task to view logs. The DAG will run every Monday at 2 AM from the start date, thanks to the schedule_interval.

Expected output (from logs):

[2025-01-06 02:00:00,000] {python.py:174} INFO - Fetched 10000 rows
[2025-01-06 02:00:05,000] {python.py:174} INFO - Preprocessed to 9800 rows
[2025-01-06 02:00:10,000] {python.py:174} INFO - Model trained and saved
[2025-01-06 02:00:12,000] {python.py:174} INFO - Accuracy: 0.923
[2025-01-06 02:00:15,000] {python.py:174} INFO - Deploying model to production endpoint...

This pipeline is idempotent if you design your tasks to be repeatable — use external storage for intermediate data, avoid global state, and always overwrite artifacts.

Compare options / when to choose what

Airflow isn't the only way to schedule retraining. Here's a comparison with common alternatives:

When to choose Airflow: You have multiple dependent steps (data extraction, validation, training, evaluation, deployment) and need scheduling, retries, and visibility. It's the de facto standard in many ML platforms.

When to avoid: Very simple one-step retraining might be overkill — a cron job or serverless function could suffice. But as your pipeline grows, Airflow pays off.

Troubleshooting & edge cases

Even with Airflow, things go wrong. Here are common issues and fixes:

• DAG not showing up in UI — Ensure your DAG file is in the correct dags_folder (check airflow.cfg), and that there are no syntax errors. Run python <your_dag.py> — it should execute without exceptions.
• Task stuck in "running" state — Often a zombie process. Check the scheduler logs; if the task is crashing, you'll see a traceback. Increase retries or adjust execution_timeout.
• Schedule not triggering — Verify start_date is in the past and the timezone is consistent (Airflow uses UTC by default). Set catchup=False to avoid backfills of past runs.
• Dependencies not executing in order — Double-check your bitshift operators. Use the Graph view to confirm the DAG structure.
• Data race on /tmp — If tasks run concurrently across workers, they might overwrite files. Use unique paths (e.g., {{ ds }} in file names) or use a shared object store like S3.
• Airflow standalone resets DB — For learning that's fine. In production, use a persistent database (PostgreSQL) and externalize the metadata DB.

Pro tip: Use {{ ds }} (execution date) in file paths to make your pipeline idempotent — each DAG run writes to a unique location.

What you learned & what's next

You've learned how to schedule retraining pipelines with Airflow — from the core concepts of DAGs, tasks, and schedules, to building a complete, scheduled retraining pipeline that fetches data, trains a model, evaluates it, and deploys it automatically. You now understand the problem of manual retraining and how Airflow solves it with automation, retries, and monitoring.

Next in this track, you'll explore how to wire this into a broader MLOps workflow — perhaps hooking Airflow to model registries, monitoring drift after deployment, or integrating with Kubernetes for distributed training. Keep building your automation muscle: your model will thank you.

Key insight: Automated retraining is not just about convenience; it's about maintaining model reliability and business trust. Airflow gives you a structured, auditable way to do that.