Pavel T.

.gitignore

@@ -172,3 +172,4 @@ dist
 vite.config.js.timestamp-*
 vite.config.ts.timestamp-*
 
+task-2/.venv/

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
+}

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.`

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

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 %}

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.

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 %}

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 %}

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.