diff --git a/task-1/pyspark_exploration.ipynb.ipynb b/task-1/pyspark_exploration.ipynb.ipynb new file mode 100644 index 0000000..0793b74 --- /dev/null +++ b/task-1/pyspark_exploration.ipynb.ipynb @@ -0,0 +1,534 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "implicitDf": true, + "rowLimit": 10000 + }, + "finishTime": 1785313134403, + "inputWidgets": {}, + "nuid": "384caec0-6ed0-4f8f-ada3-83f05a637596", + "showTitle": false, + "startTime": 1785313104558, + "submitTime": 1785313099108, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "data": { + "text/html": [ + "
trip_count
128202548
" + ] + }, + "metadata": { + "application/vnd.databricks.v1+output": { + "addedWidgets": {}, + "aggData": [], + "aggError": "", + "aggOverflow": false, + "aggSchema": [], + "aggSeriesLimitReached": false, + "aggType": "", + "arguments": {}, + "columnCustomDisplayInfos": {}, + "data": [ + [ + 128202548 + ] + ], + "datasetInfos": [ + { + "name": "_sqldf", + "schema": { + "fields": [ + { + "metadata": {}, + "name": "trip_count", + "nullable": false, + "type": "long" + } + ], + "type": "struct" + }, + "tableIdentifier": null, + "typeStr": "pyspark.sql.connect.dataframe.DataFrame" + } + ], + "dbfsResultPath": null, + "isJsonSchema": true, + "metadata": { + "createTempViewForImplicitDf": true, + "dataframeName": "_sqldf", + "executionCount": 2 + }, + "overflow": false, + "plotOptions": { + "customPlotOptions": {}, + "displayType": "table", + "pivotAggregation": null, + "pivotColumns": null, + "xColumns": null, + "yColumns": null + }, + "removedWidgets": [], + "schema": [ + { + "metadata": "{}", + "name": "trip_count", + "type": "\"long\"" + } + ], + "type": "table" + } + }, + "output_type": "display_data" + } + ], + "source": [ + "%sql\n", + "SELECT count(*) AS trip_count\n", + "FROM hyf.nyc_yellow.raw_trips" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785313332917, + "inputWidgets": {}, + "nuid": "1f6f7fbe-1d11-4ddd-abc9-7869dde15149", + "showTitle": false, + "startTime": 1785313332799, + "submitTime": 1785313331851, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "hello from the cluster\n" + ] + } + ], + "source": [] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785313426630, + "inputWidgets": {}, + "nuid": "e6d89e1c-d831-4bb3-bd1e-13207f65a3fa", + "showTitle": false, + "startTime": 1785313425361, + "submitTime": 1785313424562, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "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", + "\n", + "128202548\n" + ] + } + ], + "source": [ + "trips = spark.read.table(\"hyf.nyc_yellow.raw_trips\")\n", + "zones = spark.read.table(\"hyf.nyc_yellow.raw_zones\")\n", + "\n", + "trips.printSchema()\n", + "print(trips.count()) # action: runs a job; expect roughly 128_000_000" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785313485719, + "inputWidgets": {}, + "nuid": "c2d9e0f6-df63-4df4-bddd-db3a1c06c27b", + "showTitle": false, + "startTime": 1785313482132, + "submitTime": 1785313481394, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "+-------------------+------------+------------+\n", + "| pickup_datetime|total_amount|payment_type|\n", + "+-------------------+------------+------------+\n", + "|2025-06-01 00:05:23| 17.15| 2|\n", + "|2025-06-01 00:26:15| 11.55| 2|\n", + "|2025-06-01 00:02:04| 14.35| 2|\n", + "|2025-06-01 00:44:13| 24.15| 2|\n", + "|2025-06-01 00:10:48| 40.25| 2|\n", + "+-------------------+------------+------------+\n", + "only showing top 5 rows\n" + ] + } + ], + "source": [ + "from pyspark.sql import functions as F\n", + "\n", + "# Transformation: builds a plan, does not run yet\n", + "# F.col(\"payment_type\") is a Column in that plan (not a Python bool).\n", + "# TLC: payment_type == 2 means cash.\n", + "cash_trips = trips.filter(F.col(\"payment_type\") == 2)\n", + "\n", + "# Action: runs the plan and prints a few sample rows\n", + "cash_trips.select(\"pickup_datetime\", \"total_amount\", \"payment_type\").show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785313991578, + "inputWidgets": {}, + "nuid": "f71664b7-682e-4935-becd-2ba7fcb4aa91", + "showTitle": false, + "startTime": 1785313990928, + "submitTime": 1785313990070, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [], + "source": [ + "# Transformations: these describe work, they do not run it yet\n", + "manhattan_trips = (\n", + " trips\n", + " .join(zones, trips.pickup_location_id == zones.location_id)\n", + " .filter(F.col(\"borough\") == \"Manhattan\")\n", + " .groupBy(\"payment_type\")\n", + " .agg(F.count(\"*\").alias(\"trip_count\"))\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785314043227, + "inputWidgets": {}, + "nuid": "88d70550-c916-4f62-bec9-e65db65a04eb", + "showTitle": false, + "startTime": 1785314037434, + "submitTime": 1785314036697, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "+------------+----------+\n", + "|payment_type|trip_count|\n", + "+------------+----------+\n", + "| 0| 14070919|\n", + "| 2| 14211189|\n", + "| 1| 81254108|\n", + "| 3| 618454|\n", + "| 5| 6|\n", + "| 4| 1873813|\n", + "+------------+----------+\n", + "\n", + "[Row(payment_type=0, trip_count=14070919), Row(payment_type=2, trip_count=14211189), Row(payment_type=1, trip_count=81254108), Row(payment_type=3, trip_count=618454), Row(payment_type=5, trip_count=6), Row(payment_type=4, trip_count=1873813)]\n" + ] + } + ], + "source": [ + "# An action: now Spark actually runs the plan across the cluster\n", + "manhattan_trips.show()\n", + "\n", + "# Another action: bring the (small) result back to the driver as a list\n", + "rows = manhattan_trips.collect()\n", + "print(rows)" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785314182901, + "inputWidgets": {}, + "nuid": "41849d72-8c37-4bbd-b0c2-a43a87bc2bef", + "showTitle": false, + "startTime": 1785314175757, + "submitTime": 1785314175118, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [], + "source": [ + "# Write the aggregated result as a managed Delta table\n", + "(\n", + " manhattan_trips\n", + " .write\n", + " .mode(\"overwrite\")\n", + " .saveAsTable(\"hyf.dev_hannahwn.manhattan_payment_summary\")\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785315530181, + "inputWidgets": {}, + "nuid": "05f38cb9-bf16-4d8f-a77e-86eb6e1aee81", + "showTitle": true, + "startTime": 1785315529747, + "submitTime": 1785315527912, + "tableResultSettingsMap": {}, + "title": "Cell 8" + } + }, + "outputs": [], + "source": [ + "from pyspark.sql.functions import count, col\n", + "\n", + "most_trips = (\n", + " trips\n", + " .join(zones, trips.pickup_location_id == zones.location_id, \"left\")\n", + " .groupBy(\"borough\")\n", + " .agg(count(\"*\").alias(\"total_trips\"))\n", + " .orderBy(col(\"total_trips\").desc())\n", + " \n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785315539399, + "inputWidgets": {}, + "nuid": "9a85a8e3-aae3-47ae-ad0f-776817122ce7", + "showTitle": false, + "startTime": 1785315535590, + "submitTime": 1785315534322, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "+-------------+-----------+\n", + "| borough|total_trips|\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", + "[Row(borough='Manhattan', total_trips=112028489), Row(borough='Queens', total_trips=12292035), Row(borough='Brooklyn', total_trips=2648890), Row(borough='Bronx', total_trips=570457), Row(borough='Unknown', total_trips=559802), Row(borough='N/A', total_trips=76659), Row(borough='EWR', total_trips=17103), Row(borough='Staten Island', total_trips=9113)]\n" + ] + } + ], + "source": [ + "most_trips.show()\n", + "\n", + "rows = most_trips.collect()\n", + "print(rows)\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": 0, + "metadata": { + "application/vnd.databricks.v1+cell": { + "cellMetadata": { + "byteLimit": 2048000, + "rowLimit": 10000 + }, + "finishTime": 1785315746627, + "inputWidgets": {}, + "nuid": "940615b3-0fd0-4a32-8c7f-0e95bc95de8f", + "showTitle": false, + "startTime": 1785315744470, + "submitTime": 1785315743764, + "tableResultSettingsMap": {}, + "title": "" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "+------------+------------------+\n", + "|payment_type| avg_total_amount|\n", + "+------------+------------------+\n", + "| 0| 23.48503280062518|\n", + "| 2| 23.74697069224019|\n", + "| 1|29.999458495708435|\n", + "| 3| 9.028713996517435|\n", + "| 5|14.887777777777778|\n", + "| 4|2.1613990066710844|\n", + "+------------+------------------+\n", + "\n", + "[Row(payment_type=0, avg_total_amount=23.48503280062518), Row(payment_type=2, avg_total_amount=23.74697069224019), Row(payment_type=1, avg_total_amount=29.999458495708435), Row(payment_type=3, avg_total_amount=9.028713996517435), Row(payment_type=5, avg_total_amount=14.887777777777778), Row(payment_type=4, avg_total_amount=2.1613990066710844)]\n" + ] + } + ], + "source": [ + "from pyspark.sql.functions import col\n", + "\n", + "avg_total_amount = (\n", + " trips\n", + " .groupBy(\"payment_type\")\n", + " .agg(F.avg(col(\"total_amount\")).alias(\"avg_total_amount\"))\n", + ")\n", + "avg_total_amount.show()\n", + "\n", + "rows = avg_total_amount.collect()\n", + "print(rows)\n", + " " + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "When would I use Pyspark over sql\n", + "1.When i need real time response on the data(streaming)\n", + "2.When i am exploring data like i did above" + ] + } + ], + "metadata": { + "application/vnd.databricks.v1+notebook": { + "computePreferences": null, + "dashboards": [], + "environmentMetadata": { + "base_environment": "", + "environment_version": "5" + }, + "inputWidgetPreferences": null, + "language": "python", + "notebookMetadata": { + "mostRecentlyExecutedCommandWithImplicitDF": { + "commandId": 6944720080495233, + "dataframes": [ + "_sqldf" + ] + }, + "pythonIndentUnit": 4 + }, + "notebookName": "NYC Yellow Trip Aggregations", + "widgets": {} + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 0 +} diff --git a/task-2/.gitignore b/task-2/.gitignore new file mode 100644 index 0000000..9cf96fc --- /dev/null +++ b/task-2/.gitignore @@ -0,0 +1 @@ +profiles.yml diff --git a/task-2/WRITEUP.md b/task-2/WRITEUP.md index f3d1eb3..f2a76c0 100644 --- a/task-2/WRITEUP.md +++ b/task-2/WRITEUP.md @@ -5,15 +5,20 @@ Fill in after running `dbt build --select fct_trips --full-refresh` baseline fol ## First build (full / initial load with --full-refresh) - **Wall-clock time:** +13:19:59 1 of 4 START sql incremental model dev_hannahwn.fct_trips ...................... [RUN] +13:20:55 1 of 4 OK created sql incremental model dev_hannahwn.fct_trips ................. [OK in 55.89s] - **Notes:** (optional: warehouse size, any errors you fixed) ## Second build (incremental rerun) - **Wall-clock time:** + 1 of 4 START sql incremental model dev_hannahwn.fct_trips ...................... [RUN] +13:21:40 1 of 4 OK created sql incremental model dev_hannahwn.fct_trips ................. [OK in 9.84s] ## 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 first build builds from scratch thus taking more time. The second run using is_incremental() and so only new rows are read and updated + `___` @@ -22,3 +27,5 @@ Write two or three sentences in your own words (see the assignment for the conce 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: `___` + + diff --git a/task-2/dbt_project.yml b/task-2/dbt_project.yml new file mode 100644 index 0000000..df663c3 --- /dev/null +++ b/task-2/dbt_project.yml @@ -0,0 +1,29 @@ +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 + +# dbt_project.yml +flags: + use_materialization_v2: true diff --git a/task-2/history_fct_trips.png b/task-2/history_fct_trips.png new file mode 100644 index 0000000..2ce46f4 Binary files /dev/null and b/task-2/history_fct_trips.png differ diff --git a/task-2/models/marts/_fct_daily_borough_stats.yml b/task-2/models/marts/_fct_daily_borough_stats.yml new file mode 100644 index 0000000..919a765 --- /dev/null +++ b/task-2/models/marts/_fct_daily_borough_stats.yml @@ -0,0 +1,30 @@ +version: 2 + +models: + - name: fct_daily_borough_stats + + description: > + + one row per pickup_borough and pickup_date,the source lineage(built from stg_trips + and stg_zones), and atleast one known caveat(we drop rows in staging where pickup location + id is null)) + test: + dbt_utils.unique_combination_of_columns: + combination_of_columns: + - pickup_borough + - pickup_date + columns: + - name: pickup_borough + description: groups trips by the pickup_borough, which is derived from the pickup_location_id and the stg_zones table + + - name: pickup_date + description: groups trips by the pickup_date, which is derived from the pickup_datetime in stg_trips + + - name: trip_count + description: counts the number of trips for each pickup_borough and pickup_date combination + - name: total_fare + description: sums the fare_amount for each pickup_borough and pickup_date combination + - name: avg_tip_pct + description: calculates the average tip percentage for each pickup_borough and pickup_date combination + - name: avg_trip_distance + description: calculates the average trip distance for each pickup_borough and pickup_date combination percentage diff --git a/task-2/models/marts/_fct_trips.yml b/task-2/models/marts/_fct_trips.yml new file mode 100644 index 0000000..92851d5 --- /dev/null +++ b/task-2/models/marts/_fct_trips.yml @@ -0,0 +1,57 @@ +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: + - unique: + config: + severity: warn + - not_null + - 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_daily_borough_stats.sql b/task-2/models/marts/fct_daily_borough_stats.sql new file mode 100644 index 0000000..d3ebf93 --- /dev/null +++ b/task-2/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 AS pickup_borough, + t.pickup_datetime::date AS pickup_date, + COUNT(*) AS trip_count, + SUM(t.fare_amount) AS total_fare, + AVG({{ dbt_utils.safe_divide('t.tip_amount', 't.fare_amount') }}) 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/task-2/models/marts/fct_trips.sql b/task-2/models/marts/fct_trips.sql new file mode 100644 index 0000000..d916c20 --- /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() %} + -- On incremental runs, only process trips newer than what we already have. + where t.pickup_datetime > (select max(pickup_datetime) from {{ this }}) +{% endif %} 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..02762e2 --- /dev/null +++ b/task-2/models/staging/_sources.yml @@ -0,0 +1,10 @@ +version: 2 + +sources: + - name: nyc_taxi + schema: nyc_yellow + tables: + - name: raw_trips + description: One trip per row with detailed information + - name: raw_zones + description: a table for taxi zones and their boroughs diff --git a/task-2/models/staging/_stg_trips.yml b/task-2/models/staging/_stg_trips.yml new file mode 100644 index 0000000..5bc5c14 --- /dev/null +++ b/task-2/models/staging/_stg_trips.yml @@ -0,0 +1,27 @@ +version: 2 + +models: + - name: stg_trips + description: cleaned green taxi trips,one row per trip + columns: + - name: pickup_datetime + description: When trip started. This is used to group trips by day in the mart + tests: + - not_null + + - name: pickup_location_id + description: TLC location ID of the pickup location. This is used to join to the zones table in the mart + tests: + - not_null + - name: fare_amount + description: Total fare amount for the trip + + - name: tip_amount + description: Tip amount for the trip + + - name: trip_distance + description: Distance of the trip + + - name: tip_pct + description: Percentage of the fare that was tipped + \ No newline at end of file diff --git a/task-2/models/staging/_stg_zones.yml b/task-2/models/staging/_stg_zones.yml new file mode 100644 index 0000000..e3b947f --- /dev/null +++ b/task-2/models/staging/_stg_zones.yml @@ -0,0 +1,18 @@ +version: 2 + +models: + - name: stg_zones + description: one row per taxi zone from the + columns: + - name: location_id + description: TLC location ID of the pickup location. This is used to join to the trips table in the mart + tests: + - not_null + - unique + + - name: borough + description: The borough of NYC that the taxi zone is in + tests: + - not_null + + \ 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..bf46767 --- /dev/null +++ b/task-2/models/staging/stg_trips.sql @@ -0,0 +1,38 @@ +-- 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 + {{ dbt_utils.generate_surrogate_key([ + 'pickup_datetime', 'dropoff_datetime', 'pickup_location_id', + 'dropoff_location_id', 'fare_amount', 'total_amount', 'passenger_count' + ]) }} AS trip_id, + pickup_datetime, + dropoff_datetime, + pickup_location_id, + dropoff_location_id, + fare_amount, + tip_amount, + trip_distance, + {{ dbt_utils.safe_divide('tip_amount', 'fare_amount') }} AS tip_pct, + {{ dbt_utils.safe_divide('fare_amount', 'trip_distance') }} AS fare_per_mile, + CASE + WHEN payment_type = 1 THEN 'Credit Card' + WHEN payment_type = 2 THEN 'Cash' + WHEN payment_type = 3 THEN 'No Charge' + WHEN payment_type = 4 THEN 'Dispute' + WHEN payment_type = 5 THEN 'Unknown' + WHEN payment_type = 6 THEN 'Voided Trip' + ELSE NULL + END AS payment_type_label, + CASE + WHEN pickup_datetime IS NULL OR dropoff_datetime IS NULL THEN NULL + ELSE (unix_timestamp(dropoff_datetime) - unix_timestamp(pickup_datetime)) / 60 + END AS trip_duration_minutes +FROM {{ source('nyc_taxi', 'raw_trips') }} +WHERE + pickup_location_id IS NOT NULL + AND fare_amount >= 0 + + + + diff --git a/task-2/models/staging/stg_zones.sql b/task-2/models/staging/stg_zones.sql new file mode 100644 index 0000000..6cb1258 --- /dev/null +++ b/task-2/models/staging/stg_zones.sql @@ -0,0 +1,11 @@ +-- 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, + 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..cf58e99 --- /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: 8b27037b26f3f630c6661194d2470e720c49f6ee diff --git a/task-2/packages.yml b/task-2/packages.yml new file mode 100644 index 0000000..97e3e38 --- /dev/null +++ b/task-2/packages.yml @@ -0,0 +1,8 @@ +# TODO: Task 5 -- declare the dbt-labs/dbt_utils package here, then run +# `dbt deps` to install it. You need it for the compound uniqueness test +# on the mart. See https://hub.getdbt.com/dbt-labs/dbt_utils/latest/ +# for the package block syntax. +packages: + - package: dbt-labs/dbt_utils + version: 1.4.1 + diff --git a/task-3/SCHEDULING.md b/task-3/SCHEDULING.md index 2780686..009b97d 100644 --- a/task-3/SCHEDULING.md +++ b/task-3/SCHEDULING.md @@ -5,6 +5,7 @@ Paste the URL of your successful Job run from the Databricks UI address bar: `___` +https://adb-7405619530719547.7.azuredatabricks.net/jobs/727982853732525/runs/959594307168856?o=7405619530719547 ## Screenshots @@ -21,3 +22,5 @@ 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 schedule Databricks when I want to see job runs in UI and Apache Airflow when I want to focus on dag runs and task logs diff --git a/task-3/screenshots/job_config.png b/task-3/screenshots/job_config.png new file mode 100644 index 0000000..78c9719 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..2eaac8b 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..e4e07ab Binary files /dev/null and b/task-3/screenshots/job_schedule_paused.png differ