diff --git a/.gitignore b/.gitignore index 03c790e..5c864f5 100644 --- a/.gitignore +++ b/.gitignore @@ -172,3 +172,4 @@ dist vite.config.js.timestamp-* vite.config.ts.timestamp-* +task-2/.venv/ diff --git a/AI_ASSIST.md b/AI_ASSIST.md index b134b96..8c8a786 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -4,13 +4,13 @@ Record at least one point where you used an AI coding assistant (ChatGPT, Claude ## Interaction 1 -- **Tool used:** (e.g. ChatGPT / Cursor / Claude) -- **Task / Problem:** (e.g. debugging dbt connection profile / writing PySpark join / configuring Job trigger) +- **Tool used:** ChatGPT +- **Task / Problem:** Resolving an unexpected Git error when switching to the Week 13 dbt branch - **Prompt sent:** - > `___` + > `When I run `git switch --track origin/week-13-ch-4-dbt`, Git stops with the error: `The following untracked working tree files would be overwritten by checkout`. The files are `models/marts/_fct_trips.yml` and `packages.yml`. How can I switch branches without accidentally losing my work?` - **Output provided by AI:** - > `___` + > `The AI explained that Git was protecting untracked local files because files with the same paths already existed in the target branch. It suggested checking whether the files were needed, then either moving them temporarily, adding and committing them, or deleting them before switching branches` - **What I kept, changed, or rejected, and why:** - > `___` + > `I first inspected the conflicting files instead of deleting them immediately. I then removed or moved the local copies that were not needed and switched to the required branch successfully. I kept the safety-first approach because it prevented accidental loss of work` *(Ensure no personal passwords, Databricks tokens, or unapproved credentials are included in prompts or logged outputs.)* diff --git a/task-1/pyspark_exploration.ipynb b/task-1/pyspark_exploration.ipynb new file mode 100644 index 0000000..8fdd479 --- /dev/null +++ b/task-1/pyspark_exploration.ipynb @@ -0,0 +1,170 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785360606181, + "inputWidgets": {}, + "nuid": "ca37ef00-5146-4dc8-b431-8b64c6d68851", + "showTitle": false, + "startTime": 1785360605204, + "submitTime": 1785360604458, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "output_type": "stream", + "text": [ + "root\n |-- vendor_id: long (nullable = true)\n |-- pickup_datetime: timestamp_ntz (nullable = true)\n |-- dropoff_datetime: timestamp_ntz (nullable = true)\n |-- passenger_count: long (nullable = true)\n |-- trip_distance: double (nullable = true)\n |-- rate_code_id: long (nullable = true)\n |-- store_and_fwd_flag: string (nullable = true)\n |-- pickup_location_id: long (nullable = true)\n |-- dropoff_location_id: long (nullable = true)\n |-- payment_type: long (nullable = true)\n |-- fare_amount: double (nullable = true)\n |-- extra: double (nullable = true)\n |-- mta_tax: double (nullable = true)\n |-- tip_amount: double (nullable = true)\n |-- tolls_amount: double (nullable = true)\n |-- improvement_surcharge: double (nullable = true)\n |-- total_amount: double (nullable = true)\n |-- congestion_surcharge: double (nullable = true)\n |-- airport_fee: double (nullable = true)\n\nroot\n |-- location_id: integer (nullable = true)\n |-- borough: string (nullable = true)\n |-- zone: string (nullable = true)\n |-- service_zone: string (nullable = true)\n\n" + ] + } + ], + "source": [ + "trips_df = spark.table(\"hyf.nyc_yellow.raw_trips\")\n", + "zones_df = spark.table(\"hyf.nyc_yellow.raw_zones\")\n", + "\n", + "trips_df.printSchema()\n", + "zones_df.printSchema()" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785360830062, + "inputWidgets": {}, + "nuid": "c27a8b01-dc79-48db-a3a0-3887fdf9b5f2", + "showTitle": false, + "startTime": 1785360827738, + "submitTime": 1785360826877, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "output_type": "stream", + "text": [ + "+-------------+----------+\n| borough|trip_count|\n+-------------+----------+\n| Manhattan| 112028489|\n| Queens| 12292035|\n| Brooklyn| 2648890|\n| Bronx| 570457|\n| Unknown| 559802|\n| N/A| 76659|\n| EWR| 17103|\n|Staten Island| 9113|\n+-------------+----------+\n\n" + ] + } + ], + "source": [ + "from pyspark.sql import functions as F\n", + "\n", + "pickup_borough_counts = (\n", + " trips_df\n", + " .join(\n", + " zones_df,\n", + " trips_df.pickup_location_id == zones_df.location_id,\n", + " \"left\"\n", + " )\n", + " .groupBy(\"borough\")\n", + " .agg(F.count(\"*\").alias(\"trip_count\"))\n", + " .orderBy(F.desc(\"trip_count\"))\n", + ")\n", + "\n", + "pickup_borough_counts.show()" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785361060815, + "inputWidgets": {}, + "nuid": "6e4d9cae-4678-4dab-b7ad-4687e5b10a83", + "showTitle": false, + "startTime": 1785361058542, + "submitTime": 1785361057864, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "output_type": "stream", + "name": "stdout", + "output_type": "stream", + "text": [ + "+------------+------------------+\n|payment_type| avg_total_amount|\n+------------+------------------+\n| 0| 23.48503280062518|\n| 1|29.999458495708435|\n| 2| 23.74697069224019|\n| 3| 9.028713996517435|\n| 4|2.1613990066710844|\n| 5|14.887777777777778|\n+------------+------------------+\n\n" + ] + } + ], + "source": [ + "average_amount_by_payment = (\n", + " trips_df\n", + " .groupBy(\"payment_type\")\n", + " .agg(\n", + " F.avg(\"total_amount\").alias(\"avg_total_amount\")\n", + " )\n", + " .orderBy(\"payment_type\")\n", + ")\n", + "\n", + "average_amount_by_payment.show()" + ] + }, + { + "cell_type": "markdown", + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": {}, + "inputWidgets": {}, + "nuid": "c28a6dfd-4a87-4eca-a5af-c6f35477b34b", + "showTitle": false, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "source": [ + "## PySpark vs dbt SQL\n", + "\n", + "I would use dbt SQL for SQL-based transformations, data marts, and analytics models because it is simpler and easier to maintain. I would use PySpark when I need Python code, large-scale distributed processing, or functionality that cannot be expressed well in SQL, such as machine learning or external API integration." + ] + } + ], + "metadata": { + "application/vnd.databricks.v1+notebook": { + "computePreferences": null, + "dashboards": [], + "environmentMetadata": { + "base_environment": "", + "environment_version": "5" + }, + "inputWidgetPreferences": null, + "language": "python", + "notebookMetadata": { + "pythonIndentUnit": 4 + }, + "notebookName": "pyspark_exploration", + "widgets": {} + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 0 +} \ No newline at end of file diff --git a/task-2/WRITEUP.md b/task-2/WRITEUP.md index f3d1eb3..f7f2cf2 100644 --- a/task-2/WRITEUP.md +++ b/task-2/WRITEUP.md @@ -4,21 +4,21 @@ Fill in after running `dbt build --select fct_trips --full-refresh` baseline fol ## First build (full / initial load with --full-refresh) -- **Wall-clock time:** -- **Notes:** (optional: warehouse size, any errors you fixed) +- **Wall-clock time:** 2m 38.05s +- **Notes:** The full-refresh build completed successfully with `PASS=4 WARN=0 ERROR=0`. Before the successful run, I fixed the Databricks source schema from `public` to `nyc_yellow` and rebuilt the staging views. ## Second build (incremental rerun) -- **Wall-clock time:** +- **Wall-clock time:** 2m 25.82s ## Why was the second run faster? Write two or three sentences in your own words (see the assignment for the concepts you must name): -`___` +`The second run was slightly faster because `is_incremental()` evaluated to true and applied a filter using the maximum `pickup_datetime` already stored in `{{ this }}`. As a result, only rows with a strictly later timestamp were passed to the `MERGE`, instead of rebuilding the whole table. The difference was modest because Databricks still had to evaluate the upstream view, execute the merge, and run the model tests.` ## Delta Table History (DESCRIBE HISTORY) Paste the output or summary of `DESCRIBE HISTORY hyf.dev_yourname.fct_trips` (showing `CREATE OR REPLACE TABLE` and `MERGE` operations) or reference a screenshot: -`___` +`See `delta_history.png`. The Delta table history shows a `CREATE OR REPLACE TABLE AS SELECT` operation for the initial `--full-refresh` build, followed by a `MERGE` operation for the incremental rerun. This confirms that the second build updated the existing Delta table rather than recreating it.` diff --git a/task-2/dbt_project.yml b/task-2/dbt_project.yml new file mode 100644 index 0000000..94eed94 --- /dev/null +++ b/task-2/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/task-2/delta_history.png b/task-2/delta_history.png new file mode 100644 index 0000000..0370d69 Binary files /dev/null and b/task-2/delta_history.png differ diff --git a/task-2/macros/safe_divide.sql b/task-2/macros/safe_divide.sql new file mode 100644 index 0000000..476726b --- /dev/null +++ b/task-2/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 %} diff --git a/task-2/models/marts/_fct_trips.yml b/task-2/models/marts/_fct_trips.yml new file mode 100644 index 0000000..4f82008 --- /dev/null +++ b/task-2/models/marts/_fct_trips.yml @@ -0,0 +1,53 @@ +version: 2 + +models: + - name: fct_trips + description: | + One row per completed NYC yellow taxi trip (2023-2025), with + pickup/dropoff zone attributes folded in (OBT-style mart). Queried + directly by dashboards and ad-hoc analysis. + + **Grain:** one row per trip (`trip_id` surrogate key). + **Source:** `hyf.nyc_yellow.raw_trips` joined to `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: trip_id + description: Surrogate key generated in `stg_trips` for incremental merge. + tests: [not_null, unique] + - 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/task-2/models/marts/fct_trips.sql b/task-2/models/marts/fct_trips.sql new file mode 100644 index 0000000..4c09960 --- /dev/null +++ b/task-2/models/marts/fct_trips.sql @@ -0,0 +1,33 @@ +{{ config( + materialized='incremental', + incremental_strategy='merge', + unique_key='trip_id' +) }} + +select + t.trip_id, + t.pickup_datetime, + t.dropoff_datetime, + t.fare_amount, + t.tip_amount, + t.trip_distance, + t.trip_duration_minutes, + 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 + +{% if is_incremental() %} +where t.pickup_datetime > ( + select max(pickup_datetime) + from {{ this }} +) +{% endif %} \ No newline at end of file diff --git a/task-2/models/marts/fct_trips_docs.md b/task-2/models/marts/fct_trips_docs.md new file mode 100644 index 0000000..2c1db8d --- /dev/null +++ b/task-2/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/task-2/models/staging/_sources.yml b/task-2/models/staging/_sources.yml new file mode 100644 index 0000000..acd53e6 --- /dev/null +++ b/task-2/models/staging/_sources.yml @@ -0,0 +1,12 @@ +version: 2 + +sources: + - name: nyc_taxi + description: Raw NYC yellow taxi trip records and zone lookup. + database: hyf + schema: nyc_yellow + tables: + - name: raw_trips + description: One row per yellow taxi trip. + - name: raw_zones + description: NYC taxi zone lookup. diff --git a/task-2/models/staging/_stg_trips.yml b/task-2/models/staging/_stg_trips.yml new file mode 100644 index 0000000..f461212 --- /dev/null +++ b/task-2/models/staging/_stg_trips.yml @@ -0,0 +1,52 @@ +version: 2 + +models: + - name: stg_trips + description: Cleaned yellow 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. Yellow 2023-2025 adds code 0 (Flex Fare / unknown) on top of the classic 1-6. + tests: + - not_null: + severity: warn + - accepted_values: + values: [0, 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/task-2/models/staging/_stg_zones.yml b/task-2/models/staging/_stg_zones.yml new file mode 100644 index 0000000..14f8227 --- /dev/null +++ b/task-2/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/task-2/models/staging/stg_trips.sql b/task-2/models/staging/stg_trips.sql new file mode 100644 index 0000000..b7341f7 --- /dev/null +++ b/task-2/models/staging/stg_trips.sql @@ -0,0 +1,49 @@ +{{ config(materialized='view') }} + +{% set payment_types = { + 1: 'Credit card', + 2: 'Cash', + 3: 'No charge', + 4: 'Dispute', + 5: 'Unknown', + 6: 'Voided trip' +} %} + +select + {{ dbt_utils.generate_surrogate_key([ + 'vendor_id', 'pickup_datetime', 'dropoff_datetime', 'pickup_location_id', + 'dropoff_location_id', 'fare_amount', 'trip_distance', 'total_amount', 'passenger_count' + ]) }} as trip_id, + 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, 4) + else null + end as tip_pct, + case + when trip_distance > 0 then round(fare_amount / trip_distance, 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, + round((unix_timestamp(dropoff_datetime) - unix_timestamp(pickup_datetime)) / 60.0, 2) as trip_duration_minutes +from {{ source('nyc_taxi', 'raw_trips') }} +where pickup_location_id is not null + and fare_amount >= 0 +qualify row_number() over ( + partition by {{ dbt_utils.generate_surrogate_key([ + 'vendor_id', 'pickup_datetime', 'dropoff_datetime', 'pickup_location_id', + 'dropoff_location_id', 'fare_amount', 'trip_distance', 'total_amount', 'passenger_count' + ]) }} + order by pickup_datetime +) = 1 diff --git a/task-2/models/staging/stg_zones.sql b/task-2/models/staging/stg_zones.sql new file mode 100644 index 0000000..1e98897 --- /dev/null +++ b/task-2/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/task-2/package-lock.yml b/task-2/package-lock.yml new file mode 100644 index 0000000..e397fd7 --- /dev/null +++ b/task-2/package-lock.yml @@ -0,0 +1,5 @@ +packages: + - name: dbt_utils + package: dbt-labs/dbt_utils + version: 1.3.0 +sha1_hash: 226ae69cdfbc9367e2aa2c472b01f99dbce11de0 diff --git a/task-2/packages.yml b/task-2/packages.yml new file mode 100644 index 0000000..39f82d4 --- /dev/null +++ b/task-2/packages.yml @@ -0,0 +1,3 @@ +packages: + - package: dbt-labs/dbt_utils + version: 1.3.0 diff --git a/task-3/SCHEDULING.md b/task-3/SCHEDULING.md index 2780686..da4ab18 100644 --- a/task-3/SCHEDULING.md +++ b/task-3/SCHEDULING.md @@ -4,7 +4,7 @@ Paste the URL of your successful Job run from the Databricks UI address bar: -`___` +`https://adb-7405619530719547.7.azuredatabricks.net/jobs/591256975661340/runs/296764459619110?o=7405619530719547` ## Screenshots @@ -20,4 +20,4 @@ Ensure the following screenshot files exist in `task-3/screenshots/`: Write two to three sentences comparing Databricks Jobs and Apache Airflow in your own words: -`___` +`I would use Databricks Jobs when the pipeline mainly runs inside Databricks and depends on Databricks SQL, Delta tables, notebooks, or dbt. I would choose Apache Airflow when the workflow has to coordinate many different systems, services, and dependencies outside Databricks, because Airflow is more flexible as a general-purpose orchestrator.` diff --git a/task-3/screenshots/job_config.png b/task-3/screenshots/job_config.png new file mode 100644 index 0000000..b03ad88 Binary files /dev/null and b/task-3/screenshots/job_config.png differ diff --git a/task-3/screenshots/job_run_success.png b/task-3/screenshots/job_run_success.png new file mode 100644 index 0000000..704b194 Binary files /dev/null and b/task-3/screenshots/job_run_success.png differ diff --git a/task-3/screenshots/job_schedule_paused.png b/task-3/screenshots/job_schedule_paused.png new file mode 100644 index 0000000..70471ce Binary files /dev/null and b/task-3/screenshots/job_schedule_paused.png differ