diff --git a/.env.example b/.env.example index 8e4f793..d1237a0 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,5 @@ # Copy to .env (git-ignored). Never commit real values. DATABRICKS_HOST=adb-xxxx.azuredatabricks.net DATABRICKS_HTTP_PATH=/sql/1.0/warehouses/xxxxxxxx -DATABRICKS_TOKEN=dapi...your-token-here +DATABRICKS_TOKEN=your_databricks_token_here DBT_SCHEMA=dev_yourname diff --git a/AI_ASSIST.md b/AI_ASSIST.md index b134b96..842d1b5 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -4,13 +4,15 @@ 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:** Claude +- **Task / Problem:** Debugging PySpark Task 1 aggregation and understanding Spark transformations - **Prompt sent:** - > `___` + > I am working on a Databricks PySpark task. I need to join `hyf.nyc_yellow.raw_trips` with `hyf.nyc_yellow.raw_zones` to find which pickup borough has the most trips and calculate the average `total_amount` per `payment_type`. Help me check my transformation chain and explain why we use `show()` instead of `collect()`. + - **Output provided by AI:** - > `___` + > The AI explained that the join is needed because `pickup_location_id` is only an ID, while `raw_zones` contains the readable borough names. It also explained that `groupBy()` and `agg()` are transformations, while `show()` is an action that triggers Spark execution. + It suggested checking the aggregation code and using `F.avg("total_amount").alias("avg_total_amount")` to create the required output column. - **What I kept, changed, or rejected, and why:** - > `___` + > I kept the explanation of Spark transformations and actions because it helped me understand the execution model. I changed my code by fixing the aggregation step and adding the correct alias for the average column. I verified the final code by running it in Databricks and checking the output with `show()`. *(Ensure no personal passwords, Databricks tokens, or unapproved credentials are included in prompts or logged outputs.)* diff --git a/task-1/pyspark_exploration.py.ipynb b/task-1/pyspark_exploration.py.ipynb new file mode 100644 index 0000000..5768ada --- /dev/null +++ b/task-1/pyspark_exploration.py.ipynb @@ -0,0 +1,157 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785321447930, + "inputWidgets": {}, + "nuid": "beac355e-6975-4e66-b736-6d08e5edb261", + "showTitle": false, + "startTime": 1785321428252, + "submitTime": 1785321428063, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "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", + "+------------+------------------+\n", + "|payment_type| avg_total_amount|\n", + "+------------+------------------+\n", + "| 0|23.485032800615247|\n", + "| 1|29.999458495737827|\n", + "| 2| 23.7469706922074|\n", + "| 3| 9.02871399651705|\n", + "| 4| 2.16139900667104|\n", + "| 5|14.887777777777778|\n", + "+------------+------------------+\n", + "\n" + ] + } + ], + "source": [ + "from pyspark.sql import functions as F\n", + "\n", + "\n", + "trips_df = spark.read.table(\"hyf.nyc_yellow.raw_trips\")\n", + "\n", + "zones_df = spark.read.table(\"hyf.nyc_yellow.raw_zones\")\n", + "\n", + "\n", + "# Task1 Q1\n", + "# Which pickup borough has the most trips?\n", + "\n", + "trips_with_borough_df = (\n", + " trips_df\n", + " .join(\n", + " zones_df,\n", + " trips_df.pickup_location_id == zones_df.location_id,\n", + " \"inner\"\n", + " )\n", + ")\n", + "\n", + "\n", + "borough_trip_counts_df = (\n", + " trips_with_borough_df\n", + " .groupBy(\"borough\")\n", + " .agg(\n", + " F.count(\"*\").alias(\"trip_count\")\n", + " )\n", + " .orderBy(\n", + " F.desc(\"trip_count\")\n", + " )\n", + ")\n", + "\n", + "borough_trip_counts_df.show()\n", + "\n", + "\n", + "# Task1 Q2\n", + "# What is the average total_amount per payment_type?\n", + "\n", + "avg_payment_df = (\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", + "\n", + "avg_payment_df.show()\n", + "\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "I would use PySpark when the data is too large for one machine or when I need complex transformations that require distributed processing.\n", + "\n", + "I would use dbt SQL for analytics transformations, building data models, and testing data in a warehouse because SQL is simpler and easier to maintain.\n", + "\n", + "PySpark is useful for large-scale data processing, while dbt is better for structured analytical workflows." + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": {}, + "inputWidgets": {}, + "nuid": "75a4683c-e7f3-4b9a-94d5-41722f13228e", + "showTitle": false, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [], + "source": [] + } + ], + "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.py", + "widgets": {} + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 0 +} diff --git a/task-2/dbt_project.yml b/task-2/dbt_project.yml new file mode 100644 index 0000000..1da9e46 --- /dev/null +++ b/task-2/dbt_project.yml @@ -0,0 +1,21 @@ +name: 'nyc_taxi' +version: '1.0.0' +profile: 'nyc_taxi' + +model-paths: ["models"] +analysis-paths: ["analyses"] +test-paths: ["tests"] +seed-paths: ["seeds"] +macro-paths: ["macros"] +snapshot-paths: ["snapshots"] + +clean-targets: + - "target" + - "dbt_packages" + +models: + nyc_taxi: + staging: + +materialized: view + marts: + +materialized: table diff --git a/task-2/delta_history.png b/task-2/delta_history.png new file mode 100644 index 0000000..4c8d8ff Binary files /dev/null and b/task-2/delta_history.png differ 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/costumer.sql b/task-2/models/marts/costumer.sql new file mode 100644 index 0000000..0cac67b --- /dev/null +++ b/task-2/models/marts/costumer.sql @@ -0,0 +1,2 @@ +select * +from {{ ref('stg_test') }} \ No newline at end of file diff --git a/task-2/models/marts/fct_trips.sql b/task-2/models/marts/fct_trips.sql new file mode 100644 index 0000000..09fc93f --- /dev/null +++ b/task-2/models/marts/fct_trips.sql @@ -0,0 +1,40 @@ +{{ 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/schema.yml b/task-2/models/schema.yml new file mode 100644 index 0000000..9909c5e --- /dev/null +++ b/task-2/models/schema.yml @@ -0,0 +1,14 @@ +version: 2 + +models: + + - name: payment_type + tests: + - accepted_values: + values: [1, 2, 3, 4, 5, 6] + + - name: pickup_location_id + tests: + - relationships: + to: ref('stg_zones') + field: location_id \ No newline at end of file diff --git a/task-2/models/staging/_sources.yml b/task-2/models/staging/_sources.yml new file mode 100644 index 0000000..1a596f8 --- /dev/null +++ b/task-2/models/staging/_sources.yml @@ -0,0 +1,14 @@ +version: 2 + +version: 2 + +sources: + - name: nyc_taxi + description: Raw NYC yellow taxi trip records and zone lookup on Databricks (Unity Catalog). + database: hyf + schema: nyc_yellow + tables: + - name: raw_trips + - name: raw_zones + + 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_test.sql b/task-2/models/staging/stg_test.sql new file mode 100644 index 0000000..6bb32b4 --- /dev/null +++ b/task-2/models/staging/stg_test.sql @@ -0,0 +1 @@ +select 1 \ No newline at end of file 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..f683b32 --- /dev/null +++ b/task-2/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/task-2/packages.yml b/task-2/packages.yml new file mode 100644 index 0000000..b26b1ff --- /dev/null +++ b/task-2/packages.yml @@ -0,0 +1,3 @@ +packages: + - package: dbt-labs/dbt_utils + version: [">=1.1.0", "<2.0.0"] diff --git a/task-3/SCHEDULING.md b/task-3/SCHEDULING.md index 2780686..854a119 100644 --- a/task-3/SCHEDULING.md +++ b/task-3/SCHEDULING.md @@ -3,7 +3,7 @@ ## Databricks Job Run URL Paste the URL of your successful Job run from the Databricks UI address bar: - +https://adb-7405619530719547.7.azuredatabricks.net/jobs/402767432860412/runs/733351880002668?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: -`___` +Databricks Jobs is a good choice when the pipeline runs mainly inside the Databricks platform, for example running dbt models, notebooks, or Spark jobs. It provides simple scheduling and monitoring without extra infrastructure. Apache Airflow is better when orchestrating complex workflows across multiple systems, such as APIs, databases, cloud services, and different data platforms. \ No newline at end of file diff --git a/task-3/screenshots/job_config.png b/task-3/screenshots/job_config.png new file mode 100644 index 0000000..e494465 Binary files /dev/null and b/task-3/screenshots/job_config.png differ diff --git a/task-3/screenshots/job_run_success1.png b/task-3/screenshots/job_run_success1.png new file mode 100644 index 0000000..d985b15 Binary files /dev/null and b/task-3/screenshots/job_run_success1.png differ diff --git a/task-3/screenshots/job_run_success2.png b/task-3/screenshots/job_run_success2.png new file mode 100644 index 0000000..1940456 Binary files /dev/null and b/task-3/screenshots/job_run_success2.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..ca92b12 Binary files /dev/null and b/task-3/screenshots/job_schedule_paused.png differ