Imagine you're a data engineer at a retail company. Your team just finished building a pipeline that loads daily sales data from your point-of-sale system into a data warehouse. You flip the switch, Airflow starts running it every morning at 6 AM, and life is good — for about two weeks.
Then your manager asks: "Can we also get the last 18 months of historical data in there? We need it for the annual review." Suddenly you're not just thinking about tomorrow's run. You need to load hundreds of past days of data, do it reliably, and do it without duplicating records you may have already loaded or accidentally skipping gaps. This is the moment when your understanding of how Airflow thinks about time and scheduling becomes the difference between a clean warehouse and a disaster.
This lesson will give you a thorough, practical understanding of how Apache Airflow handles scheduling, historical backfills, and the design patterns that make pipelines safe to run more than once. By the end, you'll be able to confidently configure Airflow's scheduling behavior, trigger backfills on demand, and build pipelines that won't corrupt your data even when things go sideways.
What you'll learn:
catchup does and when to enable or disable itBefore we talk about backfilling, you need to understand something counterintuitive about how Airflow schedules work. Most people assume that when a DAG is scheduled to run at midnight, it runs at midnight for that day's data. Airflow does something different, and if you don't understand it, backfilling will confuse you completely.
Airflow uses a concept called the execution date (called logical_date in Airflow 2.2+, though execution_date still works). Here's the key insight: Airflow runs a DAG at the end of the interval it covers, not the beginning.
Let's make that concrete. You have a DAG with schedule_interval='@daily' and a start_date of 2024-01-01. The first DAG run fires on January 2nd at midnight — but its execution_date is January 1st. Airflow is saying: "This run is responsible for processing the data from the January 1st interval."
Think of it like a night security guard. The guard who shows up Monday morning is reporting on what happened Sunday night. The execution date is Sunday night; the actual trigger time is Monday morning.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
def log_execution_context(**context):
execution_date = context['execution_date']
next_execution_date = context['next_execution_date']
print(f"This run covers: {execution_date}")
print(f"The next run will cover: {next_execution_date}")
print(f"Actual time this task started: {datetime.utcnow()}")
with DAG(
dag_id='time_model_demo',
start_date=datetime(2024, 1, 1),
schedule_interval='@daily',
catchup=False,
) as dag:
explain_time = PythonOperator(
task_id='explain_time',
python_callable=log_execution_context,
)
When you run this DAG today and look at the logs, you'll see the execution date is yesterday — Airflow is processing yesterday's interval, even though it ran this morning. This interval-based thinking is exactly what makes backfilling possible and coherent.
A DAG Run is a single execution of your entire pipeline for a specific time interval. Every time Airflow triggers your DAG — whether on a schedule, manually, or through a backfill — it creates a DAG Run with a unique execution_date.
You can see all DAG Runs for a given DAG in the Airflow UI by clicking on the DAG's name in the main dashboard, then navigating to the "Runs" tab. Each row represents one DAG Run, showing its status (running, success, failed), the execution date it covers, and when it actually started.
DAG Runs have three possible origins:
schedule_intervalUnderstanding these origins matters because Airflow will not create a duplicate DAG Run for the same execution_date. That's a safety mechanism — it's how you avoid processing the same day's data twice by accident.
Now we get to catchup, which is the single most important setting for controlling what Airflow does when you deploy a new DAG or bring one back online after a pause.
Here's the scenario: You deploy a DAG on March 15th with a start_date of January 1st. Airflow looks at that gap — January 1st through March 14th — and asks: "Should I create DAG Runs for all those missed intervals?"
If catchup=True (which is Airflow's default), the answer is yes. Airflow will immediately queue up DAG Runs for every single day between January 1st and March 14th. That's 74 DAG Runs firing at once.
If catchup=False, Airflow ignores the historical gap and starts scheduling from the current date forward. You get one run, not 74.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
# This DAG will NOT create runs for past intervals
with DAG(
dag_id='sales_daily_load',
start_date=datetime(2024, 1, 1),
schedule_interval='@daily',
catchup=False, # <-- Start fresh, don't backfill automatically
max_active_runs=1,
) as dag:
load_sales = PythonOperator(
task_id='load_daily_sales',
python_callable=lambda **ctx: print(f"Loading sales for {ctx['ds']}"),
)
Tip:
catchup=Falseis the right default for most production pipelines. Automatic catchup flooding your task queue with hundreds of simultaneous runs is a common source of Airflow cluster meltdowns. When you actually need to backfill, do it deliberately with the backfill command — you'll have much more control.
When should you use catchup=True? When your pipeline is explicitly designed to process one interval at a time, your tasks are idempotent (we'll cover this shortly), your infrastructure can handle the parallelism, and you want Airflow to automatically self-heal any missed runs without manual intervention. Some teams use it for audit pipelines or data quality checks where gaps are simply not acceptable.
If you do use catchup=True, pair it with max_active_runs. This parameter limits how many DAG Runs from the same DAG can execute simultaneously.
with DAG(
dag_id='sales_daily_load',
start_date=datetime(2024, 1, 1),
schedule_interval='@daily',
catchup=True,
max_active_runs=3, # Never more than 3 historical runs at once
) as dag:
...
With max_active_runs=3, Airflow will process historical backlog runs three at a time, queuing the rest. Your workers won't get crushed.
Here's the right way to handle that "we need 18 months of historical data" request from your manager: use the Airflow CLI's backfill command. This gives you explicit control over the date range, parallelism, and whether to re-run already-successful runs.
The basic syntax is:
airflow dags backfill \
--start-date 2023-01-01 \
--end-date 2023-12-31 \
sales_daily_load
This command tells Airflow to create and execute DAG Runs for every scheduled interval between January 1st and December 31st, 2023. For a daily DAG, that's 365 runs.
What actually happens under the hood: Airflow iterates through every execution date in the range, checks if a successful DAG Run already exists for that date, and if not, creates one and runs it. If a successful run already exists for a date, Airflow skips it by default — another safety mechanism.
# Run at most 4 parallel task instances at once
airflow dags backfill \
--start-date 2023-01-01 \
--end-date 2023-12-31 \
--max-jobs 4 \
sales_daily_load
# Force re-run even if successful runs exist (use with caution)
airflow dags backfill \
--start-date 2023-06-01 \
--end-date 2023-06-30 \
--reset-dagruns \
sales_daily_load
# Do a dry run - see what would execute without actually running anything
airflow dags backfill \
--start-date 2023-01-01 \
--end-date 2023-01-07 \
--dry-run \
sales_daily_load
Warning: The
--reset-dagrunsflag will delete existing DAG Runs in that date range before recreating them. Use this carefully in production — it will re-process data that was already successfully loaded, which can create duplicates unless your tasks are idempotent.
At this point you've seen the word "idempotent" come up twice. Let's actually define it.
A task is idempotent if running it multiple times with the same inputs produces the same result as running it once. If you run the same DAG Run for January 15th five times — maybe because of failures, retries, or accidental re-runs — your data warehouse should end up in exactly the same state as if you'd run it once.
This is not optional. Here's why.
Airflow retries failed tasks. Backfills occasionally need to be re-run due to upstream data fixes. Engineers accidentally trigger DAG Runs twice. Bugs happen. If your tasks aren't idempotent, every one of these situations corrupts your data.
Here's a task that looks reasonable but will destroy you:
def load_daily_sales(**context):
ds = context['ds'] # execution date as string, e.g. '2024-01-15'
# Fetch sales data for that day
sales_records = fetch_from_api(date=ds)
# Insert into the warehouse
db.execute(
"INSERT INTO sales_fact SELECT * FROM staging_sales WHERE sale_date = %s",
(ds,)
)
Run this twice for January 15th and you'll have two copies of every January 15th sale in your warehouse. Your revenue totals will double. Your finance team will not be amused.
The most common pattern for making SQL-based loads idempotent is to delete the data for that interval before inserting it:
def load_daily_sales(**context):
ds = context['ds']
sales_records = fetch_from_api(date=ds)
# Step 1: Remove any existing data for this date
db.execute(
"DELETE FROM sales_fact WHERE sale_date = %s",
(ds,)
)
# Step 2: Insert fresh data
db.execute(
"INSERT INTO sales_fact SELECT * FROM staging_sales WHERE sale_date = %s",
(ds,)
)
Now you can run this ten times for the same date. The final state is always: exactly one copy of that day's sales.
In modern data warehouses (Snowflake, BigQuery, Redshift, Postgres 9.5+), you often have access to upsert or merge operations, which update existing rows if they exist and insert them if they don't:
def load_daily_sales(**context):
ds = context['ds']
# This MERGE handles both new and re-run scenarios cleanly
db.execute("""
MERGE INTO sales_fact AS target
USING staging_sales AS source
ON target.sale_id = source.sale_id
AND target.sale_date = %s
WHEN MATCHED THEN
UPDATE SET
amount = source.amount,
customer_id = source.customer_id
WHEN NOT MATCHED THEN
INSERT (sale_id, sale_date, amount, customer_id)
VALUES (source.sale_id, source.sale_date, source.amount, source.customer_id)
""", (ds,))
This is elegant but requires your records to have a stable unique key (sale_id in this case). If you can guarantee that, MERGE is your best friend for idempotent loads.
Not all tasks are SQL. If your task writes a file to S3 or a filesystem, idempotency means overwriting, not appending:
def export_daily_report(**context):
ds = context['ds']
report_data = generate_report(date=ds)
# Writing to a deterministic path means re-runs just overwrite
s3_key = f"reports/daily/sales_report_{ds}.parquet"
s3_client.put_object(
Bucket='my-data-bucket',
Key=s3_key,
Body=report_data.to_parquet()
)
# Uploading twice to the same key? Same result. Idempotent.
The key insight here is that the output path includes the execution date (ds), making it deterministic. If this task runs again for the same date, it writes to the same path and overwrites the existing file. No duplicates.
Airflow provides a set of template variables (also called Jinja macros) that let you inject the execution date and related values directly into your task logic. This is how you make a task genuinely interval-aware.
The most commonly used ones:
| Variable | Description | Example |
|---|---|---|
{{ ds }} |
Execution date as YYYY-MM-DD |
2024-01-15 |
{{ ds_nodash }} |
Execution date without dashes | 20240115 |
{{ execution_date }} |
Full datetime object | 2024-01-15T00:00:00+00:00 |
{{ next_ds }} |
Next execution date as YYYY-MM-DD |
2024-01-16 |
{{ prev_ds }} |
Previous execution date as YYYY-MM-DD |
2024-01-14 |
In Python operators, you access these through the context dictionary. In SQL-templated operators like PostgresOperator, you can use the Jinja syntax directly:
from airflow.providers.postgres.operators.postgres import PostgresOperator
load_sales_task = PostgresOperator(
task_id='load_sales_to_warehouse',
postgres_conn_id='warehouse_postgres',
sql="""
DELETE FROM sales_fact
WHERE sale_date = '{{ ds }}';
INSERT INTO sales_fact (sale_id, sale_date, amount, customer_id)
SELECT sale_id, sale_date, amount, customer_id
FROM sales_staging
WHERE sale_date = '{{ ds }}';
""",
)
The {{ ds }} gets substituted with the actual execution date at runtime. This single template makes the delete-then-insert pattern work across all backfill dates automatically — run it for January 1st, and {{ ds }} becomes 2024-01-01. Run it for March 15th, and it becomes 2024-03-15. One piece of code, all dates handled correctly.
Let's build a complete, backfill-ready pipeline from scratch. You'll create a DAG that simulates loading daily sales summaries and verify that it behaves idempotently.
Setup: Create a SQLite database file at /tmp/sales_warehouse.db.
# dags/sales_backfill_exercise.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import sqlite3
import random
DB_PATH = '/tmp/sales_warehouse.db'
def initialize_database():
"""Create the target table if it doesn't exist."""
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS daily_sales_summary (
sale_date TEXT PRIMARY KEY,
total_revenue REAL,
transaction_count INTEGER,
loaded_at TEXT
)
""")
conn.commit()
conn.close()
def load_daily_summary(**context):
"""Idempotent load of daily sales summary."""
ds = context['ds']
# Simulate fetching aggregated data from a source system
# In reality, this would be an API call or a read from a source DB
random.seed(ds) # Same seed = same data every time for same date
total_revenue = round(random.uniform(10000, 50000), 2)
transaction_count = random.randint(100, 800)
conn = sqlite3.connect(DB_PATH)
# Idempotent: DELETE existing record for this date, then INSERT
conn.execute(
"DELETE FROM daily_sales_summary WHERE sale_date = ?",
(ds,)
)
conn.execute(
"""INSERT INTO daily_sales_summary
(sale_date, total_revenue, transaction_count, loaded_at)
VALUES (?, ?, ?, datetime('now'))""",
(ds, total_revenue, transaction_count)
)
conn.commit()
# Verify the insert
result = conn.execute(
"SELECT * FROM daily_sales_summary WHERE sale_date = ?", (ds,)
).fetchone()
conn.close()
print(f"Successfully loaded: {result}")
def verify_no_duplicates(**context):
"""Audit task: fail if any date has duplicate records."""
conn = sqlite3.connect(DB_PATH)
duplicates = conn.execute("""
SELECT sale_date, COUNT(*) as cnt
FROM daily_sales_summary
GROUP BY sale_date
HAVING cnt > 1
""").fetchall()
conn.close()
if duplicates:
raise ValueError(f"Duplicate records found: {duplicates}")
print("Audit passed: No duplicate records in daily_sales_summary")
with DAG(
dag_id='sales_backfill_exercise',
start_date=datetime(2024, 1, 1),
schedule_interval='@daily',
catchup=False,
max_active_runs=2,
tags=['exercise', 'backfill'],
) as dag:
init_db = PythonOperator(
task_id='initialize_database',
python_callable=initialize_database,
)
load_summary = PythonOperator(
task_id='load_daily_summary',
python_callable=load_daily_summary,
)
audit_duplicates = PythonOperator(
task_id='verify_no_duplicates',
python_callable=verify_no_duplicates,
)
init_db >> load_summary >> audit_duplicates
Step 1: Drop this file in your Airflow dags/ folder and let it be detected (usually within 30 seconds).
Step 2: Run a backfill for January 2024:
airflow dags backfill \
--start-date 2024-01-01 \
--end-date 2024-01-31 \
sales_backfill_exercise
Step 3: Verify the data loaded correctly:
sqlite3 /tmp/sales_warehouse.db \
"SELECT COUNT(*), MIN(sale_date), MAX(sale_date) FROM daily_sales_summary;"
You should see 31 rows, from 2024-01-01 to 2024-01-31.
Step 4: Run the backfill again for the same range (simulating a re-run):
airflow dags backfill \
--start-date 2024-01-01 \
--end-date 2024-01-31 \
--reset-dagruns \
sales_backfill_exercise
Step 5: Check again. You should still see exactly 31 rows. The idempotent delete-then-insert protected you.
"My backfill created hundreds of runs and crashed my Airflow cluster."
You ran backfill without --max-jobs on a large date range, or your DAG has catchup=True with no max_active_runs. Restart with --max-jobs 4 and add max_active_runs=3 to your DAG definition. Always test backfills on a short range first.
"The backfill skipped some dates that actually need to be re-run."
By default, backfill won't re-run dates that already have a successful DAG Run. If upstream data was corrected and you need to reload those dates, use --reset-dagruns. Be intentional — this deletes those run records before recreating them.
"My DAG ran and loaded data, but the execution date is a day behind what I expected."
This is the interval model at work. Airflow's execution date represents the start of the interval, and the run fires at the end. A daily DAG with execution date 2024-01-15 runs on January 16th. If you want to query data from "today," use {{ next_ds }} — it represents the end boundary of the current interval.
"I set start_date=datetime.now() and the DAG never runs."
This is a very common mistake. start_date should always be a fixed past date, not a dynamic one. When Airflow calculates when to start scheduling, it needs a stable anchor. datetime.now() changes every time the DAG file is parsed, and Airflow gets confused. Use datetime(2024, 1, 1) or a similar hardcoded date.
"My backfill is running but tasks are failing with duplicate key errors."
Your tasks aren't idempotent. You're inserting without deleting first, and the re-run is trying to insert records that already exist. Add the DELETE step before each INSERT, or switch to an UPSERT/MERGE pattern.
You now have a solid mental model for one of the trickier aspects of Airflow: the relationship between time, scheduling, and data correctness.
Here's what you've learned:
catchup=True automatically creates DAG Runs for all past intervals since start_date; catchup=False starts fresh from todaymax_active_runs is the safety valve that prevents catchup or backfill from overwhelming your infrastructureairflow dags backfill command lets you deliberately load historical date ranges with full control over parallelism and re-run behavior{{ ds }} and related template variables are what bind your task logic to specific time intervalsNext steps to deepen your skills:
{{ prev_ds }} and {{ ds }} to load only the changed data in each interval, rather than full refreshesLearning Path: Modern Data Stack