🎒 Assignment: Orchestration

The Scenario

Your team's morning report depends on a pipeline that someone still runs by hand. You already built ingestion and transformation logic in earlier weeks. Your next step is to operate that logic like a real data platform: automatic runs, clear dependencies, historical backfills, and failure visibility, so that nobody has to babysit it at 6 AM.

Your task is to build a production-style Airflow DAG that orchestrates a multi-step pipeline end-to-end.

Difficulty levels

Choose your level at the start.

Required (pass): one DAG with 3 sequential tasks (including dbt_run and dbt_test), schedule, retry, {{ ds }}-parameterized runs, 7-day backfill with row count check (idempotency proof), clear runbook, plus one successful deploy to the shared class Airflow (all three tasks green: dbt_run / dbt_test provision dbt via uvx --python 3.11, see Ch8).
Extra (bonus): Required + alert callback, unit tests for TaskFlow task logic, custom params, and stronger failure simulation.

The Required tier stays passable via local Astro execution alone if the shared class Airflow VM is offline or your cohort has not set it up.

Build one DAG that includes:

A schedule. @monthly matches the Ch5 snapshot's month-partitioned pipeline and is the natural choice; use a daily schedule only if you re-partition your ingest to load one day per run (otherwise every daily run reprocesses the same month).
At least three sequential tasks.
Retry behavior for transient failure.
{{ ds }}-parameterized execution so reruns are idempotent.
A 7-day (or 7-month, depending on your cadence) historical backfill.
One green run deployed to the shared class Airflow (if the shared VM is online).

Requirements

Task 1: DAG setup

