Parameterized Runs and Backfills
In Sequential Pipeline Steps your taxi_pipeline DAG downloaded one month of TLC green-taxi data: green_tripdata_2024-01.parquet, hardcoded. That works the first time, but real pipelines ingest many months. The TLC publishes monthly parquet files from 2009 through today; on a fresh project you usually want to load a few months of history before the first scheduled run. Doing that by editing the URL string on every run is the wrong answer.
This chapter shows the right answer: parameterize the DAG on the logical date, use {{ ds }} to pick the month, and backfill a contiguous date range with one CLI command.
By the end of this chapter, you should be able to:
{{ ds }}) so the same code processes any month.{{ ds }}, {{ data_interval_start }}, and params for runtime values.airflow backfill create over a date range and verify every month landed correctly.Hardcoded dates make pipelines fragile. Parameterized DAGs let you run the same logic for any month.
Use cases from the Week 12 taxi pipeline:
raw_trips before the first daily run.fct_trips miscalculates tip percentage for trips before 2024-02-01. Fix the Week 10 model, and then backfill the affected months.Airflow gives templated date values inside every task:
{{ ds }}: logical date of the run, YYYY-MM-DD (e.g., 2024-01-01).{{ data_interval_start }}: the start of the execution window this run is processing (discussed in the context-variables extra exercise).In TaskFlow, ds can also be passed as a task function argument; Airflow injects it automatically for scheduled runs. Airflow 3 manual triggers leave logical_date unset unless you pass it in the trigger payload, so the ds: str auto-injection silently becomes ds=None and the task crashes on ds[:7]. The snippet below uses a small _ds_from_context() helper that reads the date through get_current_context() with a fallback: the form that works in both modes.
<aside> 🤓 Curious Geek: From execution_date to logical_date
If you read older Airflow tutorials, you will see references to execution_date. In Airflow 2.2, this was renamed to logical_date to reduce confusion, because the "execution date" was actually the start of the data interval, not the time the task physically executed. Airflow 3 completes this cleanup by fully removing the deprecated execution_date alias and defaulting logical_date to None for manual triggers, which is why our _ds_from_context() helper uses logical_date or dr.run_after as a fallback.
</aside>
{{ ds }}Rewrite the taxi_pipeline from the previous chapter so the parquet URL derives from the run's logical date. Each monthly run downloads one month of data; the same code handles every month. Every Week 12 pattern from the previous chapter stays in place: per-student AIRFLOW_STUDENT schema isolation, DBT_ENV injection through BashOperator.env, catchup=False. The two headline changes are the schedule (@monthly) and the idempotent load (append-after-delete instead of replace); a couple of smaller settings change alongside them (max_active_runs=1 and the _ds_from_context() helper), each called out below the code:
<aside>
📦 Reference repo: The complete, runnable taxi_pipeline.py lives on the ch5-params branch of the reference repo (Step 1 of the hands-on fetches it straight into your dags/ folder), with a copy in assets/dag_snapshots/taxi_pipeline.py. Do not retype it. The snippets below highlight the changes that turn the sequential pipeline into a parameterized monthly run.
</aside>
The @dag decorator is updated with a monthly schedule, a historical start date, and serialization parameters to prevent concurrent runs from colliding:
# dags/taxi_pipeline.py
@dag(
schedule="@monthly", # one run per month (was @daily)
start_date=datetime(2024, 1, 1), # earliest month the backfill will claim
catchup=False, # stays False: load history via backfill command
max_active_runs=1, # serialize: concurrent runs on the same schema clash
default_args={"retries": 2}, # retry transient failures twice
tags=["week12", "taxi"],
)
def taxi_pipeline():
# ...
Two new helper functions derive the download URL and the run's execution date. We extract parquet_url_for as a pure function so it can be tested outside of Airflow:
def parquet_url_for(ds: str) -> str:
"""Return the TLC green-taxi parquet URL for a logical date."""
year_month = ds[:7] # "2024-01-01" -> "2024-01"
return f"{TLC_BASE}/green_tripdata_{year_month}.parquet"
def _ds_from_context() -> str:
"""Return the logical-date string for the current task run.
Using get_current_context() with a run_after fallback works in
both scheduled runs and manual triggers.
"""
ctx = get_current_context()
dr = ctx["dag_run"]
dt = dr.logical_date or dr.run_after
return dt.strftime("%Y-%m-%d")
Instead of recreating the table with if_exists="replace" (which would delete all other months' data), the ingestion task now deletes the existing month's partition first, and then appends the new batch:
@task()
def ingest_taxi_month() -> int:
ds = _ds_from_context()
year_month = ds[:7]
# Fetch and parse parquet
resp = requests.get(parquet_url_for(ds), timeout=60)
resp.raise_for_status()
df = pd.read_parquet(io.BytesIO(resp.content))
# Keep only this logical month. TLC files carry a few spillover rows
# dated N-1 / N+1; without this filter a January re-run grows December
# (Gotcha #4).
df["lpep_pickup_datetime"] = pd.to_datetime(df["lpep_pickup_datetime"])
df = df[
df["lpep_pickup_datetime"].dt.strftime("%Y-%m") == year_month
].copy()
hook = PostgresHook(postgres_conn_id="azure_pg")
engine = hook.get_sqlalchemy_engine()
# On the first run the table does not exist yet: create the schema and
# materialize an empty raw_trips so the DELETE below has a table to hit.
with hook.get_conn() as conn, conn.cursor() as cur:
cur.execute(f'CREATE SCHEMA IF NOT EXISTS "{SCHEMA}"')
df.head(0).to_sql(
"raw_trips", engine, schema=SCHEMA, if_exists="append", index=False,
)
# 1. DELETE the target partition
with hook.get_conn() as conn, conn.cursor() as cur:
cur.execute(
f'DELETE FROM "{SCHEMA}".raw_trips '
"WHERE to_char(lpep_pickup_datetime, 'YYYY-MM') = %s",
(year_month,),
)
# 2. APPEND the filtered monthly partition
df.to_sql(
"raw_trips",
engine,
schema=SCHEMA,
if_exists="append", # was "replace"
index=False,
method="multi",
chunksize=1000,
)
return len(df)
Three differences from the previous chapter matter:
schedule="@monthly") matches the TLC publication cadence. The logical date ds is now the first day of each month.2024-02-01 run deletes February's rows first, then appends this month's batch: never doubling. Compare to the previous chapter's if_exists="replace", which dropped and rewrote the whole table on every run (fine for a single-month demo, wrong for a multi-month pipeline because it would wipe Jan + Mar when rerunning Feb).catchup stays False. In Airflow 3 the explicit airflow backfill create command loads history without needing the scheduler's catchup behavior; enabling catchup on a @monthly DAG with a past start_date makes the scheduler fire one extra run per month between start_date and today on unpause, most of which the TLC has not published yet. Keep it off; use backfill create when you want history.
The max_active_runs=1 setting is new (previous chapter used the default of 16). Three concurrent backfill runs all targeting the same airflow_<name> schema would collide inside dbt: dbt run stages new models via a <name>__dbt_backup rename, and two runs racing to create stg_trips__dbt_backup crash with relation already exists. Serializing to one active run at a time costs about 2x wall-clock (three 30-second runs back-to-back instead of parallel) but makes backfill reliable. For larger pipelines you fix the root cause (per-run schemas, or dbt's --threads tuning); for Week 12, serializing is the right trade-off.
The ingest_taxi_month function uses two different hook APIs against the same connection: hook.get_conn() returns a raw psycopg connection for the CREATE SCHEMA and DELETE statements, and hook.get_sqlalchemy_engine() returns the SQLAlchemy engine that pandas.to_sql wants. Both reuse PostgresHook's underlying connection pool, so there is no duplicate authentication cost.
{{ ds }} vs params: pick the right toolAirflow offers two ways to pass runtime values. Use the right one:
| You need... | Use | Why |
|---|---|---|
| Per-run date logic | {{ ds }} / data_interval_start |
The date is the partition identity. Airflow manages it. |
| Per-run toggles the user sets at trigger time | params |
Custom values the DAG does not infer from the schedule. |
| Long-lived config (secrets, DSNs) | Connections / Variables | Rotates without code changes. |
For the taxi pipeline, the partition is the date. Use {{ ds }}. Reach for params only when a user picks something Airflow cannot derive (e.g., a minimum trip distance filter or a specific vendor ID).
Here is a minimal snippet showing how both parameters coexist in the same DAG:
from datetime import datetime
from airflow.sdk import dag, task, get_current_context, Param
@dag(
schedule="@monthly",
start_date=datetime(2024, 1, 1),
# params defines UI fields the user can customize when triggering manually
params={
"min_trip_distance": Param(0.0, type="number", description="Minimum trip distance to load"),
},
)
def taxi_pipeline():
@task
def ingest_taxi_month():
# 1. ds is inferred from the schedule partition (automated)
ds = _ds_from_context()
# 2. params are read from the run context (user-configured)
ctx = get_current_context()
min_dist = ctx["params"]["min_trip_distance"]
print(f"Loading month {ds[:7]} with min_trip_distance filter: {min_dist}")
A backfill replays historical DAG runs. When you parameterize a DAG on {{ ds }}, you can trigger it for a past date range (like January to March 2024), and Airflow will run the pipeline once per interval in that range, each run processing its respective month.
How you trigger a rerun depends on why you are rerunning:
The code block above evolves the taxi_pipeline.py you ran in the previous chapter into the parameterized version: read the annotations and the Two differences that matter notes so you know what changed and why. To get the runnable file itself, do not hand-edit your Sequential Pipeline Steps copy (the delta is a full rewrite of the ingest task plus two new helper functions, easy to get subtly wrong). Fetch the parse-verified version the same way you fetched that chapter's DAG.
Step 1: Replace your DAG with the parameterized version. From the root of your reference-repo clone, pull the file off the ch5-params branch straight into dags/. This overwrites one file without switching branches, so your .env, requirements.txt, and dbt profile stay exactly as they are:
git show ch5-params:dags/taxi_pipeline.py > dags/taxi_pipeline.py
Step 2: Reserialize and confirm the DAG is paused. Airflow reparses dags/ within a few seconds. Open the UI and confirm taxi_pipeline shows the new @monthly schedule and is paused (toggle off). Leaving it paused is deliberate: backfill create in the next step creates the runs, and unpausing later is what executes them.
Step 3: Run the three-month backfill.
astro dev run backfill create \
--dag-id taxi_pipeline \
--from-date 2024-01-01 \
--to-date 2024-03-31 \
--max-active-runs 1
Here is what these arguments do:
--dag-id specifies which DAG to execute.--from-date and --to-date define the range of partitions to create. Since schedule="@monthly", Airflow creates exactly one run per month in that range (January, February, and March).--max-active-runs 1 serializes execution to one active run at a time, preventing concurrent dbt models from colliding.<aside>
💡 Workflow: In Airflow 3, backfill create registers the runs in a queued state but does not run them immediately. Once you run the command, unpause the DAG in the Airflow UI. The scheduler will pick up the queued backfill runs and execute them. Because catchup=False is set on the DAG, the scheduler will not automatically create any other runs outside of this explicit backfill range.
</aside>
<aside> 💡 Pin the range narrow on the first run. Backfilling three months of TLC data takes seconds per month. However, running a backfill for five years at once translates to ~60 runs against a remote database, which will immediately surface any latency or connection pool issues. Start small.
</aside>
Step 4: Verify each month landed. Use psql or DBeaver to check each month's row count (green taxi: ~50K-60K trips per month in 2024):
SELECT to_char(lpep_pickup_datetime, 'YYYY-MM') AS month, count(*)
FROM airflow_<name>.raw_trips
GROUP BY 1 ORDER BY 1;
You should see exactly three rows, one per month (green taxi runs about 50-60K trips per month; this author measured 56,551 / 53,577 / 57,457 for Jan/Feb/Mar 2024 against the shared class DB). With the month filter in place, you should not see a growing Dec or Apr spillover bucket. Rerun the CLI command: the row counts should not change. That is the idempotency contract paying off.
<aside>
💡 The count above is your airflow_<name>.raw_trips, which the ingest task fills one month at a time. The dbt_run / dbt_test tasks rebuild their models from the shared nyc_taxi reference source (the same Week 10 data), so this exercise proves the load is idempotent rather than growing the dbt models month by month.
</aside>
The Grid view shows these three backfill runs side-by-side as completed green columns. You can expand the tasks to verify that ingest_taxi_month, dbt_run, and dbt_test all executed successfully:
Airflow Grid view for taxi_pipeline showing three successful backfill run columns
The Runs page for taxi_pipeline shows the three green backfill runs tagged with the distinctive Backfill run-type badge, plus (depending on when you unpause) one or two additional Scheduled runs Airflow fires for the current month:
Airflow Runs page for taxi_pipeline showing three green Backfill runs plus a current-month Scheduled run
The run-type column is the key distinction: Backfill means "created by the backfill CLI, claimed for explicit history loading," Scheduled means "created by the scheduler on its normal cadence." A failed current-month Scheduled run (as in the screenshot above, where 2026-04 has no published TLC parquet yet) does not block the backfill runs from succeeding: they ran for real months, not future ones.
Now try the UI alternative to a full CLI backfill. Clear-and-retry rewinds exactly one failed or superseded run without touching the rest of the history.