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..d650ed4 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -1,12 +1,75 @@ # AI assistance log - - ## Use 1 -**Prompt I sent:** _Replace this section._ +**Prompt I sent:** I ran: + +`astro dev pytest tests/test_dag_integrity.py --args "-v"` + +and both tests failed: + +Running your test suite… +βœ” Project image has been updated +fd6478823d7fccb6939547ace51d22b6bb76e2b49f35db4bb544bcce82e19a85 +Successfully copied 4.3kB (transferred 6.66kB) to astro-pytest:/usr/local/airflow/ +Successfully copied 1.02kB (transferred 3.07kB) to astro-pytest:/usr/local/airflow/ +Successfully copied 8.44kB (transferred 25.1kB) to astro-pytest:/usr/local/airflow/ +Successfully copied 36B (transferred 2.56kB) to astro-pytest:/usr/local/airflow/ +Astro Runtime Version: 3.3-1 +============================= test session starts ============================== +platform linux -- Python 3.14.6, pytest-9.1.1, pluggy-1.6.0 -- /usr/local/bin/python +cachedir: .pytest_cache +rootdir: / +configfile: pyproject.toml +plugins: anyio-4.13.0 +collecting ... collected 2 items + +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-h40bpd4s' +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-sr6_svbo' +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 1.62s ========================= +Error: pytest failed + +The same TypeError occurs in both tests. What is causing it, and what is the minimum change needed to make the starter tests compatible without changing my DAG? -**What the model answered:** _Replace this section._ +**What the model answered:** The installed Airflow version no longer accepts `include_examples` in the `DagBag` constructor. It recommended removing `include_examples=False` from both `DagBag` calls in `tests/test_dag_integrity.py`. -**What I kept, changed, or discarded, and why:** _Replace this section._ +**What I kept, changed, or discarded, and why:** I kept the minimum change and removed only the unsupported argument. I did not change the DAG because the failure came from the test API, not the pipeline. I reran the test and confirmed that both integrity tests passed. I ignored the pytest cache warning because it did not cause the tests to fail. diff --git a/ASSIGNMENT_REPORT.md b/ASSIGNMENT_REPORT.md index 97422f5..0a6d9cd 100644 --- a/ASSIGNMENT_REPORT.md +++ b/ASSIGNMENT_REPORT.md @@ -1,31 +1,80 @@ # Assignment report - - ## Schedule choice and reason -_Replace this section._ +The DAG uses @monthly because the NYC green-taxi source files and ingestion logic are partitioned by month. A daily schedule would unnecessarily process the same monthly file repeatedly. Normal catchup is disabled with catchup=False. ## Task dependency graph -_Replace: describe ingest -> dbt_run -> dbt_test and why order matters._ +The tasks run in this strict order: + +ingest_taxi_month -> dbt_run -> dbt_test + +Ingestion must finish before dbt transforms the data, and the models must build successfully before their tests run. A failed task therefore prevents downstream tasks from processing incomplete data. ## dbt project used -_Replace: your Week 10 project or the class reference?_ +I copied my Week 10 dbt project into include/dbt_project/. Airflow runs it with uvx --python 3.11 so dbt works with the Astro image. ## One debugging case I resolved -_Replace: what failed, how you found the cause in the logs, and the fix._ +The shared Airflow ingestion task failed with permission denied for schema airflow_mohammedalfakih. I found the cause in the task log and reported the shared connection problem to the teacher. The shared Airflow connection was corrected, and the teacher reran the DAG. ingest_taxi_month, dbt_run, and dbt_test then completed successfully. + +## Parameterized execution + +Airflow passes {{ ds }} to ingest_taxi_month. The task takes the year and month from this logical date and uses it to select the parquet URL, filter the rows, and delete and replace only that month in raw_trips. It does not use datetime.now(). + +## Backfill and idempotency + +I ran the initial seven-month backfill with: + +astro dev run backfill create \ + --dag-id taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-07-31 \ + --max-active-runs 1 + +I recorded these row counts: + +2024-01: 56,549 + +2024-02: 53,571 + +2024-03: 57,447 + +2024-04: 56,467 + +2024-05: 60,994 + +2024-06: 54,735 + +2024-07: 51,811 + +I then reran the same range with: + +astro dev run backfill create \ + --dag-id taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-07-31 \ + --max-active-runs 1 \ + --reprocess-behavior completed -## Parameterized runs and backfill +All seven reruns succeeded. The monthly counts remained identical, proving that rerunning a partition does not create duplicate rows. -_Replace: how {{ ds }} / logical date drives the partition; the exact backfill create command you ran (with --max-active-runs 1)._ +## Shared Airflow deployment -## Idempotency row counts (before / after re-run) +Merged deployment PR: https://github.com/lassebenni/c55-shared-airflow/pull/5 -_Replace: paste monthly counts before the re-run, then after. They must match._ +The deployed DAG is named mohammedalfakih_taxi_pipeline and tagged student:mohammedalfakih. -## Shared Airflow deploy proof (if VM online) +Evidence: -_Replace: merged c55-shared-airflow PR URL + path to your shared-UI screenshot in this repo._ +- Successful shared run: `screenshots/shared-airflow-green.png` +- Shared DAG filtered by student tag: `screenshots/shared-airflow-tag-filter.png` +- Local successful run: `screenshots/airflow-ui.png` +- DAG dependency graph: `screenshots/dag-graph.png` +- Task log snippet: `screenshots/ingest-log.png` +- Backfill evidence: `screenshots/backfill-green.png` +- Idempotency rerun: `screenshots/backfill-rerun-green.png` +- Row counts before rerun: `screenshots/idempotency-before.png` +- Row counts after rerun: `screenshots/idempotency-after.png` diff --git a/RUNBOOK.md b/RUNBOOK.md index 84bca1c..331eecc 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -1,22 +1,59 @@ # RUNBOOK - - ## How to trigger the DAG manually -_Replace this section._ +1. Start the local Astro environment: + +```bash +astro dev start +``` + +2. Confirm that `include/dbt_project/profiles.yml` exists. For a fresh clone, + copy it from `include/dbt_project/profiles.yml.example`. +3. Confirm that the `azure_pg` Airflow connection exists. +4. Open the Airflow UI and unpause `taxi_pipeline`. +5. Trigger a historical month whose TLC parquet file exists: + +```bash +astro dev run dags trigger taxi_pipeline \ + --logical-date 2024-08-01T00:00:00+00:00 +``` + +6. In the Airflow UI, confirm that `ingest_taxi_month`, `dbt_run`, and `dbt_test` finish successfully. ## How to run a backfill -_Replace this section._ +Run the seven monthly partitions sequentially to prevent concurrent dbt runs: + +```bash +astro dev run backfill create \ + --dag-id taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-07-31 \ + --max-active-runs 1 +``` + +To rerun completed partitions for an idempotency check: + +```bash +astro dev run backfill create \ + --dag-id taxi_pipeline \ + --from-date 2024-01-01 \ + --to-date 2024-07-31 \ + --max-active-runs 1 \ + --reprocess-behavior completed +``` ## How to inspect task logs -_Replace this section._ +1. Open `taxi_pipeline` in the Airflow UI. +2. Select **Runs** and open the required DAG run. +3. Select the failed or successful task. +4. Open the **Logs** tab. +5. Read the final error message and traceback before retrying or clearing the task. ## Top 3 likely failures and first response -1. _Replace: symptom, first check, fix_ -2. _Replace this section._ -3. _Replace this section._ +1. **TLC parquet file is unavailable** β€” `ingest_taxi_month` reports an HTTP error and retries. Check the logical date in the run. Trigger a historical month with an available parquet file instead of a future or unavailable month. +2. **PostgreSQL connection fails** β€” tasks report that `azure_pg` is missing or authentication failed. Check the `azure_pg` connection in Airflow, including its host, database, login, port, and `sslmode=require`. Never commit its password. +3. **dbt run or test fails** β€” `dbt_run` or `dbt_test` turns red. Inspect its log, confirm that the project exists under `include/dbt_project`, and check `profiles.yml`, source names, and `PG_SCHEMA`. Keep dbt running through the configured `uvx --python 3.11` command. diff --git a/dags/taxi_pipeline.py b/dags/taxi_pipeline.py index 1786d30..9d5fd22 100644 --- a/dags/taxi_pipeline.py +++ b/dags/taxi_pipeline.py @@ -1,64 +1,151 @@ -"""Week 12 assignment starter. - -Turn this into a scheduled, parameterized, retryable pipeline. The task -list, the file-by-file map, and the point breakdown are in README.md; the -full brief is in the Week 12 "Assignment: Orchestrated Pipeline" chapter. - -This starter parses, so `astro dev start` shows the DAG in the UI, but every -task body raises NotImplementedError and the decorator is not configured yet. -Replace the stubs, wire the tasks together, and fill in the decorator. The -autograder fails while any NotImplementedError remains. -""" - import os -from datetime import datetime +from datetime import datetime, timedelta from pathlib import Path +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 +from sqlalchemy import text + -# 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" def find_dbt_dir() -> str: - """Return the mounted dbt project path (Astro vs shared-VM install root).""" + """Return the mounted dbt project path.""" for candidate in ( - "/usr/local/airflow/include/dbt_project", # Astro CLI - "/opt/airflow/include/dbt_project", # shared VM docker-compose + "/usr/local/airflow/include/dbt_project", + "/opt/airflow/include/dbt_project", ): if Path(candidate).is_dir(): return candidate + return "/usr/local/airflow/include/dbt_project" DBT_DIR = find_dbt_dir() +DBT_COMMAND = "uvx --python 3.11 --from dbt-postgres==1.10.2 dbt" + +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, +} @dag( - # Task 1 (see README): configure the decorator β€” schedule, start_date, - # catchup=False, max_active_runs=1, default_args retries, tags. + dag_id="taxi_pipeline", + schedule="@monthly", start_date=datetime(2024, 1, 1), + catchup=False, + max_active_runs=1, + default_args={ + "retries": 2, + "retry_delay": timedelta(minutes=5), + }, + tags=["week12"], ) 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() - - -taxi_pipeline() + def ingest_taxi_month(ds: str) -> int: + """Load one logical month into the personal raw_trips table.""" + year_month = ds[:7] + month_start = pd.Timestamp(f"{year_month}-01") + next_month = month_start + pd.DateOffset(months=1) + + url = f"{TLC_BASE}/green_tripdata_{year_month}.parquet" + trips = pd.read_parquet(url) + + trips = trips[ + [ + "lpep_pickup_datetime", + "PULocationID", + "fare_amount", + "tip_amount", + "trip_distance", + ] + ].rename( + columns={ + "lpep_pickup_datetime": "pickup_datetime", + "PULocationID": "pickup_location_id", + } + ) + + trips = trips[ + (trips["pickup_datetime"] >= month_start) + & (trips["pickup_datetime"] < next_month) + ] + + hook = PostgresHook(postgres_conn_id="azure_pg") + engine = hook.get_sqlalchemy_engine() + + with engine.begin() as connection: + connection.execute( + text(f'CREATE SCHEMA IF NOT EXISTS "{SCHEMA}"') + ) + + trips.head(0).to_sql( + "raw_trips", + connection, + schema=SCHEMA, + if_exists="append", + index=False, + ) + + connection.execute( + text( + f'DELETE FROM "{SCHEMA}"."raw_trips" ' + "WHERE pickup_datetime >= :month_start " + "AND pickup_datetime < :next_month" + ), + { + "month_start": month_start.to_pydatetime(), + "next_month": next_month.to_pydatetime(), + }, + ) + + trips.to_sql( + "raw_trips", + connection, + schema=SCHEMA, + if_exists="append", + index=False, + chunksize=1000, + ) + + return len(trips) + + ingest = ingest_taxi_month(ds="{{ ds }}") + + dbt_run = BashOperator( + task_id="dbt_run", + bash_command=( + f"{DBT_COMMAND} deps " + f"--project-dir {DBT_DIR} " + f"--profiles-dir {DBT_DIR} && " + f"{DBT_COMMAND} 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_COMMAND} test " + f"--project-dir {DBT_DIR} " + f"--profiles-dir {DBT_DIR}" + ), + env=DBT_ENV, + append_env=True, + ) + ingest >> dbt_run >> dbt_test + +taxi_pipeline() \ No newline at end of file diff --git a/include/dbt_project/.gitignore b/include/dbt_project/.gitignore new file mode 100644 index 0000000..fa85403 --- /dev/null +++ b/include/dbt_project/.gitignore @@ -0,0 +1,5 @@ +target/ +dbt_packages/ +logs/ +profiles.yml +.user.yml diff --git a/include/dbt_project/dbt_project.yml b/include/dbt_project/dbt_project.yml new file mode 100644 index 0000000..7f1f451 --- /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/macros/safe_divide.sql b/include/dbt_project/macros/safe_divide.sql new file mode 100644 index 0000000..7cd2e95 --- /dev/null +++ b/include/dbt_project/macros/safe_divide.sql @@ -0,0 +1,7 @@ +-- safe_divide(numerator, denominator) +-- Returns numerator / denominator, or NULL when denominator is 0 or NULL. + + +{% macro safe_divide(numerator, denominator) %} + ({{ numerator }}::numeric / NULLIF({{ denominator }}::numeric, 0)) +{% endmacro %} 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..758b4e2 --- /dev/null +++ b/include/dbt_project/models/marts/_fct_daily_borough_stats.yml @@ -0,0 +1,38 @@ +version: 2 + +models: + - name: fct_daily_borough_stats + description: > + Daily borough-level taxi mart with one row per pickup borough and pickup date. + It is built from stg_trips joined to stg_zones. Rows with missing pickup_location_id + or negative fare_amount are removed in staging, and trips with pickup location IDs + that do not exist in stg_zones are dropped by the inner join. The avg_tip_pct warning + test can return rows where small borough/date groups have unusually high average tip ratios. + data_tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: + - pickup_borough + - pickup_date + + columns: + - name: pickup_borough + description: "Borough or TLC zone label where the trip was picked up, taken from the taxi zone lookup table." + data_tests: + - not_null + + - name: pickup_date + description: "Calendar date of the trip pickup, derived from pickup_datetime. Unit: date." + data_tests: + - not_null + + - name: trip_count + description: "Number of trips picked up in this borough on this date. Unit: trips." + + - name: total_fare + description: "Sum of fare_amount for trips picked up in this borough on this date. Unit: US dollars." + + - name: avg_tip_pct + description: "Average tip ratio for trips in this borough on this date, calculated as tip_amount divided by fare_amount. Unit: ratio, not multiplied by 100." + + - name: avg_trip_distance + description: "Average recorded trip distance for trips picked up in this borough on this date. Unit: 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..3591c63 --- /dev/null +++ b/include/dbt_project/models/marts/fct_daily_borough_stats.sql @@ -0,0 +1,31 @@ +-- 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 AS pickup_borough, + t.pickup_datetime::date AS pickup_date, + COUNT(*) AS trip_count, + SUM(t.fare_amount) AS total_fare, + AVG(t.tip_pct) AS avg_tip_pct, + AVG(t.trip_distance) AS avg_trip_distance + +FROM trips t + +INNER JOIN zones z + ON t.pickup_location_id = z.location_id + +GROUP BY + z.borough, + t.pickup_datetime::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..921298e --- /dev/null +++ b/include/dbt_project/models/staging/_sources.yml @@ -0,0 +1,22 @@ +version: 2 + +sources: + - name: airflow_taxi + description: Raw green taxi trips loaded monthly by the Airflow DAG. + schema: "{{ env_var('PG_SCHEMA') }}" + + tables: + - name: raw_trips + description: > + NYC Green Taxi trip records loaded into the student's personal + airflow schema. Each Airflow run replaces one logical month. + + - name: nyc_taxi + description: Shared NYC taxi reference data loaded during an earlier week. + schema: nyc_taxi + + tables: + - name: raw_zones + description: > + Shared TLC taxi zone lookup containing the location ID and borough + used by the Week 10 dbt models. 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..a27ae1a --- /dev/null +++ b/include/dbt_project/models/staging/_stg_trips.yml @@ -0,0 +1,22 @@ +version: 2 + +models: + - name: stg_trips + description: "Staging model with one row per NYC Green Taxi trip, built from nyc_taxi.raw_trips. It keeps the columns needed for the borough daily mart and filters rows with missing pickup location or negative fare." + columns: + - name: pickup_datetime + description: "Timestamp when the trip started. Used to derive the pickup date in the mart." + data_tests: + - not_null + - name: pickup_location_id + description: "TLC pickup zone identifier. Used to join trips to the zone lookup table." + data_tests: + - not_null + - name: fare_amount + description: "Base fare amount charged for the trip, in US dollars. Negative fares are removed in staging" + - name: tip_amount + description: "Tip amount paid for the trip, in US dollars" + - name: trip_distance + description: "Trip distance recorded by the taxi meter, in miles." + - name: tip_pct + description: "Tip ratio calculated as tip_amount divided by fare_amount. It is NULL when fare_amount is zero." 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..cdd72bd --- /dev/null +++ b/include/dbt_project/models/staging/_stg_zones.yml @@ -0,0 +1,16 @@ +version: 2 + +models: + - name: stg_zones + description: "Staging model with one row per TLC taxi zone, built from nyc_taxi.raw_zones." + columns: + - name: location_id + description: "Unique TLC zone identifier. Used as the join key from trips to zones." + data_tests: + - not_null + - unique + + - name: borough + description: "Borough or zone label for the TLC location, such as Manhattan, Brooklyn, Queens, Bronx, Staten Island, EWR, Unknown, or NaN." + data_tests: + - not_null 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..49719b6 --- /dev/null +++ b/include/dbt_project/models/staging/stg_trips.sql @@ -0,0 +1,17 @@ +-- 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, + pickup_location_id, + fare_amount, + tip_amount, + trip_distance, + {{ safe_divide('tip_amount', 'fare_amount') }} AS tip_pct + +FROM {{ source('airflow_taxi', 'raw_trips') }} + +WHERE + pickup_location_id IS NOT NULL + AND fare_amount >= 0 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..b378c7a --- /dev/null +++ b/include/dbt_project/models/staging/stg_zones.sql @@ -0,0 +1,7 @@ +-- 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..7bf509f --- /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: 8b27037b26f3f630c6661194d2470e720c49f6ee diff --git a/include/dbt_project/packages.yml b/include/dbt_project/packages.yml new file mode 100644 index 0000000..60c4d13 --- /dev/null +++ b/include/dbt_project/packages.yml @@ -0,0 +1,3 @@ +packages: + - package: dbt-labs/dbt_utils + version: 1.4.1 diff --git a/include/dbt_project/profiles.yml.example b/include/dbt_project/profiles.yml.example new file mode 100644 index 0000000..b117161 --- /dev/null +++ b/include/dbt_project/profiles.yml.example @@ -0,0 +1,14 @@ +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') }}" + schema: "{{ env_var('PG_SCHEMA') }}" + sslmode: require + threads: 1 \ No newline at end of file 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..6dd039f --- /dev/null +++ b/include/dbt_project/tests/assert_avg_tip_pct_within_bounds.sql @@ -0,0 +1,14 @@ +{{ config(severity='warn') }} + +-- Business-rule warning test: +-- returns borough/date rows where the average tip ratio is greater than 1. +-- This means the average tip exceeded the fare amount, which is unusual. + +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/screenshots/airflow-ui.png b/screenshots/airflow-ui.png new file mode 100644 index 0000000..71c34ca Binary files /dev/null and b/screenshots/airflow-ui.png differ diff --git a/screenshots/backfill-green.png b/screenshots/backfill-green.png new file mode 100644 index 0000000..25ffaad Binary files /dev/null and b/screenshots/backfill-green.png differ diff --git a/screenshots/backfill-rerun-green.png b/screenshots/backfill-rerun-green.png new file mode 100644 index 0000000..99a9bf8 Binary files /dev/null and b/screenshots/backfill-rerun-green.png differ diff --git a/screenshots/dag-graph.png b/screenshots/dag-graph.png new file mode 100644 index 0000000..d90baf8 Binary files /dev/null and b/screenshots/dag-graph.png differ diff --git a/screenshots/failure-retry.png b/screenshots/failure-retry.png new file mode 100644 index 0000000..9e405d1 Binary files /dev/null and b/screenshots/failure-retry.png differ diff --git a/screenshots/idempotency-after.png b/screenshots/idempotency-after.png new file mode 100644 index 0000000..79a127e Binary files /dev/null and b/screenshots/idempotency-after.png differ diff --git a/screenshots/idempotency-before.png b/screenshots/idempotency-before.png new file mode 100644 index 0000000..45f0aa5 Binary files /dev/null and b/screenshots/idempotency-before.png differ diff --git a/screenshots/ingest-log.png b/screenshots/ingest-log.png new file mode 100644 index 0000000..2a7a1b3 Binary files /dev/null and b/screenshots/ingest-log.png differ diff --git a/screenshots/shared-airflow-green.png b/screenshots/shared-airflow-green.png new file mode 100644 index 0000000..2eb5df5 Binary files /dev/null and b/screenshots/shared-airflow-green.png differ diff --git a/screenshots/shared-airflow-tag-filter.png b/screenshots/shared-airflow-tag-filter.png new file mode 100644 index 0000000..df1bf4d Binary files /dev/null and b/screenshots/shared-airflow-tag-filter.png differ