Create an Astro project and add your DAG in dags/.
DAG must include:
a schedule (for example @monthly to match the Ch5 snapshot, or "0 6 1 * *" for 06:00 on the first of each month)
start_date
catchup=False for normal operation
max_active_runs=1 on the @dag(...) decorator (not only on the backfill CLI; see Gotcha #6)
at least one descriptive tag
retries and retry_delay in default_args (or on a critical task)

Task 2: Sequential workflow with the Week 10 dbt project

Create at least three tasks in a strict dependency chain:

ingest: produce or fetch the raw input your DAG will process
dbt_run: run dbt with BashOperator, using the uvx --python 3.11 ... dbt run --project-dir <path> --profiles-dir <path> form from Sequential Pipelines, against your Week 10 project. Plain dbt crashes on the Astro image's Python 3.14; uvx provisions a working Python 3.11 dbt.
dbt_test: run dbt test with the same uvx wrapper and flags

The dbt integration is required; it is the realistic use case for Week 12 and the mechanics were covered in Sequential Pipelines. Mount your Week 10 project under include/dbt_project/ (Astro's convention) and use --project-dir / --profiles-dir flags rather than cd.

If your Week 10 project is not runnable, the Week 12 reference repo ships a working dbt project at lassebenni/nyc-taxi-airflow-reference under include/dbt_project/. Copy that directory into your assignment project:

git clone <https://github.com/lassebenni/nyc-taxi-airflow-reference> /tmp/class-ref
cp -r /tmp/class-ref/include/dbt_project include/dbt_project

Document which project you used (your Week 10 project or the class reference) in ASSIGNMENT_REPORT.md.

Task 3: Parameterized runs

Your DAG should process one month (or day, depending on your cadence choice) of data per run, with the partition derived from Airflow's logical date rather than datetime.now(). If you followed the Parameterized Runs and Backfills snapshot, you already have this: _ds_from_context() returns the current run's date, and the ingest task slices it down to a year-month for the DELETE and the parquet URL.

Read {{ ds }} (via the _ds_from_context() helper, or as a templated task argument for scheduled runs) in at least one task and use it to pick the partition.
Date-driven partitioning is required: your teacher checks that the partition comes from the logical date, not datetime.now().
Filter the parquet to the logical month before you write. After read_parquet, keep only rows whose pickup timestamp falls in that month, then DELETE that month and append. Without the filter, TLC spillover rows dated N-1 / N+1 make adjacent months drift on every re-run (Gotcha #4). Your idempotency proof depends on this.
Optional extra: add a custom DAG params entry (for example {"dataset": "green"} to pick between green_tripdata and yellow_tripdata) and use it in your ingest task.

Task 4: Retry and failure handling

Configure retry behavior (retries, retry_delay) for at least one critical task.
Demonstrate one failure and recovery path in your documentation.
Optional extra: add alerting callback if your environment supports it.

Task 5: Backfill

Perform a 7-run backfill. Airflow 3 replaced airflow dags backfill with backfill create. The date range you pass should be large enough to fire 7 runs for your schedule:
Monthly schedule (@monthly, matching the Ch5 snapshot): 7 months → --from-date 2024-01-01 --to-date 2024-07-31
Daily schedule (only if your ingest partitions by day): 7 days → --from-date 2024-01-01 --to-date 2024-01-07

  astro dev run backfill create \
    --dag-id <your_dag_id> \
    --from-date 2024-01-01 \
    --to-date 2024-07-31 \
    --max-active-runs 1

--max-active-runs 1 on the CLI is required if your DAG invokes dbt (concurrent dbt_run tasks collide on __dbt_backup relations; see Gotcha #6). Keep the matching max_active_runs=1 on the @dag(...) decorator too: the CLI flag alone does not protect scheduled or manually triggered concurrent runs.
Verify each month processed correctly (only that month's rows; no spillover growth in adjacent months).
Document evidence that reruns do not create duplicates or inconsistent results: query your airflow_<yourname> schema, paste the monthly row counts into ASSIGNMENT_REPORT.md, re-run the same backfill, confirm the counts are identical, and paste the after counts too. Screenshots of the query results help; numbers in the report are required.

Task 6: Operational notes

Create RUNBOOK.md with:

how to trigger the DAG manually
how to run a backfill
how to inspect task logs
top 3 likely failures and first-response steps

Task 7: Deploy to the shared Airflow (Required)

Deploy your final DAG to the class's shared Airflow using the workflow you practiced in Deploying to Shared Airflow. The deploy PR goes to the shared repo, but your graded submission does not. Everything a reviewer needs, your DAG code, the screenshot, and the merged-PR link, goes into this assignment repo. Reviewers grade from your assignment repo only and never open the shared deploy repo. To log into the shared UI (to trigger the run and take the screenshot), use student-<yourname>; your password is your own Key Vault secret airflow-ui-password-<yourname>, fetched exactly as in Ch8 Step 1. If the shared class Airflow VM is offline or unavailable, you can skip this deploy step.

<aside> 💡 Local vs shared dag_id: keep dag_id="taxi_pipeline" (or whatever you used) in this assignment repo for local Astro. Only the copy you put in c55-shared-airflow must be renamed to <yourname>_taxi_pipeline. Local Graph / Grid / log screenshots may show taxi_pipeline. The shared-UI screenshot must show your namespaced DAG with all three tasks green. You do not need to redo Tasks 1–6 under the new name.

</aside>

Copy your DAG into dags/<yourname>/ of the shared deploy repo, lassebenni/c55-shared-airflow (you are a collaborator; see Ch8 for setup).
In that shared copy only: rename the dag_id to <yourname>_taxi_pipeline (example: alex_taxi_pipeline) and add "student:<yourname>" to tags. CI rejects student DAGs whose dag_id is not prefixed with the folder name.
Open a PR. If it only touches your own dags/<yourname>/ folder, guarded auto-merge merges it the moment the integrity check passes, and the DAG appears in the shared UI within ~60 seconds, no teacher needed.
Trigger one manual run on the shared scheduler, explicitly passing a past logical date (e.g. 2024-01-01) so the TLC parquet actually exists. Confirm it writes to your airflow_<yourname> schema on the shared Azure PG.
Capture a screenshot of the shared UI grid view showing your green run on <yourname>_taxi_pipeline, filtered by your student: tag (all three tasks green).
Record the URL of your merged c55-shared-airflow PR in ASSIGNMENT_REPORT.md as your deploy proof. That link plus the screenshot is all the reviewer uses: they do not visit the shared repo.

<aside> ⚠️ Airflow 3's backfill create uses deterministic run-ids like backfill__2024-01-01T00:00:00+00:00. If a classmate already triggered a backfill for the same DAG + logical date, your second invocation is a no-op (no duplicate run is created). For the assignment, one manual trigger is enough; backfill on the shared VM is not required.

</aside>

This is what submitting real production work looks like: your code runs next to other people's code on shared infrastructure, you own your own schema, and you do not touch classmates' DAGs.

Task 8: Document your AI usage (all tiers)

Create AI_ASSIST.md recording at least one point where you used an LLM during this assignment (debugging a red task, drafting the runbook, explaining a backfill create flag). For each use, write down:

the prompt you sent (⚠️ no connection strings, passwords, or real data)
what the model answered
what you kept, changed, or discarded, and why

A submission with no documented AI use is incomplete on this dimension, even if the rest is strong. The point is to show critical evaluation, not to prove you avoided AI.

Deliverables

Required submissions (all of these live in this assignment PR; a green autograder alone is not a pass):

Astro project with runnable DAG code (max_active_runs=1 on @dag, month filter on ingest, logical-date partition)
RUNBOOK.md
AI_ASSIST.md (at least one documented, critically-evaluated LLM use)
filled ASSIGNMENT_REPORT.md containing:
schedule choice and reason
task dependency graph explanation
one debugging case you resolved
explanation of the {{ ds }} parameter usage and the backfill command(s) you ran
before/after monthly row counts for the idempotency re-run
the URL of your merged c55-shared-airflow deploy PR (deploy proof, if the VM was online)
screenshots committed into the PR (any folder is fine: screenshots/, evidence/, images/):
Graph view of DAG (local Astro; taxi_pipeline is fine)
successful run in Grid view (local Astro; taxi_pipeline is fine)
one task log snippet (local Astro is fine)
a successful deploy in the shared Airflow UI on <yourname>_taxi_pipeline, filtered by your student:<yourname> tag, showing all three tasks green (dbt_run / dbt_test run dbt via uvx --python 3.11, per Ch8). Required only if the VM was online.

(tests/test_dag_integrity.py is required for the assignment, listed in the Definition-of-done checklist below.)

Definition of done checklist

Before submission, confirm all items. The CI autograder is a smoke test: it does not look at screenshots, row counts, or the shared deploy.

Required:

[ ] DAG is visible in Airflow UI without import errors (astro dev pytest tests/test_dag_integrity.py --args "-v" passes)
[ ] @dag(...) sets max_active_runs=1 (not only the backfill CLI flag)
[ ] Ingest filters parquet rows to the logical month before DELETE/append (Gotcha #4)