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..2b62f5a 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -1,12 +1,285 @@ # AI assistance log +## Use 1 - +**Prompt I sent:** Astro Runtime Version: 3.3-1 -## Use 1 +tests/test_dag_integrity.py::test_no_import_errors FAILED                [ 50%] +tests/test_dag_integrity.py::test_every_dag_has_tags FAILED              [100%] + +=================================== FAILURES =================================== +____________________________ test_no_import_errors _____________________________ + +    def test_no_import_errors(): +        """Every .py in dags/ must import cleanly.""" +>       dag_bag = DagBag(dag_folder="dags", include_examples=False) +                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +E       TypeError: DagBag.__init__() got an unexpected keyword argument 'include_examples' + +tests/test_dag_integrity.py:20: TypeError +___________________________ test_every_dag_has_tags ____________________________ + +    def test_every_dag_has_tags(): +        """Light convention check so DAGs are discoverable via the UI tag filter.""" +>       dag_bag = DagBag(dag_folder="dags", include_examples=False) +                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +E       TypeError: DagBag.__init__() got an unexpected keyword argument 'include_examples' + +tests/test_dag_integrity.py:28: TypeError +=============================== warnings summary =============================== +../lib/python3.14/site-packages/_pytest/cacheprovider.py:469 +  /usr/local/lib/python3.14/site-packages/_pytest/cacheprovider.py:469: PytestCacheWarning: could not create cache path /.pytest_cache/v/cache/nodeids: [Errno 13] Permission denied: '/pytest-cache-files-7646t09h' +    config.cache.set("cache/nodeids", sorted(self.cached_nodeids)) + +../lib/python3.14/site-packages/_pytest/cacheprovider.py:423 +  /usr/local/lib/python3.14/site-packages/_pytest/cacheprovider.py:423: PytestCacheWarning: could not create cache path /.pytest_cache/v/cache/lastfailed: [Errno 13] Permission denied: '/pytest-cache-files-g152s9jj' +    config.cache.set("cache/lastfailed", self.lastfailed) + +-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html +=========================== short test summary info ============================ +FAILED tests/test_dag_integrity.py::test_no_import_errors - TypeError: DagBag... +FAILED tests/test_dag_integrity.py::test_every_dag_has_tags - TypeError: DagB... +======================== 2 failed, 2 warnings in 3.03s ========================= +Error: pytest failed + +**What the model answered:** The root cause of your test failure is that Airflow 3 removed the include_examples argument from DagBag.__init__(). + +Because Astro Runtime 3.3 is built on top of Airflow 3.x, passing include_examples=False to DagBag() causes Python to throw a TypeError. +How to Fix It +1. Update tests/test_dag_integrity.py + +Remove include_examples=False from your DagBag initializations: +Python + +# Before +dag_bag = DagBag(dag_folder="dags", include_examples=False) + +# After +dag_bag = DagBag(dag_folder="dags") + +In Airflow 3, example DAGs are disabled by default or loaded via configuration (AIRFLOW__CORE__LOAD_EXAMPLES=False), making the parameter unnecessary in the class constructor. +Secondary Issue: Pytest Cache Permission Warning + +Your logs also showed this warning: +Plaintext + +PytestCacheWarning: could not create cache path /.pytest_cache/v/cache/nodeids: [Errno 13] Permission denied: '/pytest-cache-files-7646t09h' + +Why it happens + +When running inside a container, pytest is trying to write .pytest_cache to the root directory / where the container user doesn't have write permissions. +Quick Fixes + + Option A: Add -o cache_dir=/tmp/.pytest_cache when running pytest locally: + Bash + + astro dev pytest -- -o cache_dir=/tmp/.pytest_cache + + Option B: Update your pyproject.toml file so pytest defaults to a writable directory: + Ini, TOML + + [tool.pytest.ini_options] + cache_dir = "/tmp/.pytest_cache" + + Summary: Update lines 20 and 28 in tests/test_dag_integrity.py to instantiate DagBag(dag_folder="dags") without include_examples, and your test suite will pass cleanly on Airflow 3 / Astro Runtime 3.x! + +**What I kept, changed, or discarded, and why:** removed include examples, ignored cache permession, since its only a warning and wouldnt affect passing or failing + + + + + + + + + + + +second instance> **Prompt I sent:** + +i am creating a runbook # RUNBOOK + + +## How to trigger the DAG manually + +astro dev start +az login +PGURL +hover to url then navigate then trigger + +## How to run a backfill + + +astro dev run backfill create \ + --dag-id bader_taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-03-31 \ + --max-active-runs 1 + + +## How to inspect task logs + +using url + +## Top 3 likely failures and first response + +1. forgotten az login to fetch pg config +2. astro config set postgres.port 5433 on port conflict +3. forgotten to add PG_URL using astro dev run connections add azure_pg --conn-uri "$PG_URL" +create runbook.md using information and pictures similiar to https://hub.hackyourfuture.nl/Airflow-Fundamentals-de0ff4a4371946ee9945b61b6f47d3eb +using only directions to run without learning material. +Keep in mind (dag id = bader_taxi_pipeline), (tags: [student:bader](http://c55-data-week-12.localhost:6563/dags?tags=student%3Abader), [taxi](http://c55-data-week-12.localhost:6563/dags?tags=taxi), [week12](http://c55-data-week-12.localhost:6563/dags?tags=week12)) +airflow pipeline name is bader_taxi_pipeline +populate all steps example: (## How to trigger the DAG manually + +astro dev start +az login +PGURL +hover to url then navigate then trigger follow all steps and explain all steps based on URL given from hackyourfuture) + +**What the model answered:** # RUNBOOK — bader_taxi_pipeline + +**DAG ID:** `bader_taxi_pipeline` +**Tags:** `student:bader`, `taxi`, `week12` +**Airflow UI base URL:** `http://c55-data-week-12.localhost:6563` + +This document is written so that another student can operate this DAG end‑to‑end +without reading the Python source. Every step is a command to run or a click +to make — no background theory. + +--- + +## 0. Prerequisites (one-time setup) + +- Astro CLI installed and working (`astro version`) +- Azure CLI installed (`az version`) +- Access to the Azure subscription/project this course uses +- This repo cloned locally, with your terminal `cd`'d into the project folder + +--- + +## 1. How to trigger the DAG manually + +1. **Start the local Airflow environment** + ```bash + astro dev start + ``` + Wait until the terminal reports the webserver and scheduler are up. + +2. **Authenticate with Azure** + ```bash + az login + ``` + This opens a browser window — log in with your course/organization account. + Without this step, the next command (fetching the Postgres connection + string) will fail with an authentication error. + +3. **Fetch the Postgres connection string (`PG_URL`)** + ```bash + export PG_URL=$(az keyvault secret show \ + --name pg-connection-string \ + --vault-name \ + --query value -o tsv) + ``` + > ⚠️ Replace `` (and the secret name if different) with + > the actual Key Vault used in this course. Ask your instructor/team if + > you don't know it — this is the one value in this runbook that depends + > on your specific environment. + +4. **Register the connection in Airflow** + ```bash + astro dev run connections add azure_pg --conn-uri "$PG_URL" + ``` + This tells Airflow how to reach the Azure Postgres database under the + connection ID `azure_pg`, which the DAG's tasks reference internally. + +5. **Open the Airflow UI** + Navigate your browser to: + ``` + http://c55-data-week-12.localhost:6563/dags?tags=student%3Abader + ``` + You can also filter by the `taxi` or `week12` tags using the same + pattern (`?tags=taxi`, `?tags=week12`), or combine them in the UI's tag + filter bar. + +6. **Find and trigger the DAG** + - Locate `bader_taxi_pipeline` in the DAG list. + - Hover over the row — a **Trigger** (▶) button appears on the right side + of the row. + - Click **Trigger DAG** (the play icon), then confirm in the dialog that + pops up (leave config JSON empty unless you specifically need to pass + parameters). + - The DAG run will appear at the top of the run history with a status of + `running`. + +--- + +## 2. How to run a backfill + +Use this when you need to (re)process a historical date range instead of +triggering a single run. + +```bash +astro dev run backfill create \ + --dag-id bader_taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-03-31 \ + --max-active-runs 1 +``` + +Notes: +- `--from-date` / `--to-date` define the inclusive date range to backfill. + Adjust these to the range you actually need. +- `--max-active-runs 1` forces the backfill to run one date at a time. This + is slower but avoids overloading the shared Postgres database — don't + raise it unless you know the DB can handle concurrent runs. +- You still need `az login` and the `azure_pg` connection (steps 2–4 above) + set up before running a backfill. + +--- + +## 3. How to inspect task logs + +1. Go to the DAG's run history: + ``` + http://c55-data-week-12.localhost:6563/dags?tags=student%3Abader + ``` +2. Click on `bader_taxi_pipeline` to open the DAG detail view. +3. Click on the specific **DAG run** (identified by its execution date/run + ID) you want to inspect. +4. In the Grid or Graph view, click on the individual **task instance** + (the colored square/box for that task). +5. In the panel that opens, click **Logs**. This shows the full stdout/ + stderr for that task's execution, including tracebacks on failure. + +If the UI is unreachable, you can also stream logs directly from the CLI: +```bash +astro dev logs --scheduler +astro dev logs --webserver +``` + +--- + +## 4. Top 3 likely failures and first response + +| # | Symptom | First check / fix | +|---|---------|--------------------| +| 1 | `astro dev start` fails with `port is already allocated` (Postgres port conflict) | Another local Postgres/container is using the default port. Set a different port and restart: `astro config set postgres.port 5433`, then `astro dev start` again. | +| 2 | Task fails with `Connection azure_pg not found` / DB connection error | You forgot to register the connection. Re-run steps 3–4 above: fetch `PG_URL` (requires `az login` first), then `astro dev run connections add azure_pg --conn-uri "$PG_URL"`. | +| 3 | `az`/Azure calls fail with `Please run 'az login'` or a 401/403 error when fetching `PG_URL` | You forgot to authenticate before fetching config. Run `az login`, confirm with `az account show`, then retry step 3. | + +--- + +## 5. Quick reference -**Prompt I sent:** _Replace this section._ +```bash +# Full manual trigger sequence, start to finish +astro dev start +az login +export PG_URL=$(az keyvault secret show --name pg-connection-string --vault-name --query value -o tsv) +astro dev run connections add azure_pg --conn-uri "$PG_URL" +# then open http://c55-data-week-12.localhost:6563/dags?tags=student%3Abader and trigger via the UI +``` -**What the model answered:** _Replace this section._ -**What I kept, changed, or discarded, and why:** _Replace this section._ +**What I kept, changed, or discarded, and why:** +edited links and infor and populated runbook.md accordingly after few changes diff --git a/ASSIGNMENT_REPORT.md b/ASSIGNMENT_REPORT.md index 97422f5..55b8b38 100644 --- a/ASSIGNMENT_REPORT.md +++ b/ASSIGNMENT_REPORT.md @@ -4,28 +4,54 @@ ## Schedule choice and reason -_Replace this section._ +{{ds}} logical date. because choosing timenow i couldnt manually trigger runs on airflow choosing a specifec past date. ## Task dependency graph -_Replace: describe ingest -> dbt_run -> dbt_test and why order matters._ +ingest_taxi_month() >> dbt_run >> dbt_test +chain matter because its linear and not exponential it follows that exact order in one line. -## dbt project used -_Replace: your Week 10 project or the class reference?_ +## dbt project used +Week 10 project ## One debugging case I resolved -_Replace: what failed, how you found the cause in the logs, and the fix._ +placed dbt project in the wrong place and project couldnt find them. +found out from the logs in the dbt run. +Task failed with exceptionAirflowException: Bash command failed. The command returned a non-zero exit code 2. + + + +## deploy proof +https://github.com/lassebenni/c55-shared-airflow/pull/6 +image is in images folder; +![Shared deploy](/images/deploy.png) + + + -## Parameterized runs and backfill -_Replace: how {{ ds }} / logical date drives the partition; the exact backfill create command you ran (with --max-active-runs 1)._ +backfill proof ![alt text](/images/backfill.png) -## Idempotency row counts (before / after re-run) +Row count after and before double backfill is the same. -_Replace: paste monthly counts before the re-run, then after. They must match._ +SELECT to_char(lpep_pickup_datetime, 'YYYY-MM') AS month, count(*) +FROM airflow_bader.raw_trips +GROUP BY 1 ORDER BY -## Shared Airflow deploy proof (if VM online) +![Shared deploy](/images/rowcount.png) -_Replace: merged c55-shared-airflow PR URL + path to your shared-UI screenshot in this repo._ +month |count| +-------+-----+ +2008-12| 2| +2009-01| 2| +2023-12| 4| +2024-01|56555| +2024-02|53578| +2024-03|57451| +2024-04|56472| +2024-05|61007| +2024-06|54736| +2024-07|51811| +2024-08| 125| diff --git a/RUNBOOK.md b/RUNBOOK.md index 84bca1c..2a69b4f 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -1,22 +1,142 @@ -# RUNBOOK +# RUNBOOK — bader_taxi_pipeline - +**DAG ID:** `bader_taxi_pipeline` +**Tags:** `student:bader`, `taxi`, `week12` +**Airflow UI base URL:** `http://c55-data-week-12.localhost:6563` -## How to trigger the DAG manually +This document is written so that another student can operate this DAG end‑to‑end +without reading the Python source. Every step is a command to run or a click +to make — no background theory. -_Replace this section._ +--- -## How to run a backfill +## 0. Prerequisites (one-time setup) -_Replace this section._ +- Astro CLI installed and working (`astro version`) +- Azure CLI installed (`az version`) +- Access to the Azure subscription/project this course uses +- This repo cloned locally, with your terminal `cd`'d into the project folder -## How to inspect task logs +--- -_Replace this section._ +## 1. How to trigger the DAG manually -## Top 3 likely failures and first response +1. **Start the local Airflow environment** + ```bash + astro dev start + ``` + Wait until the terminal reports the webserver and scheduler are up. -1. _Replace: symptom, first check, fix_ -2. _Replace this section._ -3. _Replace this section._ +2. **Authenticate with Azure** + ```bash + az login + ``` + This opens a browser window — log in with your course/organization account. + Without this step, the next command (fetching the Postgres connection + string) will fail with an authentication error. + +3. **Fetch the Postgres connection string (`PG_URL`)** + ```bash + export PG_URL=$(az keyvault secret show \ + --name pg-connection-string \ + --vault-name \ + --query value -o tsv) + ``` + > ⚠️ Replace `` (and the secret name if different) with + > the actual Key Vault used in this course. Ask your instructor/team if + > you don't know it — this is the one value in this runbook that depends + > on your specific environment. + +4. **Register the connection in Airflow** + ```bash + astro dev run connections add azure_pg --conn-uri "$PG_URL" + ``` + This tells Airflow how to reach the Azure Postgres database under the + connection ID `azure_pg`, which the DAG's tasks reference internally. + +5. **Open the Airflow UI** + Navigate your browser to: + ``` + http://c55-data-week-12.localhost:6563/dags?tags=student%3Abader + ``` + You can also filter by the `taxi` or `week12` tags using the same + pattern (`?tags=taxi`, `?tags=week12`), or combine them in the UI's tag + filter bar. + +6. **Find and trigger the DAG** + - Locate `bader_taxi_pipeline` in the DAG list. + - Hover over the row — a **Trigger** (▶) button appears on the right side + of the row. + - Click **Trigger DAG** (the play icon), then confirm in the dialog that + pops up (leave config JSON empty unless you specifically need to pass + parameters). + - The DAG run will appear at the top of the run history with a status of + `running`. + +--- + +## 2. How to run a backfill + +Use this when you need to (re)process a historical date range instead of +triggering a single run. + +```bash +astro dev run backfill create \ + --dag-id bader_taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-03-31 \ + --max-active-runs 1 +``` + +Notes: +- `--from-date` / `--to-date` define the inclusive date range to backfill. + Adjust these to the range you actually need. +- `--max-active-runs 1` forces the backfill to run one date at a time. This + is slower but avoids overloading the shared Postgres database — don't + raise it unless you know the DB can handle concurrent runs. +- You still need `az login` and the `azure_pg` connection (steps 2–4 above) + set up before running a backfill. + +--- + +## 3. How to inspect task logs + +1. Go to the DAG's run history: + ``` http://c55-data-week-12.localhost:6563/dags ``` +2. Click on `bader_taxi_pipeline` to open the DAG detail view. +or directly to ``` http://c55-data-week-12.localhost:6563/dags/bader_taxi_pipeline ``` +3. Click on the specific **DAG run** (identified by its execution date/run + ID) you want to inspect. +4. In the Grid or Graph view, click on the individual **task instance** + (the colored square/box for that task). +5. In the panel that opens, click **Logs**. This shows the full stdout/ + stderr for that task's execution, including tracebacks on failure. + +If the UI is unreachable, you can also stream logs directly from the CLI: +```bash +astro dev logs --scheduler +astro dev logs --webserver +``` + +--- + +## 4. Top 3 likely failures and first response + +| # | Symptom | First check / fix | +|---|---------|--------------------| +| 1 | Error: error building, (re)creating or starting project containers: Error response from daemon: ports are not available: exposing port TCP 127.0.0.1:5432 -> 127.0.0.1:0: /forwards/expose returned unexpected status: 500: `astro config set postgres.port 5433`, then `astro dev start` again. | [5433 port can be changed if busy] +| 2 | `az`/Azure calls fail with `Please run 'az login'` or a 401/403 error when fetching `PG_URL` | You forgot to authenticate before fetching config. Run `az login`, confirm with `az account show`, then retry step 3. | +| 3 | ask fails with `Connection azure_pg not found` / DB connection error | You forgot to register the connection. Re-run steps 3–4 above: fetch `PG_URL` (requires `az login` first), then `astro dev run connections add azure_pg --conn-uri "$PG_URL"`. | [Make sure the PGURL is populated from UI by clicking left panel Admin> connections OR http://c55-data-week-12.localhost:6563/connections] +T +--- + +## 5. Quick reference + +```bash +# Full manual trigger sequence, start to finish +astro dev start +az login +export PG_URL=$(az keyvault secret show --name pg-connection-string --vault-name --query value -o tsv) +astro dev run connections add azure_pg --conn-uri "$PG_URL" +# then open http://c55-data-week-12.localhost:6563/dags/bader_taxi_pipeline and trigger via the UI +``` 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..5e22dd9 100644 --- a/dags/taxi_pipeline.py +++ b/dags/taxi_pipeline.py @@ -10,24 +10,40 @@ autograder fails while any NotImplementedError remains. """ +import io + import os from datetime import datetime from pathlib import Path +from pydoc import text +import pandas as pd +from airflow.providers.postgres.hooks.postgres import PostgresHook +from airflow.providers.standard.operators.bash import BashOperator + -from airflow.sdk import dag, task +import requests + +from airflow.sdk import dag, get_current_context, task # 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. STUDENT = os.environ.get("AIRFLOW_STUDENT") or Path(__file__).parent.name SCHEMA = f"airflow_{STUDENT}" TLC_BASE = "https://d37ci6vzurychx.cloudfront.net/trip-data" +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 find_dbt_dir() -> str: """Return the mounted dbt project path (Astro vs shared-VM install root).""" for candidate in ( "/usr/local/airflow/include/dbt_project", # Astro CLI - "/opt/airflow/include/dbt_project", # shared VM docker-compose + "/opt/airflow/include/dbt_project", # shared VM docker-compose ): if Path(candidate).is_dir(): return candidate @@ -37,28 +53,94 @@ def find_dbt_dir() -> str: DBT_DIR = find_dbt_dir() +def parquet_url_for(ds: str) -> str: + """Return the TLC green-taxi parquet URL for a logical date. + + Pure function, extracted from ``download_taxi_month`` so it can be + unit-tested without an Airflow runtime (see Chapter 6). + + >>> parquet_url_for("2024-01-01") + 'https://d37ci6vzurychx.cloudfront.net/trip-data/green_tripdata_2024-01.parquet' + """ + 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. + + TaskFlow's ``ds: str`` auto-injection works for *scheduled* runs + (``logical_date`` is set to the interval boundary) but breaks for + *manual* triggers in Airflow 3, where ``logical_date`` defaults to + ``None``. Reading through ``get_current_context()`` with a + ``run_after`` fallback is the form that works in both modes. + """ + ctx = get_current_context() + dr = ctx["dag_run"] + dt = dr.logical_date or dr.run_after + return dt.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. + dag_id="bader_taxi_pipeline", + schedule="@monthly", start_date=datetime(2024, 1, 1), + catchup=False, + max_active_runs=1, # serialize: concurrent dbt runs collide on __dbt_backup relations + default_args={ + "retries": 2 + }, # retry transient failures twice before marking the task failed + tags=["week12", "taxi", "student:bader"], ) 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 - # 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() + ds = _ds_from_context() + year_month = ds[:7] + resp = requests.get(parquet_url_for(ds), timeout=60) + resp.raise_for_status() + df = pd.read_parquet(io.BytesIO(resp.content)) + + hook = PostgresHook(postgres_conn_id="azure_pg") + engine = hook.get_sqlalchemy_engine() + 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, + ) + 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,), + ) + df.to_sql( + "raw_trips", + engine, + schema=SCHEMA, + if_exists="append", + index=False, + ) + return len(df) + + dbt_run = BashOperator( + task_id="dbt_run", + bash_command=f"uvx --python 3.11 --from 'dbt-core==1.10.*' --with 'dbt-postgres==1.10.*' dbt deps --project-dir {DBT_DIR} --profiles-dir {DBT_DIR} && uvx --python 3.11 --from 'dbt-core==1.10.*' --with 'dbt-postgres==1.10.*' 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"uvx --python 3.11 --from 'dbt-core==1.10.*' --with 'dbt-postgres==1.10.*' 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() diff --git a/image.png b/image.png new file mode 100644 index 0000000..f6b12d6 Binary files /dev/null and b/image.png differ diff --git a/images/backfill.png b/images/backfill.png new file mode 100644 index 0000000..f6b12d6 Binary files /dev/null and b/images/backfill.png differ diff --git a/images/bader_taxi_pipeline-graph.png b/images/bader_taxi_pipeline-graph.png new file mode 100644 index 0000000..ce85545 Binary files /dev/null and b/images/bader_taxi_pipeline-graph.png differ diff --git a/images/deploy.png b/images/deploy.png new file mode 100644 index 0000000..e095a90 Binary files /dev/null and b/images/deploy.png differ diff --git a/images/rowcount.png b/images/rowcount.png new file mode 100644 index 0000000..b9156a1 Binary files /dev/null and b/images/rowcount.png differ diff --git a/include/dbt_project/.devcontainer/devcontainer.json b/include/dbt_project/.devcontainer/devcontainer.json new file mode 100644 index 0000000..2ac7336 --- /dev/null +++ b/include/dbt_project/.devcontainer/devcontainer.json @@ -0,0 +1,10 @@ +{ + "name": "HYF Week 10: dbt assignment", + "image": "mcr.microsoft.com/devcontainers/python:3.11", + "postCreateCommand": "pip install --user dbt-core dbt-postgres", + "customizations": { + "vscode": { + "extensions": ["innoverio.vscode-dbt-power-user"] + } + } +} diff --git a/include/dbt_project/.github/workflows/dbt-parse.yml b/include/dbt_project/.github/workflows/dbt-parse.yml new file mode 100644 index 0000000..df03e6f --- /dev/null +++ b/include/dbt_project/.github/workflows/dbt-parse.yml @@ -0,0 +1,44 @@ +name: dbt parse + +# Validates project structure and Jinja without a database connection. +# dbt_project.yml ships with the template, so this always runs. +# The static autograder (.hyf/test.sh) runs separately via grade-assignment.yml. + +on: + pull_request: + branches: + - main + +jobs: + parse: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Check for dbt_project.yml + id: check + run: | + if [[ -f dbt_project.yml ]]; then + echo "present=true" >> "$GITHUB_OUTPUT" + else + echo "present=false" >> "$GITHUB_OUTPUT" + echo "::warning::dbt_project.yml not found -- it ships with the template; did you delete it? Skipping parse." + fi + - uses: actions/setup-python@v5 + if: steps.check.outputs.present == 'true' + with: + python-version: "3.11" + - name: Install dbt + if: steps.check.outputs.present == 'true' + run: pip install dbt-core dbt-postgres + - name: dbt deps + parse (no database connection needed) + if: steps.check.outputs.present == 'true' + env: + PG_HOST: localhost + PG_USER: ci + PG_PASSWORD: ci-dummy + PG_DBNAME: postgres + run: | + cp profiles.yml.example profiles.yml + sed -i 's/dev_/dev_ci/' profiles.yml + dbt deps + dbt parse diff --git a/include/dbt_project/.gitignore b/include/dbt_project/.gitignore new file mode 100644 index 0000000..5f1af3c --- /dev/null +++ b/include/dbt_project/.gitignore @@ -0,0 +1,165 @@ +# dbt +target/ +dbt_packages/ +logs/ +profiles.yml +.user.yml + +# System files +.DS_Store +Thumbs.db +[Dd]esktop.ini + +# hyf +.hyf/score.json + +# Editor and IDE settings +.vscode/ +.idea/ +*.iml +*.code-workspace +*.sublime-project +*.sublime-workspace +.history/ +.ionide/ + +# Logs +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* +lerna-debug.log* + +# Diagnostic reports (https://nodejs.org/api/report.html) +report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json + +# Runtime data +pids +*.pid +*.seed +*.pid.lock + +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov + +# Coverage directory used by tools like istanbul +coverage +*.lcov + +# nyc test coverage +.nyc_output + +# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files) +.grunt + +# Bower dependency directory (https://bower.io/) +bower_components + +# node-waf configuration +.lock-wscript + +# Compiled binary addons (https://nodejs.org/api/addons.html) +build/Release + +# Dependency directories +node_modules/ +jspm_packages/ + +# Snowpack dependency directory (https://snowpack.dev/) +web_modules/ + +# TypeScript cache +*.tsbuildinfo + +# Optional npm cache directory +.npm + +# Optional eslint cache +.eslintcache + +# Optional stylelint cache +.stylelintcache + +# Optional REPL history +.node_repl_history + +# Output of 'npm pack' +*.tgz + +# Yarn Integrity file +.yarn-integrity + +# dotenv environment variable files +.env +.env.* +!.env.example + +# parcel-bundler cache (https://parceljs.org/) +.cache +.parcel-cache + +# Next.js build output +.next +out + +# Nuxt.js build / generate output +.nuxt +dist + +# Gatsby files +.cache/ +# Comment in the public line in if your project uses Gatsby and not Next.js +# https://nextjs.org/blog/next-9-1#public-directory-support +# public + +# vuepress build output +.vuepress/dist + +# vuepress v2.x temp and cache directory +.temp +.cache + +# Sveltekit cache directory +.svelte-kit/ + +# vitepress build output +**/.vitepress/dist + +# vitepress cache directory +**/.vitepress/cache + +# Docusaurus cache and generated files +.docusaurus + +# Serverless directories +.serverless/ + +# FuseBox cache +.fusebox/ + +# DynamoDB Local files +.dynamodb/ + +# Firebase cache directory +.firebase/ + +# TernJS port file +.tern-port + +# Stores VSCode versions used for testing VSCode extensions +.vscode-test + +# yarn v3 +.pnp.* +.yarn/* +!.yarn/patches +!.yarn/plugins +!.yarn/releases +!.yarn/sdks +!.yarn/versions + +# Vite logs files +vite.config.js.timestamp-* +vite.config.ts.timestamp-* + diff --git a/include/dbt_project/.hyf/README.md b/include/dbt_project/.hyf/README.md new file mode 100644 index 0000000..38f1a4d --- /dev/null +++ b/include/dbt_project/.hyf/README.md @@ -0,0 +1,15 @@ +# Auto grade tool + +## How it works +1. The auto grade tool runs the `test.sh` script located in this directory. +2. `test.sh` should write to a file named `score.json` with following JSON format: + ```json + { + "score": , + "passingScore": , + "pass": "" + } + ``` + All scores are out of 100. It is up to the assignment to determine how to calculate the score. +3. The auto grade runs via a github action on PR creation and updates the PR with the score. + diff --git a/include/dbt_project/dbt_project.yml b/include/dbt_project/dbt_project.yml new file mode 100644 index 0000000..94eed94 --- /dev/null +++ b/include/dbt_project/dbt_project.yml @@ -0,0 +1,25 @@ +name: 'nyc_taxi_borough_daily' +version: '1.0.0' +config-version: 2 + +# This project connects to the profile of the same name in profiles.yml. +profile: 'nyc_taxi_borough_daily' + +model-paths: ["models"] +macro-paths: ["macros"] +test-paths: ["tests"] + +target-path: "target" +clean-targets: + - "target" + - "dbt_packages" + +# Folder-level materialization defaults. Staging models stay as views (cheap, +# always fresh); the mart is built as a table (queried repeatedly by the +# dashboard). You can override per model with {{ config(materialized='...') }}. +models: + nyc_taxi_borough_daily: + staging: + +materialized: view + marts: + +materialized: table diff --git a/include/dbt_project/docs/.gitkeep b/include/dbt_project/docs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/include/dbt_project/macros/safe_divide.sql b/include/dbt_project/macros/safe_divide.sql new file mode 100644 index 0000000..8695fe0 --- /dev/null +++ b/include/dbt_project/macros/safe_divide.sql @@ -0,0 +1,6 @@ +{% macro safe_divide(numerator, denominator) %} + case + when {{ denominator }} > 0 then round(({{ numerator }} / {{ denominator }})::numeric, 4) + else null + end +{% endmacro %} \ No newline at end of file diff --git a/include/dbt_project/models/marts/_fct_daily_borough_stats.yml b/include/dbt_project/models/marts/_fct_daily_borough_stats.yml new file mode 100644 index 0000000..716bc4e --- /dev/null +++ b/include/dbt_project/models/marts/_fct_daily_borough_stats.yml @@ -0,0 +1,36 @@ +version: 2 + +models: + - name: fct_daily_borough_stats + 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`." + # TODO: Task 5 -- add the compound uniqueness test on the mart's primary + # key (pickup_borough, pickup_date). You need the dbt_utils package for + # this: declare it in packages.yml and run `dbt deps` first. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: + - pickup_borough + - pickup_date + severity: warn + columns: + - name: pickup_borough + description: "TLC borough where the trip started" + - name: pickup_date + description: "the day the trip started (date only, no time)" + - name: trip_count + description: "total trips that started in this borough on this day" + - name: total_fare + description: "total revenue from trips that started in this borough on this day in USD" + - name: avg_tip_pct + description: "average tip percentage for trips that started in this borough on this day, expressed as a decimal (e.g. 0.15 = 15%)" + - name: avg_trip_distance + description: "average distance of trips that started in this borough on this day, in miles" diff --git a/include/dbt_project/models/marts/fct_daily_borough_stats.sql b/include/dbt_project/models/marts/fct_daily_borough_stats.sql new file mode 100644 index 0000000..7c56b93 --- /dev/null +++ b/include/dbt_project/models/marts/fct_daily_borough_stats.sql @@ -0,0 +1,28 @@ +-- Mart: daily borough trip statistics. +-- Grain: one row per (pickup_borough, pickup_date). +-- Used to answer: trip volume, revenue, tipping behaviour, and distance profile +-- per borough per day for January 2024. + +WITH trips AS ( + SELECT * + FROM {{ ref('stg_trips') }} +), + +zones AS ( + SELECT * + FROM {{ ref('stg_zones') }} +) + +SELECT + + z.borough::text AS pickup_borough, + t.pickup_datetime::date AS pickup_date, + COUNT(*) AS trip_count, + SUM(t.fare_amount)::numeric(10,2) AS total_fare, + AVG(t.tip_pct)::numeric(10,2) AS avg_tip_pct, + AVG(t.trip_distance)::numeric(10,2) AS avg_trip_distance + +FROM trips t +INNER JOIN zones z + ON t.pickup_location_id = z.location_id +GROUP BY pickup_borough, pickup_date diff --git a/include/dbt_project/models/staging/_sources.yml b/include/dbt_project/models/staging/_sources.yml new file mode 100644 index 0000000..e5bfc3c --- /dev/null +++ b/include/dbt_project/models/staging/_sources.yml @@ -0,0 +1,10 @@ +version: 2 + +sources: + - name: nyc_taxi + 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..5a947ef --- /dev/null +++ b/include/dbt_project/models/staging/_stg_trips.yml @@ -0,0 +1,30 @@ +version: 2 + +models: + - name: stg_trips + description: "Cleaned green taxi trips, one row per trip. This reads from the raw_trips source." + 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: fare_amount + description: "cost of the trip in USD" + - name: tip_amount + description: "tip paid for the trip" + + - name: trip_distance + description: "distance of the trip in miles" + + - name: tip_pct + description: "percentage of tip relative to trip cost" + 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..13c591d --- /dev/null +++ b/include/dbt_project/models/staging/_stg_zones.yml @@ -0,0 +1,15 @@ +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). + tests: + - not_null \ No newline at end of file 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..b613728 --- /dev/null +++ b/include/dbt_project/models/staging/stg_trips.sql @@ -0,0 +1,16 @@ +-- Staging model: one row per NYC green taxi trip (January 2024). +-- Renames source columns, adds derived columns, and filters bad rows. +-- Downstream: fct_daily_borough_stats joins this to stg_zones. + +SELECT + pickup_datetime, + dropoff_datetime, + pickup_location_id, + dropoff_location_id, + fare_amount, + tip_amount, + trip_distance, + {{ safe_divide('tip_amount', 'fare_amount') }} as tip_pct +FROM {{ source('nyc_taxi', 'raw_trips') }} +WHERE pickup_location_id IS NOT NULL + AND fare_amount >= 0 \ No newline at end of file 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..a44c667 --- /dev/null +++ b/include/dbt_project/models/staging/stg_zones.sql @@ -0,0 +1,8 @@ +-- Staging model: one row per TLC zone (265 zones). +-- Exposes location_id and borough for use as a lookup in the mart. + +SELECT + location_id, + borough + +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..1ce78fc --- /dev/null +++ b/include/dbt_project/package-lock.yml @@ -0,0 +1,5 @@ +packages: + - name: dbt_utils + package: dbt-labs/dbt_utils + version: 1.4.1 +sha1_hash: e6424ba9e5a22487e47f023803aa4f0411946808 diff --git a/include/dbt_project/packages.yml b/include/dbt_project/packages.yml new file mode 100644 index 0000000..71fdb1a --- /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"] \ No newline at end of file diff --git a/include/dbt_project/profiles.yml.example b/include/dbt_project/profiles.yml.example new file mode 100644 index 0000000..f5102c0 --- /dev/null +++ b/include/dbt_project/profiles.yml.example @@ -0,0 +1,17 @@ +nyc_taxi_borough_daily: + target: dev + outputs: + dev: + type: postgres + host: "{{ env_var('PG_HOST') }}" + port: 5432 + user: "{{ env_var('PG_USER') }}" + password: "{{ env_var('PG_PASSWORD') }}" + dbname: "{{ env_var('PG_DBNAME', 'postgres') }}" + schema: "dev_bader" + threads: 1 + sslmode: require + +# Copy this file to profiles.yml (same directory), fill in your name, and ensure +# PG_HOST, PG_USER, PG_PASSWORD, and PG_DBNAME are set in your environment. +# profiles.yml is git-ignored — never commit it with a real password. diff --git a/include/dbt_project/reports/answers.md b/include/dbt_project/reports/answers.md new file mode 100644 index 0000000..dea1e8c --- /dev/null +++ b/include/dbt_project/reports/answers.md @@ -0,0 +1,79 @@ +# Business Question Answers + +Queries run against `dev_.fct_daily_borough_stats`. + +## Q1: Highest total `total_fare` across the whole loaded dataset + +**SQL:** + +```sql +-- TODO: query fct_daily_borough_stats grouped by pickup_borough, sum total_fare, order DESC +select +pickup_borough, +SUM(total_fare) as total_fares +from dev_bader.fct_daily_borough_stats +group by pickup_borough +order by total_fares desc +``` + +**Result:** Manhatten 493955.62 + +**Interpretation:** highest total revenue per borough is manhatten + +--- + +## Q2: Day with the highest overall `trip_count` + +**SQL:** + +```sql +-- TODO: query fct_daily_borough_stats grouped by pickup_date, sum trip_count, order DESC LIMIT 1 +select +pickup_date, +SUM(trip_count) as total_trips +from dev_bader.fct_daily_borough_stats +group by pickup_date +order by total_trips desc +limit 1 +``` + +**Result:** 17-01-2024 + +**Interpretation:** highest trips count day is 17-01-2024 + +--- + +## Q3: Highest `avg_tip_pct` for any (borough, day) combination + +**SQL:** + +```sql +select * +from dev_bader.fct_daily_borough_stats +order by avg_tip_pct desc +limit 5 +``` + +**Result:** unknown borough + +**Interpretation:** shows highest average tip precentages per borough/date. any average tips precentages above 1 do show, because we didnt add where tip_pct > 1 when building mart and we used a warn assert test instead which keeps them in the data + +--- + +## Q4: Median daily `trip_count` for Manhattan vs Brooklyn + +**SQL:** + +```sql +-- TODO: use percentile_cont(0.5) WITHIN GROUP (ORDER BY trip_count) filtered by borough +SELECT + pickup_borough, + percentile_cont(0.5) WITHIN GROUP (ORDER BY trip_count) AS median_daily +FROM dev_bader.fct_daily_borough_stats +WHERE pickup_borough IN ('Manhattan', 'Brooklyn') +GROUP BY pickup_borough; +``` + +**Result:** 248 for brooklyn and 1169.5 for manhatten + +**Interpretation:** manhatten has almost 4.7 times the trip count volume as brooklyn diff --git a/include/dbt_project/tests/assert_avg_tip_pct_within_bounds.sql b/include/dbt_project/tests/assert_avg_tip_pct_within_bounds.sql new file mode 100644 index 0000000..054736d --- /dev/null +++ b/include/dbt_project/tests/assert_avg_tip_pct_within_bounds.sql @@ -0,0 +1,22 @@ +-- Singular test: flag (borough, date) combinations where avg_tip_pct > 1. +-- A tip_pct > 1 means the average tip exceeded the total fare for that cell, +-- which is unusual and almost always indicates a small-sample bucket (e.g. the +-- Unknown borough) where a few high-tip outliers dominate the average. +-- +-- Set this test to WARN severity by adding an inline config at the top of this +-- file (below these comments): {{ config(severity='warn') }} +-- That keeps a few expected Unknown-borough rows from blocking `dbt build`, while +-- still surfacing them for your reports/answers.md write-up (the rubric requires +-- documenting this finding). Do NOT set a project-level test severity in +-- dbt_project.yml: that would also downgrade your not_null and +-- unique_combination primary-key tests, which you want to stay at ERROR. +-- +-- The test passes (no WARN) when zero rows are returned; any returned rows are flagged. + +-- TODO: write the SELECT here. +-- Query {{ ref('fct_daily_borough_stats') }} and return rows where avg_tip_pct > 1. + +{{ config(severity='warn') }} +select pickup_borough, pickup_date, avg_tip_pct +from {{ ref('fct_daily_borough_stats') }} +where avg_tip_pct > 1 \ No newline at end of file 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." diff --git a/tests/test_dag_integrity.py b/tests/test_dag_integrity.py index 83295ff..9f49bf4 100644 --- a/tests/test_dag_integrity.py +++ b/tests/test_dag_integrity.py @@ -17,16 +17,12 @@ def test_no_import_errors(): """Every .py in dags/ must import cleanly.""" - # Airflow 3 DagBag no longer accepts include_examples. dag_bag = DagBag(dag_folder="dags") - assert dag_bag.import_errors == {}, ( - f"DAG import errors: {dag_bag.import_errors}" - ) + assert dag_bag.import_errors == {}, f"DAG import errors: {dag_bag.import_errors}" def test_every_dag_has_tags(): """Light convention check so DAGs are discoverable via the UI tag filter.""" - # Airflow 3 DagBag no longer accepts include_examples. dag_bag = DagBag(dag_folder="dags") for dag_id, dag in dag_bag.dags.items(): assert dag.tags, f"DAG {dag_id} is missing tags"