diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..a334663 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,8 @@ +astro +.git +.env +airflow_settings.yaml +logs/ +.venv +airflow.db +airflow.cfg diff --git a/.hyf/grader_lib.sh b/.hyf/grader_lib.sh index 3142cfe..1ba13d3 100644 --- a/.hyf/grader_lib.sh +++ b/.hyf/grader_lib.sh @@ -7,11 +7,9 @@ # and a set of common static-analysis checks derived from recurring # PR review patterns across cohort c55. # -# blocker(): use for leaked-secret findings (a committed profiles.yml/.env, -# a hardcoded password/connection string). It behaves like fail() for the -# printed report, but also flips a flag that forces write_score() to report -# pass=false regardless of the earned point total -- a leaked secret must -# be fixed before the PR can pass, it cannot be "pointed around." +# blocker(): use for findings that must fail the PR regardless of points +# (leaked secrets, missing required evidence like screenshots). Behaves like +# fail() in the printed report, but forces write_score() to pass=false. _grader_details=() _grader_blocker=false @@ -38,7 +36,7 @@ write_score() { [[ "$score" -ge "$passing" ]] && pass_flag="true" if [[ "$_grader_blocker" == true ]]; then pass_flag="false" - echo "๐Ÿšซ A blocker was found (leaked secret) -- forcing pass=false regardless of score." >&2 + echo "๐Ÿšซ A blocker was found -- forcing pass=false regardless of score." >&2 fi cat > "$outfile" << JSON { diff --git a/.hyf/test.sh b/.hyf/test.sh index e693692..c596847 100755 --- a/.hyf/test.sh +++ b/.hyf/test.sh @@ -3,8 +3,9 @@ # The DAG needs a running Astro/Airflow stack and a live Azure PostgreSQL # connection that CI cannot reach, so this checks file presence and code # patterns in dags/taxi_pipeline.py and the docs. The actual green run, -# Screenshot files are presence-checked; content, backfill idempotency, and -# shared-Airflow deploy are reviewed by a teacher. +# Screenshot files are required (โ‰ฅ3): missing screenshots force pass=false. +# Content of those shots, backfill idempotency, and shared-Airflow deploy +# are still reviewed by a teacher. # Total points: 100. Passing score: 60. set -euo pipefail @@ -218,9 +219,10 @@ shot_count=$( if [[ "$shot_count" -ge 3 ]]; then l6=$((l6 + 3)); pass "screenshots: found ${shot_count} image file(s) (need โ‰ฅ3 for Graph + Grid/run + task log)" elif [[ "$shot_count" -gt 0 ]]; then - fail "screenshots: only ${shot_count} image file(s) โ€” commit at least 3 (local Graph, green Grid/run, one task log; add shared-UI shot when the VM is up)" + # Screenshots are required evidence for teacher review โ€” cannot pass without them. + blocker "screenshots: only ${shot_count} image file(s) โ€” commit at least 3 (local Graph, green Grid/run, one task log; add shared-UI shot when the VM is up)" else - fail "screenshots: none found โ€” commit Graph, Grid/run, and task-log images into the PR (any folder)" + blocker "screenshots: none found โ€” commit Graph, Grid/run, and task-log images into the PR (any folder). Screenshots are required; a high code score without them still fails." fi score=$((score + l6)) pass "Level 6: documentation + screenshots ($l6/10 pts)" diff --git a/AI_ASSIST.md b/AI_ASSIST.md index 171da98..e09021b 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -1,12 +1,20 @@ # AI assistance log + Never paste connection strings, passwords, or real data. Replace TODO. --> ## Use 1 -**Prompt I sent:** _Replace this section._ +**Prompt I sent:** +I received an Airflow task failure with the error: +AirflowNotFoundException: The conn_id azure_pg isn't defined -**What the model answered:** _Replace this section._ +I asked the LLM to explain the cause of this error and what steps were needed to fix the Airflow PostgreSQL connection. -**What I kept, changed, or discarded, and why:** _Replace this section._ + +**What the model answered:** +The model explained that the DAG was trying to use an Airflow connection named `azure_pg` through `PostgresHook`but this connection hadnt been created in the Airflow environment. It suggested checking the Airflow connections and adding the missing connection using the Astro CLI. + +**What I kept, changed, or discarded, and why:** +I kept the explanation that the issue was caused by a missing Airflow connection. I verified the solution by creating the `azure_pg` connection in my Astro Airflow environment and rerunning the DAG. The task succeeded afterward. +I configured my own environment variables and connection settings. diff --git a/ASSIGNMENT_REPORT.md b/ASSIGNMENT_REPORT.md index 97422f5..82f53c8 100644 --- a/ASSIGNMENT_REPORT.md +++ b/ASSIGNMENT_REPORT.md @@ -1,31 +1,49 @@ # Assignment report - + ## Schedule choice and reason -_Replace this section._ + we did a monthly schedule (@monthly) because the TLC taxi dataset is partitioned by month. Each DAG run processes one month of data based on Airflow logical_date. ## Task dependency graph -_Replace: describe ingest -> dbt_run -> dbt_test and why order matters._ +describe the chain (ingest -> dbt_run -> dbt_test) and why the order matters. + +ingest_taxi_month -> dbt_run -> dbt_test + +The ingest task runs first because it downloads and loads the raw taxi data into PostgreSQL. After the raw data is available, 'dbt_run' transforms the data using the the class reference. Finally, 'dbt_test' validates the dbt models to ensure data quality. + +ingest loads raw data +dbt transforms data +dbt tests validate models. ## dbt project used -_Replace: your Week 10 project or the class reference?_ +the class reference ## One debugging case I resolved -_Replace: what failed, how you found the cause in the logs, and the fix._ +what failed, how you found the cause in the logs, and the fix. + +The first manual run failed because the DAG used the current date and tried downloading green_tripdata_2026-07.parquet, which was unavailable I think . + +I fixed this by triggering the DAG with a logical date (2024-01-01) and using Airflow logical_date instead of datetime.now(). -## Parameterized runs and backfill +The DAG uses Airflow's logical date to derive the year-month partition. -_Replace: how {{ ds }} / logical date drives the partition; the exact backfill create command you ran (with --max-active-runs 1)._ +The ingest task reads dag_run.logical_date and converts it to YYYY-MM format to build the TLC parquet URL. -## Idempotency row counts (before / after re-run) +Backfill command used: +astro dev run backfill create \ + --dag-id taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-07-31 \ + --max-active-runs 1 -_Replace: paste monthly counts before the re-run, then after. They must match._ +## shared Airflow deploy PR + https://github.com/lassebenni/c55-shared-airflow/pull/7 -## Shared Airflow deploy proof (if VM online) -_Replace: merged c55-shared-airflow PR URL + path to your shared-UI screenshot in this repo._ + diff --git a/RUNBOOK.md b/RUNBOOK.md index 84bca1c..4c339b0 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -1,22 +1,57 @@ # RUNBOOK - ## How to trigger the DAG manually -_Replace this section._ +1. Start the local Astro Airflow environment: -## How to run a backfill - -_Replace this section._ +```bash +astro dev start +## How to run a backfill +The DAG uses a monthly schedule, so backfills should run month by month. + +astro dev run backfill create \ + --dag-id taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-07-31 \ + --max-active-runs 1 + +After the backfill finishes, verify the processed rows in PostgreSQL: + +SELECT + to_char(lpep_pickup_datetime, 'YYYY-MM') AS month, + COUNT(*) AS rows +FROM airflow_baraah.raw_trips +GROUP BY 1 +ORDER BY 1; ## How to inspect task logs -_Replace this section._ +1. open the UI airflow +2. open taxi_pipeline +3. slect the run DAG and then click the task +ingest_taxi_month +dbt_run +dbt_test +4. open logs ## Top 3 likely failures and first response -1. _Replace: symptom, first check, fix_ -2. _Replace this section._ -3. _Replace this section._ +1. TLC data download failure (HTTP 403 or missing parquet file) + +Symptom: ingest_taxi_month fails with an HTTP error. +First check: Inspect the task log and verify the logical date used for the run. +Fix: Use a valid historical logical date where the TLC parquet file exists (for example 2024-01-01). + +2. PostgreSQL connection failure +Symptom: Error such as The conn_id azure_pg isn't defined. +First check: Verify the Airflow connection exist. +Fix: Add or update the PostgreSQL connection. + +3. dbt run or dbt test failure +Symptom: dbt_run or dbt_test task becomes failed. +First check: Open the task logs and check the dbt error message. +Fix: Verify that the dbt project exists under include/dbt_project , +and that dbt is executed through: uvx --python 3.11 diff --git a/dags/.airflowignore b/dags/.airflowignore new file mode 100644 index 0000000..e69de29 diff --git a/dags/exampledag.py b/dags/exampledag.py new file mode 100644 index 0000000..7c024cf --- /dev/null +++ b/dags/exampledag.py @@ -0,0 +1,98 @@ +""" +## Astronaut ETL example DAG + +This DAG queries the list of astronauts currently in space from the +Open Notify API and prints each astronaut's name and flying craft. + +There are two tasks, one to get the data from the API and save the results, +and another to print the results. Both tasks are written in Python using +Airflow's TaskFlow API, which allows you to easily turn Python functions into +Airflow tasks, and automatically infer dependencies and pass data. + +The second task uses dynamic task mapping to create a copy of the task for +each Astronaut in the list retrieved from the API. This list will change +depending on how many Astronauts are in space, and the DAG will adjust +accordingly each time it runs. + +For more explanation and getting started instructions, see our Write your +first DAG tutorial: https://www.astronomer.io/docs/learn/get-started-with-airflow + +![Picture of the ISS](https://www.esa.int/var/esa/storage/images/esa_multimedia/images/2010/02/space_station_over_earth/10293696-3-eng-GB/Space_Station_over_Earth_card_full.jpg) +""" + +from airflow.sdk import Asset, dag, task +from pendulum import datetime +import requests + + +# Define the basic parameters of the DAG, like schedule and start_date +@dag( + start_date=datetime(2025, 4, 22), + schedule="@daily", + doc_md=__doc__, + default_args={"owner": "Astro", "retries": 3}, + tags=["example"], +) +def example_astronauts(): + # Define tasks + @task( + # Define an asset outlet for the task. This can be used to schedule downstream DAGs when this task has run. + outlets=[Asset("current_astronauts")] + ) # Define that this task updates the `current_astronauts` Asset + def get_astronauts(**context) -> list[dict]: + """ + This task uses the requests library to retrieve a list of Astronauts + currently in space. The results are pushed to XCom with a specific key + so they can be used in a downstream pipeline. The task returns a list + of Astronauts to be used in the next task. + """ + try: + r = requests.get("http://api.open-notify.org/astros.json") + r.raise_for_status() + number_of_people_in_space = r.json()["number"] + list_of_people_in_space = r.json()["people"] + except Exception: + print("API currently not available, using hardcoded data instead.") + number_of_people_in_space = 12 + list_of_people_in_space = [ + {"craft": "ISS", "name": "Oleg Kononenko"}, + {"craft": "ISS", "name": "Nikolai Chub"}, + {"craft": "ISS", "name": "Tracy Caldwell Dyson"}, + {"craft": "ISS", "name": "Matthew Dominick"}, + {"craft": "ISS", "name": "Michael Barratt"}, + {"craft": "ISS", "name": "Jeanette Epps"}, + {"craft": "ISS", "name": "Alexander Grebenkin"}, + {"craft": "ISS", "name": "Butch Wilmore"}, + {"craft": "ISS", "name": "Sunita Williams"}, + {"craft": "Tiangong", "name": "Li Guangsu"}, + {"craft": "Tiangong", "name": "Li Cong"}, + {"craft": "Tiangong", "name": "Ye Guangfu"}, + ] + + context["ti"].xcom_push( + key="number_of_people_in_space", value=number_of_people_in_space + ) + return list_of_people_in_space + + @task + def print_astronaut_craft(greeting: str, person_in_space: dict) -> None: + """ + This task creates a print statement with the name of an + Astronaut in space and the craft they are flying on from + the API request results of the previous task, along with a + greeting which is hard-coded in this example. + """ + craft = person_in_space["craft"] + name = person_in_space["name"] + + print(f"{name} is currently in space flying on the {craft}! {greeting}") + + # Use dynamic task mapping to run the print_astronaut_craft task for each + # Astronaut in space + print_astronaut_craft.partial(greeting="Hello! :)").expand( + person_in_space=get_astronauts() # Define dependencies using TaskFlow API syntax + ) + + +# Instantiate the DAG +example_astronauts() diff --git a/dags/taxi_pipeline.py b/dags/taxi_pipeline.py index 1786d30..581c7a0 100644 --- a/dags/taxi_pipeline.py +++ b/dags/taxi_pipeline.py @@ -11,10 +11,19 @@ """ import os -from datetime import datetime +import io +from datetime import datetime, timedelta from pathlib import Path -from airflow.sdk import dag, task +import pandas as pd +import requests + +from airflow.sdk import dag, task, get_current_context +from airflow.providers.standard.operators.bash import BashOperator +from airflow.providers.postgres.hooks.postgres import PostgresHook + + + # Your per-student schema. AIRFLOW_STUDENT is set in .env for local Astro dev; # on the shared VM it falls back to the dags// directory name. @@ -37,28 +46,162 @@ def find_dbt_dir() -> str: DBT_DIR = find_dbt_dir() +DBT_ENV = { + "PG_HOST": "{{ conn.azure_pg.host }}", + "PG_USER": "{{ conn.azure_pg.login }}", + "PG_PASSWORD": "{{ conn.azure_pg.password }}", + "PG_DBNAME": "{{ conn.azure_pg.schema }}", + "PG_SCHEMA": SCHEMA, +} + +def _partition_date(): + + context = get_current_context() + + dag_run = context["dag_run"] + + date = dag_run.logical_date or dag_run.run_after + + return date.strftime("%Y-%m-%d") + + + @dag( - # Task 1 (see README): configure the decorator โ€” schedule, start_date, - # catchup=False, max_active_runs=1, default_args retries, tags. - start_date=datetime(2024, 1, 1), + # TODO Task 1 (see README): configure the decorator. + dag_id="baraah_taxi_pipeline", + schedule="@monthly", + start_date=datetime(2024,1,1), + catchup=False, + tags=["week12","taxi", "student:baraah"], + default_args={ + "retries": 2, + "retry_delay": timedelta(minutes=5), + }, + ) def taxi_pipeline(): @task def ingest_taxi_month() -> int: - """Download one month of TLC green-taxi data and load it into - ``{SCHEMA}.raw_trips`` idempotently. Return the number of rows. - - Task 2 and Task 3 (see README): derive the partition from the - logical date, DELETE-then-append that month, and filter the - parquet to the logical month before write (Gotcha #4). - """ - raise NotImplementedError - # Task 2 (see README): add the two transform tasks, wire the full + ds = _partition_date() + + year_month = ds[:7] + + print(f"Processing month: {year_month}") + + url = ( + f"{TLC_BASE}/" + f"green_tripdata_{year_month}.parquet" + ) + #download parquet + response = requests.get( + url, + timeout=60 + ) + + response.raise_for_status() + + #parquet to dataframe + df = pd.read_parquet( + io.BytesIO(response.content) + ) + + + hook = PostgresHook( + postgres_conn_id="azure_pg" + ) + + + engine = hook.get_sqlalchemy_engine() + #create schema + with hook.get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + f'CREATE SCHEMA IF NOT EXISTS "{SCHEMA}"' + ) + #create table if missing + df.head(0).to_sql( + "raw_trips", + engine, + schema=SCHEMA, + if_exists="append", + index=False, + ) + + # idempotency remove old data for same month + with hook.get_conn() as conn: + with conn.cursor() as cur: + + cur.execute( + f""" + DELETE FROM "{SCHEMA}".raw_trips + WHERE to_char( + lpep_pickup_datetime, + 'YYYY-MM' + ) = %s + """, + (year_month,), + ) + #insert fresh data + df.to_sql( + "raw_trips", + engine, + schema=SCHEMA, + if_exists="append", + index=False, + ) + + + return len(df) + + + + + dbt = ( + "uvx --python 3.11 " + "--from 'dbt-core==1.10.*' " + "--with 'dbt-postgres==1.10.*' " + "dbt" + ) + + + dbt_run = BashOperator( + task_id="dbt_run", + bash_command=( + f"{dbt} deps " + f"--project-dir {DBT_DIR} " + f"--profiles-dir {DBT_DIR} && " + f"{dbt} run " + f"--project-dir {DBT_DIR} " + f"--profiles-dir {DBT_DIR}" + ), + env=DBT_ENV, + append_env=True, + ) + + + dbt_test = BashOperator( + task_id="dbt_test", + bash_command=( + f"{dbt} test " + f"--project-dir {DBT_DIR} " + f"--profiles-dir {DBT_DIR}" + ), + env=DBT_ENV, + append_env=True, + ) + + + ingest_taxi_month() >> dbt_run >> dbt_test + """Download one month of TLC green-taxi data and load it into + ``{SCHEMA}.raw_trips`` idempotently. Return the number of rows. + + TODO Task 2 and Task 3 (see README). + """ + + # TODO Task 2 (see README): add the two transform tasks, wire the full # chain, and run the transform through the Chapter 4 command so it works - # on the image's Python. Task 4: add retry behaviour. - - ingest_taxi_month() + # on the image's Python. TODO Task 4: add retry behaviour. taxi_pipeline() diff --git a/include/dbt_project/.github/workflows/dbt-build.yml b/include/dbt_project/.github/workflows/dbt-build.yml new file mode 100644 index 0000000..5ea8bcf --- /dev/null +++ b/include/dbt_project/.github/workflows/dbt-build.yml @@ -0,0 +1,37 @@ +name: dbt build + +on: + push: + branches: [main] + pull_request: + branches: [main] + schedule: + # Daily at 06:00 UTC โ€” catches drift even when no PRs land. + - cron: "0 6 * * *" + +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install dbt-core + dbt-postgres + run: pip install "dbt-core==1.11.*" "dbt-postgres==1.10.*" + + - name: Render profiles.yml from template + run: | + mkdir -p ~/.dbt + sed "s/dev_/dev_ci/" profiles.yml.example > ~/.dbt/profiles.yml + + - name: dbt deps + run: | + if [ -f packages.yml ]; then dbt deps; fi + + - name: dbt build + env: + PG_PASSWORD: ${{ secrets.PG_PASSWORD }} + run: dbt build --select +fct_trips --profiles-dir ~/.dbt diff --git a/include/dbt_project/.gitignore b/include/dbt_project/.gitignore new file mode 100644 index 0000000..e07de29 --- /dev/null +++ b/include/dbt_project/.gitignore @@ -0,0 +1,7 @@ +target/ +dbt_packages/ +logs/ +profiles.yml +.env +.DS_Store +.user.yml diff --git a/include/dbt_project/README.md b/include/dbt_project/README.md index 67f2b5f..8389dbb 100644 --- a/include/dbt_project/README.md +++ b/include/dbt_project/README.md @@ -1,21 +1,100 @@ -# Put your dbt project here +# nyc-taxi-dbt-reference -Your `dbt_run` / `dbt_test` tasks run dbt against a project mounted at -`include/dbt_project/`. Astro mounts this folder into the container at -`/usr/local/airflow/include/dbt_project`. +Reference dbt project for **HYF Data Track Week 10**. Mirrors the `nyc_taxi` project that the +chapters in `Data Track/Week 10/` walk you through building. -Use **one** of these: +This repo is a **safety net**, not a starter. The chapters expect you to type each file +yourself. If you fall behind or something breaks, check out the tag for the chapter you are +on and diff your local copy against this one. -1. **Your own Week 10 project.** Copy your `nyc-taxi-dbt` models, `dbt_project.yml`, - and `profiles.yml` into this directory. -2. **The class reference project** (if your Week 10 project is not runnable): +## Tags (one per chapter checkpoint) - ```bash - git clone https://github.com/lassebenni/nyc-taxi-airflow-reference /tmp/class-ref - cp -r /tmp/class-ref/include/dbt_project/. include/dbt_project/ - ``` +| Tag | Chapter | What you should have at this point | +| ----------------------- | ---------------------------------------- | ------------------------------------------------------------------- | +| `ch2-dbt-setup-azure` | Ch 2 โ€” dbt Setup for Azure PostgreSQL | `dbt_project.yml`, `profiles.yml.example`, `_sources.yml`, plain `stg_trips.sql` | +| `ch3-sql-jinja` | Ch 3 โ€” SQL and Jinja Templating | `stg_trips.sql` extended with `payment_type_label`, `tip_pct`, `fare_per_mile` | +| `ch4-materializations` | Ch 4 โ€” Materializations & Layers | `stg_zones.sql`, `fct_trips.sql`, materialization config in `dbt_project.yml` | +| `ch5-dbt-tests` | Ch 5 โ€” dbt Tests | All schema YAML, singular test, unit test, `packages.yml` | +| `ch6-docs-extras` | Ch 6 โ€” Docs & Extras | Fleshed-out `_fct_trips.yml` descriptions, doc block, `mutable_zones` seed + snapshot | -Document which one you used in `ASSIGNMENT_REPORT.md`. +```bash +git checkout ch4-materializations # rewind to end of Chapter 4 +git checkout main # latest (= end of Chapter 6) +``` -> The dbt tasks call dbt through `uvx --python 3.11`, so you do **not** put -> `dbt-core` in `requirements.txt`. See Chapter 4 for the exact command. +## Setup + +Prereqs: dbt-core 1.11 with `dbt-postgres` (see [Ch 2](https://github.com/hackyourfuture/datatrack/blob/main/Data%20Track/Week%2010/week_10__2_dbt_setup_azure.md) for install). + +```bash +cp profiles.yml.example profiles.yml # then edit `schema:` to dev_ +export PG_PASSWORD='your-week-6-password' # macOS/Linux/WSL +just deps # dbt deps (only needed from v4 onward) +just build # dbt build --select +fct_trips +``` + +`profiles.yml` is gitignored โ€” it stays local so the file never lands in version control with a real password. + +## Running from Airflow (Week 11) + +[Week 11 Chapter 4](https://github.com/hackyourfuture/datatrack/blob/main/Data%20Track/Week%2011/week_11__4_sequential_pipelines.md) runs this dbt project end-to-end from an Airflow `BashOperator`. Three concrete setup steps on top of the local-dev setup above: + +### 1. Copy the project into your Astro project's `include/` folder + +Astro mounts `include/` into every container. Putting the dbt project under `include/dbt_project/` is the convention Ch4's `DBT_DIR` constant expects: + +```bash +# from your Astro project directory +cp -r ../nyc-taxi-dbt-reference include/dbt_project + +# or, if you cloned this repo as a submodule: +git submodule add git@github.com:hackyourfuture/nyc-taxi-dbt-reference include/dbt_project +cd include/dbt_project && git checkout week-11-airflow && cd ../.. +``` + +### 2. Pass the connection credentials through `BashOperator.env` + +The `profiles.yml.example` on this branch reads **four** env vars with sensible defaults for the shared class DB: `PG_HOST`, `PG_USER`, `PG_DBNAME`, `PG_SCHEMA`, plus `PG_PASSWORD` which has no default (security). Pull them from the `azure_pg` Airflow Connection you created in Ch4 and forward them to the bash command: + +```python +DBT_DIR = "/usr/local/airflow/include/dbt_project" +DBT_ENV = { + "PG_HOST": "{{ conn.azure_pg.host }}", + "PG_USER": "{{ conn.azure_pg.login }}", + "PG_PASSWORD": "{{ conn.azure_pg.password }}", + "PG_DBNAME": "{{ conn.azure_pg.schema }}", + "PG_SCHEMA": "airflow_taxi", # team-shared schema, not per-student +} + +dbt_run = BashOperator( + task_id="dbt_run", + bash_command=f"dbt run --project-dir {DBT_DIR} --profiles-dir {DBT_DIR}", + env=DBT_ENV, + append_env=True, # inherit PATH etc. from the worker +) +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, +) +``` + +No passwords in `profiles.yml`, no passwords in DAG code, all credentials flow through the Airflow Connection the way the rest of Ch4 teaches. + +### 3. Use a shared schema, not `dev_` + +Airflow writes into a single schema that downstream dashboards will read from. Pick one team-shared name (the example above uses `airflow_taxi`) and match it to whatever schema the `load_raw_trips` task writes to. A per-student `dev_` schema defeats the point of orchestrating a shared pipeline. + +### Differences from `main` + +Only two deliberate differences from the `main` branch: + +- `profiles.yml.example` reads `PG_HOST`, `PG_USER`, `PG_DBNAME`, `PG_SCHEMA` as env vars in addition to `PG_PASSWORD`, so the same file works for local dbt CLI (env vars fall back to the class defaults) and for Airflow (env vars come from the `azure_pg` connection). +- This README gained the Airflow section above. + +All model code, tests, seeds, and snapshots are byte-identical to `main`. Students who finished Week 10 on `main` can switch to `week-11-airflow` without needing to redo any of the SQL work. + +## CI + +Every push runs `dbt build --select +fct_trips` against the shared Azure PostgreSQL instance. If the curriculum prose drifts away from runnable reality (test result counts change, columns get renamed, etc.) CI fails and the chapter gets fixed. See `.github/workflows/dbt-build.yml`. diff --git a/include/dbt_project/dbt_project.yml b/include/dbt_project/dbt_project.yml new file mode 100644 index 0000000..1da9e46 --- /dev/null +++ b/include/dbt_project/dbt_project.yml @@ -0,0 +1,21 @@ +name: 'nyc_taxi' +version: '1.0.0' +profile: 'nyc_taxi' + +model-paths: ["models"] +analysis-paths: ["analyses"] +test-paths: ["tests"] +seed-paths: ["seeds"] +macro-paths: ["macros"] +snapshot-paths: ["snapshots"] + +clean-targets: + - "target" + - "dbt_packages" + +models: + nyc_taxi: + staging: + +materialized: view + marts: + +materialized: table diff --git a/include/dbt_project/justfile b/include/dbt_project/justfile new file mode 100644 index 0000000..8edd683 --- /dev/null +++ b/include/dbt_project/justfile @@ -0,0 +1,30 @@ +DBT := "dbt" + +# Show available recipes +default: + @just --list + +# Verify dbt can connect to Azure PostgreSQL +debug: + {{DBT}} debug --profiles-dir . + +# Install dbt packages from packages.yml (only needed from v4 onward) +deps: + {{DBT}} deps + +# Build all models from sources through fct_trips, with tests interleaved +build: + {{DBT}} build --select +fct_trips --profiles-dir . + +# Build models only (no tests) โ€” useful while iterating locally +run: + {{DBT}} run --select +fct_trips --profiles-dir . + +# Run every test in the project +test: + {{DBT}} test --profiles-dir . + +# Drop and rebuild everything from a clean target/ +clean-build: + {{DBT}} clean + just build diff --git a/include/dbt_project/models/marts/_fct_trips.yml b/include/dbt_project/models/marts/_fct_trips.yml new file mode 100644 index 0000000..4bccd80 --- /dev/null +++ b/include/dbt_project/models/marts/_fct_trips.yml @@ -0,0 +1,50 @@ +version: 2 + +models: + - name: fct_trips + description: | + One row per completed NYC green taxi trip in January 2024, with + pickup/dropoff zone attributes folded in (OBT-style mart). Queried + directly by dashboards and ad-hoc analysis. + + **Grain:** one row per trip. + **Source:** `public.raw_trips` joined to `public.raw_zones` on + `pickup_location_id` and `dropoff_location_id`. + **Not included:** trips where `pickup_location_id` is NULL (dropped + in `stg_trips`); duplicate rows from the TLC source are kept as-is + and surfaced by `dbt_utils.unique_combination_of_columns`. + columns: + - name: pickup_datetime + description: Wall-clock time the trip began (America/New_York, no timezone attached). + tests: [not_null] + - name: dropoff_datetime + description: Wall-clock time the trip ended. + - name: fare_amount + description: Metered fare in USD, not including tip, tolls, or surcharges. + - name: tip_amount + description: Tip in USD. Non-zero only when payment_type is credit card (1). + - name: trip_distance + description: Distance in miles as reported by the taximeter. + - name: tip_pct + description: | + `tip_amount / fare_amount`, rounded to 4 decimals. NULL when + `fare_amount` is 0 (voided trips, no-charge rides). + - name: fare_per_mile + description: | + `fare_amount / trip_distance`, rounded to 4 decimals. NULL when + `trip_distance` is 0 (data-quality anomalies). + - name: payment_type_label + description: | + Human-readable payment method from the TLC code. See the jinja + dictionary in `stg_trips.sql` for the 1-6 โ†’ label mapping. + - name: pickup_borough + description: | + NYC borough of the pickup zone, joined from `stg_zones.borough`. + Values: Manhattan, Brooklyn, Queens, Bronx, Staten Island, EWR, + Unknown, NaN, or NULL when `pickup_location_id` did not resolve. + - name: pickup_zone + description: Human-readable pickup-zone name from `stg_zones.zone`. + - name: dropoff_borough + description: NYC borough of the dropoff zone. + - name: dropoff_zone + description: Human-readable dropoff-zone name. diff --git a/include/dbt_project/models/marts/fct_trips.sql b/include/dbt_project/models/marts/fct_trips.sql new file mode 100644 index 0000000..adaaec1 --- /dev/null +++ b/include/dbt_project/models/marts/fct_trips.sql @@ -0,0 +1,20 @@ +{{ config(materialized='table') }} + +select + t.pickup_datetime, + t.dropoff_datetime, + t.fare_amount, + t.tip_amount, + t.trip_distance, + t.tip_pct, + t.fare_per_mile, + t.payment_type_label, + pz.borough as pickup_borough, + pz.zone as pickup_zone, + dz.borough as dropoff_borough, + dz.zone as dropoff_zone +from {{ ref('stg_trips') }} t +left join {{ ref('stg_zones') }} pz + on t.pickup_location_id = pz.location_id +left join {{ ref('stg_zones') }} dz + on t.dropoff_location_id = dz.location_id diff --git a/include/dbt_project/models/marts/fct_trips_docs.md b/include/dbt_project/models/marts/fct_trips_docs.md new file mode 100644 index 0000000..2c1db8d --- /dev/null +++ b/include/dbt_project/models/marts/fct_trips_docs.md @@ -0,0 +1,10 @@ +{% docs trip_grain %} + +One row per completed taxi trip. "Completed" means the TLC submitted the +trip record to the public dataset; cancellations and trips in progress +are not included. Duplicates exist in the source data (roughly 4 rows in +January 2024 where every column is identical) and are kept as-is; see +the `dbt_utils.unique_combination_of_columns` test results for the +current count. + +{% enddocs %} diff --git a/include/dbt_project/models/staging/_sources.yml b/include/dbt_project/models/staging/_sources.yml new file mode 100644 index 0000000..6a55f80 --- /dev/null +++ b/include/dbt_project/models/staging/_sources.yml @@ -0,0 +1,12 @@ +version: 2 + +sources: + - name: nyc_taxi + description: Raw NYC green taxi trip records and zone lookup, loaded in Week 9. + database: team1 + schema: nyc_taxi + tables: + - name: raw_trips + description: One row per green taxi trip for January 2024 (~57K rows). + - name: raw_zones + description: NYC taxi zone lookup (265 rows mapping location IDs to boroughs). diff --git a/include/dbt_project/models/staging/_stg_trips.yml b/include/dbt_project/models/staging/_stg_trips.yml new file mode 100644 index 0000000..656a4fc --- /dev/null +++ b/include/dbt_project/models/staging/_stg_trips.yml @@ -0,0 +1,52 @@ +version: 2 + +models: + - name: stg_trips + description: Cleaned green taxi trips, one row per trip. + tests: + # Chapter 5 teaches this test at the default `error` severity to demonstrate + # how `dbt build` skips downstream models on a test failure. The January 2024 + # raw_trips data contains 4 genuine duplicate rows (a TLC source-data issue), + # so the test always fails. In this reference repo we soften it to `warn` to + # keep CI green while still surfacing the count. When students follow the + # chapter on their own machine, they should leave it at the default. + - dbt_utils.unique_combination_of_columns: + combination_of_columns: [pickup_datetime, dropoff_datetime, pickup_location_id, fare_amount] + config: + severity: warn + columns: + - name: pickup_datetime + description: When the trip started. + tests: + - not_null + - name: pickup_location_id + description: TLC zone ID where the trip started. + tests: + - not_null + - relationships: + to: ref('stg_zones') + field: location_id + config: + severity: warn + - name: payment_type + description: TLC payment code (1-6). + tests: + - not_null: + severity: warn + - accepted_values: + values: [1, 2, 3, 4, 5, 6] + +unit_tests: + - name: payment_type_label_maps_known_codes + model: stg_trips + given: + - input: source('nyc_taxi', 'raw_trips') + rows: + - {payment_type: 1, pickup_datetime: '2024-01-01 08:00:00', pickup_location_id: 100, fare_amount: 10.0, tip_amount: 2.0, trip_distance: 2.0} + - {payment_type: 2, pickup_datetime: '2024-01-01 09:00:00', pickup_location_id: 100, fare_amount: 10.0, tip_amount: 0.0, trip_distance: 2.0} + - {payment_type: 6, pickup_datetime: '2024-01-01 10:00:00', pickup_location_id: 100, fare_amount: 10.0, tip_amount: 0.0, trip_distance: 2.0} + expect: + rows: + - {payment_type: 1, payment_type_label: 'Credit card'} + - {payment_type: 2, payment_type_label: 'Cash'} + - {payment_type: 6, payment_type_label: 'Voided trip'} diff --git a/include/dbt_project/models/staging/_stg_zones.yml b/include/dbt_project/models/staging/_stg_zones.yml new file mode 100644 index 0000000..14f8227 --- /dev/null +++ b/include/dbt_project/models/staging/_stg_zones.yml @@ -0,0 +1,13 @@ +version: 2 + +models: + - name: stg_zones + description: One row per TLC taxi zone (265 zones total). + columns: + - name: location_id + description: TLC zone ID. + tests: + - unique + - not_null + - name: borough + description: NYC borough (Manhattan, Brooklyn, Queens, Bronx, Staten Island, EWR, Unknown). diff --git a/include/dbt_project/models/staging/stg_trips.sql b/include/dbt_project/models/staging/stg_trips.sql new file mode 100644 index 0000000..08cc069 --- /dev/null +++ b/include/dbt_project/models/staging/stg_trips.sql @@ -0,0 +1,36 @@ +{{ config(materialized='view') }} + +{% set payment_types = { + 1: 'Credit card', + 2: 'Cash', + 3: 'No charge', + 4: 'Dispute', + 5: 'Unknown', + 6: 'Voided trip' +} %} + +select + pickup_datetime, + dropoff_datetime, + pickup_location_id, + dropoff_location_id, + fare_amount, + tip_amount, + trip_distance, + payment_type, + case + when fare_amount > 0 then round((tip_amount / fare_amount)::numeric, 4) + else null + end as tip_pct, + case + when trip_distance > 0 then round((fare_amount / trip_distance)::numeric, 4) + else null + end as fare_per_mile, + case payment_type + {% for code, label in payment_types.items() %} + when {{ code }} then '{{ label }}' + {% endfor %} + else 'Other' + end as payment_type_label +from {{ source('nyc_taxi', 'raw_trips') }} +where pickup_location_id is not null diff --git a/include/dbt_project/models/staging/stg_zones.sql b/include/dbt_project/models/staging/stg_zones.sql new file mode 100644 index 0000000..1e98897 --- /dev/null +++ b/include/dbt_project/models/staging/stg_zones.sql @@ -0,0 +1,6 @@ +select + location_id, + borough, + zone, + service_zone +from {{ source('nyc_taxi', 'raw_zones') }} diff --git a/include/dbt_project/package-lock.yml b/include/dbt_project/package-lock.yml new file mode 100644 index 0000000..abff2f9 --- /dev/null +++ b/include/dbt_project/package-lock.yml @@ -0,0 +1,5 @@ +packages: + - name: dbt_utils + package: dbt-labs/dbt_utils + version: 1.3.3 +sha1_hash: e6424ba9e5a22487e47f023803aa4f0411946808 diff --git a/include/dbt_project/packages.yml b/include/dbt_project/packages.yml new file mode 100644 index 0000000..b26b1ff --- /dev/null +++ b/include/dbt_project/packages.yml @@ -0,0 +1,3 @@ +packages: + - package: dbt-labs/dbt_utils + version: [">=1.1.0", "<2.0.0"] diff --git a/include/dbt_project/profiles.yml.example b/include/dbt_project/profiles.yml.example new file mode 100644 index 0000000..ca59c8e --- /dev/null +++ b/include/dbt_project/profiles.yml.example @@ -0,0 +1,13 @@ +nyc_taxi: + target: dev + outputs: + dev: + type: postgres + host: "{{ env_var('PG_HOST', 'hyf-data-pg.postgres.database.azure.com') }}" + port: 5432 + user: "{{ env_var('PG_USER', 'hyfadmin') }}" + password: "{{ env_var('PG_PASSWORD') }}" + dbname: "{{ env_var('PG_DBNAME', 'team1') }}" + schema: "{{ env_var('PG_SCHEMA', 'dev_') }}" # โš ๏ธ Local dev: replace with your first name, lowercase. Airflow: pass PG_SCHEMA via BashOperator env (see README ยง Running from Airflow). + sslmode: require + threads: 4 diff --git a/include/dbt_project/seeds/mutable_zones.csv b/include/dbt_project/seeds/mutable_zones.csv new file mode 100644 index 0000000..0a1a11a --- /dev/null +++ b/include/dbt_project/seeds/mutable_zones.csv @@ -0,0 +1,11 @@ +location_id,borough,zone,service_zone +1,EWR,Newark Airport,EWR +4,Manhattan,Alphabet City,Yellow Zone +13,Manhattan,Battery Park City,Yellow Zone +87,Manhattan,Financial District North,Yellow Zone +132,Queens,JFK Airport,Airports +138,Queens,LaGuardia Airport,Airports +161,Manhattan,Midtown Center,Yellow Zone +236,Manhattan,Upper East Side North,Yellow Zone +264,Unknown,NV,N/A +265,Unknown,Outside of NYC,N/A diff --git a/include/dbt_project/snapshots/mutable_zones_snapshot.sql b/include/dbt_project/snapshots/mutable_zones_snapshot.sql new file mode 100644 index 0000000..f01022e --- /dev/null +++ b/include/dbt_project/snapshots/mutable_zones_snapshot.sql @@ -0,0 +1,12 @@ +{% snapshot mutable_zones_snapshot %} + +{{ config( + target_schema=target.schema ~ '_snapshots', + unique_key='location_id', + strategy='check', + check_cols=['borough', 'zone', 'service_zone'] +) }} + +select * from {{ ref('mutable_zones') }} + +{% endsnapshot %} diff --git a/include/dbt_project/tests/assert_pickup_before_dropoff.sql b/include/dbt_project/tests/assert_pickup_before_dropoff.sql new file mode 100644 index 0000000..5092eba --- /dev/null +++ b/include/dbt_project/tests/assert_pickup_before_dropoff.sql @@ -0,0 +1,6 @@ +select + pickup_datetime, + dropoff_datetime, + pickup_location_id +from {{ ref('stg_trips') }} +where pickup_datetime > dropoff_datetime diff --git a/screenshots/Graph_view.png b/screenshots/Graph_view.png new file mode 100644 index 0000000..b0d66f8 Binary files /dev/null and b/screenshots/Graph_view.png differ diff --git a/screenshots/Grid_Run_view.png b/screenshots/Grid_Run_view.png new file mode 100644 index 0000000..a443179 Binary files /dev/null and b/screenshots/Grid_Run_view.png differ diff --git a/screenshots/Log_snippet.png b/screenshots/Log_snippet.png new file mode 100644 index 0000000..bde2213 Binary files /dev/null and b/screenshots/Log_snippet.png differ diff --git a/screenshots/shared_airflow.png b/screenshots/shared_airflow.png new file mode 100644 index 0000000..85c989c Binary files /dev/null and b/screenshots/shared_airflow.png differ diff --git a/tests/dags/test_dag_example.py b/tests/dags/test_dag_example.py new file mode 100644 index 0000000..6ff3552 --- /dev/null +++ b/tests/dags/test_dag_example.py @@ -0,0 +1,83 @@ +"""Example DAGs test. This test ensures that all Dags have tags, retries set to two, and no import errors. This is an example pytest and may not be fit the context of your DAGs. Feel free to add and remove tests.""" + +import os +import logging +from contextlib import contextmanager +import pytest +from airflow.models import DagBag + + +@contextmanager +def suppress_logging(namespace): + logger = logging.getLogger(namespace) + old_value = logger.disabled + logger.disabled = True + try: + yield + finally: + logger.disabled = old_value + + +def get_import_errors(): + """ + Generate a tuple for import errors in the dag bag + """ + with suppress_logging("airflow"): + dag_bag = DagBag(include_examples=False) + + def strip_path_prefix(path): + return os.path.relpath(path, os.environ.get("AIRFLOW_HOME")) + + # prepend "(None,None)" to ensure that a test object is always created even if it's a no op. + return [(None, None)] + [ + (strip_path_prefix(k), v.strip()) for k, v in dag_bag.import_errors.items() + ] + + +def get_dags(): + """ + Generate a tuple of dag_id, in the DagBag + """ + with suppress_logging("airflow"): + dag_bag = DagBag(include_examples=False) + + def strip_path_prefix(path): + return os.path.relpath(path, os.environ.get("AIRFLOW_HOME")) + + return [(k, v, strip_path_prefix(v.fileloc)) for k, v in dag_bag.dags.items()] + + +@pytest.mark.parametrize( + "rel_path,rv", get_import_errors(), ids=[x[0] for x in get_import_errors()] +) +def test_file_imports(rel_path, rv): + """Test for import errors on a file""" + if rel_path and rv: + raise Exception(f"{rel_path} failed to import with message \n {rv}") + + +APPROVED_TAGS = {} + + +@pytest.mark.parametrize( + "dag_id,dag,fileloc", get_dags(), ids=[x[2] for x in get_dags()] +) +def test_dag_tags(dag_id, dag, fileloc): + """ + test if a DAG is tagged and if those TAGs are in the approved list + """ + assert dag.tags, f"{dag_id} in {fileloc} has no tags" + if APPROVED_TAGS: + assert not set(dag.tags) - APPROVED_TAGS + + +@pytest.mark.parametrize( + "dag_id,dag, fileloc", get_dags(), ids=[x[2] for x in get_dags()] +) +def test_dag_retries(dag_id, dag, fileloc): + """ + test if a DAG has retries set + """ + assert ( + dag.default_args.get("retries", None) >= 2 + ), f"{dag_id} in {fileloc} must have task retries >= 2."