Parameterized Runs and Backfills
Scheduling and Triggers ended with a promise: the next chapter would wire task dependencies so the execution order matches the real taxi pipeline (download → load → dbt run → dbt test). This is where you keep that promise.
The hello_pipeline DAG from Airflow Fundamentals and Scheduling and Triggers was deliberately a toy: one integer in, one print statement out. Good for learning the Airflow primitives, not enough to orchestrate anything real. In this chapter you build a second DAG called taxi_pipeline that sits alongside hello_pipeline in dags/ and runs the full Week 10 dbt project under Airflow. You can delete hello_pipeline at the end of the week if you want a clean project; the exercises assume both DAGs coexist for now.
By the end of this chapter, you should be able to:
>> operator.BashOperator + --project-dir flags, end-to-end against the shared Azure Postgres.BashOperator) instead of TaskFlow @task, and how TaskFlow passes data between tasks through XComs.In Airflow, dependencies define execution order. The common syntax is the shift-operator chain:
task_a >> task_b >> task_c
<aside> 🎬 Animated visual: how the >> operator wires tasks into an execution order
</aside>
https://gist.githack.com/lassebenni/36ee746733b3d6654513870841153d56/raw/week_12__dag_execution_order_animation.html
Equivalent APIs exist (task_a.set_downstream(task_b), task_b.set_upstream(task_a)) but nobody uses them in new code: >> is short, directional, and reads left-to-right like the pipeline itself.
Until now, the raw taxi data was already waiting for you in Postgres. Across Week 9 to Week 11 you queried nyc_taxi and built dbt models on top of it, but you never had to get the data in: it was loaded for you so you could focus on SQL and dbt. In production, no one hands you a pre-filled table. A pipeline has to fetch the source files and load them, on a schedule, every time new data lands, and building that pipeline is what orchestration is for. This chapter is where you take over that first step yourself, instead of relying on a table someone seeded ahead of time.
The Week 12 scenario is concrete: download a month of TLC green taxi data, load it into the raw_trips table your Week 10 dbt project already depends on, then run dbt. Three tasks, strict order:
ingest_taxi_month: fetch the TLC parquet for one month and load it into raw_trips in Azure Postgres, in a single transaction.dbt_run: rebuild stg_trips and fct_trips from the new raw data.dbt_test: assert the Week 10 tests still pass.# dags/taxi_pipeline.py — structure only. Copy the full, runnable file from the
# reference repo (see the callout below); the parts that matter are annotated
# underneath. Imports and the ingest body are elided here for readability.
STUDENT = os.environ.get("AIRFLOW_STUDENT", "default") # your Postgres role name
SCHEMA = f"airflow_{STUDENT}" # your own schema in the shared DB
DBT_DIR = "/usr/local/airflow/include/dbt_project"
TLC_URL = "<https://d37ci6vzurychx.cloudfront.net/trip-data/green_tripdata_2024-01.parquet>"
# dbt runs through uvx on Python 3.11 (the Astro image ships 3.14, which stable
# dbt-core does not support yet). Credentials come from the azure_pg connection.
dbt = "uvx --python 3.11 --from 'dbt-core==1.10.*' --with 'dbt-postgres==1.10.*' dbt"
DBT_ENV = {"PG_HOST": "{{ conn.azure_pg.host }}", "PG_USER": "{{ conn.azure_pg.login }}", ...}
@dag(schedule="@daily", start_date=datetime(2025, 1, 1), catchup=False, tags=["week12", "taxi"])
def taxi_pipeline():
@task()
def ingest_taxi_month() -> int:
# download the TLC parquet in memory and load it into
# airflow_<student>.raw_trips in one atomic task
...
dbt_run = BashOperator(
task_id="dbt_run",
# deps first: packages.yml is committed, dbt_packages/ is gitignored
bash_command=(
f"{dbt} deps --project-dir {DBT_DIR} --profiles-dir {DBT_DIR} && "
f"{dbt} run --project-dir {DBT_DIR} --profiles-dir {DBT_DIR}"
),
env=DBT_ENV,
append_env=True,
)
dbt_test = BashOperator(
task_id="dbt_test",
bash_command=f"{dbt} test --project-dir {DBT_DIR} --profiles-dir {DBT_DIR}",
env=DBT_ENV,
append_env=True,
)
ingest_taxi_month() >> dbt_run >> dbt_test
taxi_pipeline()
<aside>
📦 Reference repo: The complete, runnable taxi_pipeline.py lives on the ch4-taxi branch of the reference repo (Step 5 of the hands-on fetches it into your dags/ folder with one command, no branch switch needed), with a copy in assets/dag_snapshots/taxi_pipeline.py. Do not retype it. The hands-on below installs and runs it end-to-end; once it is green, the Look closer at the DAG notes underneath break down the parts that matter.
</aside>
Follow these six steps end-to-end before moving on. Each builds on the previous one; a failure in step 3 will block steps 4-6, so get the stack green at each step.
Step 1: Add the five packages the pipeline needs to your Astro project. The default Astro runtime is intentionally lean and ships none of PostgresHook, pandas, or requests. Append to requirements.txt:
apache-airflow-providers-postgres
psycopg2-binary
pyarrow
pandas
requests
Each line earns its place:
apache-airflow-providers-postgres + psycopg2-binary expose PostgresHook + the driver SQLAlchemy uses for hook.get_sqlalchemy_engine().pandas + pyarrow power pd.read_parquet on the TLC parquet (pyarrow is the columnar decoder pandas calls).requests fetches the parquet over HTTPS.<aside>
⚠️ dbt is deliberately not in this list. Astro Runtime 3.3 ships Python 3.14, and stable dbt-core does not support 3.14 yet: baking it into the image produces a dbt binary that crashes on import (mashumaro ... is not serializable). Instead the dbt_run / dbt_test tasks run dbt through uvx on Python 3.11, the exact interpreter your Week 10 dbt project targets (requires-python = ">=3.11,<3.14"). uv already ships in the Astro image; uvx --python 3.11 ... dbt run downloads Python 3.11 and dbt once, caches them, and starts instantly afterwards, so you run the same stable dbt-core 1.10 as Week 10 without a broken 3.14 build.
</aside>
Rebuild the container image: astro dev restart. First rebuild takes 2-3 minutes.
<aside> 🤓 Curious Geek: Why edit requirements.txt instead of using uv?
In earlier weeks, you used uv add to manage Python packages. You might wonder if you can use uv or pyproject.toml directly for Astronomer.
The answer is no: the Astro CLI's Docker build process expects a standard requirements.txt at the root of your Astro project directory to install dependencies inside the Airflow containers.
If you want to use uv locally to manage these packages, you can run uv add in your local project environment and export them to keep the files in sync: uv export --format requirements-txt > requirements.txt.
</aside>
Step 2: Set AIRFLOW_STUDENT to your Postgres role name. This goes in your Astro project's .env file (Astro reads it automatically on start). The DAG reads it at parse time to pick your per-student schema airflow_<name>, which your role owns. This isolation is what makes running the DAG against the shared Azure Postgres safe: your role owns only airflow_<name> and is denied write access to a classmate's, so the DAG's if_exists="replace" load can only ever touch your own raw_trips, never public.raw_trips or anyone else's data. Use the exact role name from your Key Vault login (for example firstname_lastname, with an underscore), not a nickname: it has to match the schema you have permission to write to. See AIRFLOW_STUDENT.
# <astro-project>/.env
AIRFLOW_STUDENT=your_postgres_role
Step 3: Create the azure_pg connection from your own least-privilege credentials. You do not use a shared admin login. Each student has a personal Postgres role that can read nyc_taxi and owns only its own airflow_<name> and dev_<name> schemas, so a bug in your DAG can never touch a classmate's data. Your connection URL lives in Key Vault as postgres-url-<name>, where <name> is the same role name you set in AIRFLOW_STUDENT in Step 2 (the same secret you used for dbt in Weeks 9-10). Log in with your own Entra account (az login), fetch the URL, and hand the whole URI to Airflow. Run these in bash (WSL or Git Bash on Windows), since the \ line-continuations and $(...) substitution are bash syntax:
PG_URL=$(az keyvault secret show \
--vault-name kv-hyf-data \
--name postgres-url-<name> \
--query value -o tsv)
astro dev run connections add azure_pg --conn-uri "$PG_URL"
The URI already carries the host, your login, your password, the team1 database, and ?sslmode=require (Azure Postgres Flexible Server requires SSL). Airflow parses those into the connection, so you never paste a password into a command or commit one to Git.
Confirm the connection was created before moving on, because a missing or misnamed connection is a common reason steps 4-6 fail. Open Admin → Connections in the Airflow UI and check that azure_pg is listed with the host hyf-data-pg.postgres.database.azure.com and port 5432:
Airflow Admin Connections page listing the azure_pg Postgres connection with host hyf-data-pg and port 5432
The per-row Test icon is greyed out because Astro disables connection-testing by default, so this list is your confirmation that the connection exists and points at the right host. The first real check of the credential itself is Step 6: if it is wrong, ingest_taxi_month fails with an authentication error you can read straight from its task log.
<aside> 🤓 Curious Geek: Can we just put the connection string in .env?
In Week 10, you configured your database connection by putting credentials directly into environment variables in your .env file. You can actually do the same in Airflow by adding AIRFLOW_CONN_AZURE_PG=your_postgres_url to your .env file.
However, we use the connections add CLI command instead because connections defined as environment variables are completely hidden from the Airflow UI (Admin → Connections). By writing the connection to Airflow's metadata database instead, it is stored centrally and appears in the UI list where you can easily inspect and manage it.
</aside>
Step 4: Create the dbt profile. There is nothing to copy here: the reference repo you cloned in Scheduling and Triggers already ships the dbt project at include/dbt_project/, the same nyc-taxi models and tests you built in Week 10 (this is the path the DBT_DIR constant above points at). The only file it does not commit is the active profile, so create it from the shipped example:
cp include/dbt_project/profiles.yml.example include/dbt_project/profiles.yml
That profiles.yml.example already reads its credentials from environment variables (host: "{{ env_var('PG_HOST') }}", user: "{{ env_var('PG_USER') }}", and so on), which is exactly what the DAG's DBT_ENV injects through BashOperator.env. So dbt runs under Airflow with no further profile edits.
Step 5: Add taxi_pipeline.py to dags/ and reserialize. From the root of your reference-repo clone, pull the DAG file off the ch4-taxi branch straight into your dags/ folder. This copies one file without switching branches, so your .env, requirements.txt, and hello_pipeline.py stay exactly as they are:
git show ch4-taxi:dags/taxi_pipeline.py > dags/taxi_pipeline.py
Then reserialize: astro dev run dags reserialize. Confirm taxi_pipeline appears in the UI, then unpause it. The Graph view should show a three-node horizontal chain: ingest_taxi_month → dbt_run → dbt_test.
Airflow graph view: taxi_pipeline three-node chain
Step 6: Trigger the DAG in the UI and wait for all three tasks to turn dark green. The first run is slow: dbt_run triggers uvx to download Python 3.11 and dbt (about two minutes) before dbt even starts, so if dbt_run sits in running for a minute or two on the first trigger, that is expected, not a hang. Later runs reuse the cache and start instantly. In the Grid view you should see one column with three stacked green cells matching the >> order. Click the dbt_test task and open its log: the last line should read Done. PASS=8 WARN=3 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=11. All three findings are warn-level, so the task stays green while still surfacing the counts: WARN 4 duplicate trips, WARN 3415 null payment*types, and WARN 5 trips whose pickup*location*id has no matching zone. These are the same Week 10 data-quality issues, carried through the shared nyc*taxi source.
Airflow dbt_test task log showing Done. PASS=8 WARN=3 ERROR=0 with the three warn-level Week 10 data-quality findings (4 duplicates, 3415 null payment_types, 5 orphaned pickup zones) and exit code 0
If your log matches the screenshot line-for-line you are running the same reality Week 10 tested. Triggering the DAG a second time reproduces exactly the same numbers: the pipeline is idempotent against the shared Azure Postgres, because airflow_<name> schema isolation means the replace-on-load only rewrites your own partition of the class DB.
If any task goes red, open its logs and match the error to the stage that failed: a network or DB auth/schema error points at ingest_taxi_month, a compilation error at dbt_run, a data-quality failure at dbt_test. The three-task chain keeps each stage's failures in its own grid cell.
<aside>
📦 Stuck? Compare your DAG against this chapter's exact end-state on the ch4-taxi branch of the reference repo (git switch ch4-taxi, then diff dags/taxi_pipeline.py): a complete, parse-verified project you can also astro dev start. For just the final snapshot, see assets/dag_snapshots/taxi_pipeline.py. The next chapter (Parameterized Runs and Backfills) evolves this same taxi_pipeline.py by switching to a monthly schedule and a {{ ds }}-driven DELETE-then-append so reruns stay idempotent; the three-task structure is identical.
</aside>
Now that your DAG has run green, five details in it are worth a second look. Each one lands better now that you have seen the pipeline succeed:
ingest_taxi_month fetches the parquet and writes it to raw_trips in the same task, keeping the bytes in memory (io.BytesIO(resp.content) feeds pandas directly). One process makes the task atomic: either the month got downloaded AND loaded, or neither did, with nothing half-written to your schema. (pd.read_parquet and the columnar-storage rationale come from Week 3 Ch4: Reading file formats; by Week 12 it is a tool you already have, not a new one to learn.)requests.get(url) does NOT raise on a 4xx or 5xx response; it just returns a Response whose .content holds the error body (HTML, usually). Without raise_for_status(), a TLC 403 would silently "succeed" and you would only discover the problem minutes later when pandas fails to parse the HTML as parquet. Making the HTTP error visible at the right task is the cheapest bug fix you can make. Monitoring and Debugging walks through the exact log pattern when this goes wrong.postgres_conn_id="azure_pg" looks up the credentials from the Connection you created in step 3; hook.get_sqlalchemy_engine() returns a SQLAlchemy engine that pandas' to_sql can write through. You never hardcode the password: the connection handles it.raw_trips table on every run. That is what made your second trigger reproduce the exact same numbers (the idempotency you just saw). It is fine for Week 12's monthly load but dangerous at scale: a production pipeline would use "append" plus a DELETE-by-partition pattern instead (Parameterized Runs and Backfills shows the production-grade version).dbt test fails the Airflow task cleanly, the red cell you would have seen if any test had errored instead of warned.With the stack running end-to-end, one last design note on how small to keep each task:
<aside> 💡 Keep tasks small. One task should do one clear action. The three-task chain above is deliberate: each task maps to one stage of the pipeline (ingest, transform, test), which makes the red cell in the grid view diagnostic on its own.
</aside>
Your taxi_pipeline mixes both of Airflow's task-authoring styles: ingest_taxi_month is a TaskFlow @task, while dbt_run and dbt_test are classic BashOperator instances. The rule of thumb for which to reach for:
@task) for Python-centric work: calling APIs, querying databases via Python hooks, or manipulating pandas DataFrames. Passing one @task's return value into another wires the dependency and moves the data for you (through XComs) with no boilerplate. This is why ingest_taxi_month is a @task.dbt_run and dbt_test are BashOperator: dbt is a CLI, not in-process Python. Provider operators (such as PostgresOperator or WasbHook) are classic operators built for specific external services.<aside>
📚 For the classic PythonOperator + xcom_pull style, the side-by-side comparison, and how @task instantiates a PythonOperator under the hood, see the optional Going Further page.
</aside>
The two BashOperator tasks above are the canonical way to run a dbt project from Airflow. The pattern is deliberately boring: Airflow shells out to the dbt CLI exactly the way you would at the terminal, and the exit code of dbt run / dbt test becomes the exit code of the Airflow task. A red task in the grid view means dbt exited non-zero: same signal you already know from dbt build.
Four details make or break this pattern in practice:
include/dbt_project/ because Astro mounts the include/ folder into every container. If you put the project in dags/, Airflow will try to parse it as DAG code and you will see ParseError noise in the UI.