DP-600 · Runbook

Runbook: Mini Medallion NYC Taxi

Standardní demo dataset NYC TLC: parquet → Delta → medallion v T-SQL → report

📓 Runbook · medallion / batch

Cíle kapitoly

Stáhnout NYC TLC parquet a taxi zone lookup do Lakehouse a uložit jako Delta
Ošetřit INT96 timestamps, aby prošly SQL analytics endpointem
Postavit Bronze, Silver (typed dim + enriched fact) a Gold ve Warehouse
Připravit Power BI report a story arc dema

11.1

Úvod

Microsoft-standardní demo dataset. Clean schema, multi-year, GPS přes Borough mapy. Pattern identický s CHMI/OpenAQ runbookem.

💡

Bonus: MS Fabric má NYC Taxi i v Sample Warehouse (workspace template). Můžeš porovnat svůj výsledek s jejich.

11.2

Setup (jednorázově)

Workspace v Fabricu (např. Taxi_Demo) na F2 / trial
Lakehouse: LH_Taxi
Warehouse: WH_Taxi

(Alternativně: Workspace → New → Sample data → Sample Warehouse "AdventureWorksLT" nebo "NYC Taxi" — ale my si postavíme svoje, ať máš plnou kontrolu nad pipeline.)

11.3

PySpark notebook: stáhnout NYC data + uložit jako Delta

Workspace → New → Notebook
Language: PySpark (Python)
Vlevo Add Lakehouse → Existing → LH_Taxi → Make default

Cell 1 — Stáhnout taxi zone lookup (dim) + 1 měsíc trips (fact)

PYTHON

import urllib.request, os

YEAR = 2024
MONTH = 1  # Leden 2024

URL_TRIPS = f"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{YEAR}-{MONTH:02d}.parquet"
URL_ZONES = "https://d37ci6vzurychx.cloudfront.net/misc/taxi_zone_lookup.csv"

os.makedirs("/lakehouse/default/Files/raw", exist_ok=True)

print("Stahuji yellow_tripdata...")
urllib.request.urlretrieve(URL_TRIPS, f"/lakehouse/default/Files/raw/yellow_tripdata_{YEAR}_{MONTH:02d}.parquet")
print("OK trips")

print("Stahuji taxi_zone_lookup...")
urllib.request.urlretrieve(URL_ZONES, "/lakehouse/default/Files/raw/taxi_zone_lookup.csv")
print("OK zones")

import os
print(f"Trips:  {os.path.getsize('/lakehouse/default/Files/raw/yellow_tripdata_' + str(YEAR) + '_' + f'{MONTH:02d}' + '.parquet')/1024/1024:.1f} MB")
print(f"Zones:  {os.path.getsize('/lakehouse/default/Files/raw/taxi_zone_lookup.csv')} bytes")

Run → očekáváno: trips ~50 MB, zones ~13 KB. Cold start 60-120 s.

Cell 2 — Read + uložit jako Delta tabulky

PYTHON

from pyspark.sql.functions import col, date_format

df_trips = spark.read.parquet(f"Files/raw/yellow_tripdata_{YEAR}_{MONTH:02d}.parquet")

df_trips = (df_trips
    .withColumn("tpep_pickup_datetime", col("tpep_pickup_datetime").cast("timestamp"))
    .withColumn("tpep_dropoff_datetime", col("tpep_dropoff_datetime").cast("timestamp"))
    # String fallback pro případ že ani cast neprojde SQL endpoint
    .withColumn("pickup_str", date_format(col("tpep_pickup_datetime"), "yyyy-MM-dd HH:mm:ss"))
    .withColumn("dropoff_str", date_format(col("tpep_dropoff_datetime"), "yyyy-MM-dd HH:mm:ss"))
)

df_trips.printSchema()
print(f"Trips: {df_trips.count():,} rows")
df_trips.write.format("delta").mode("overwrite").option("overwriteSchema", "true").saveAsTable("trips_raw")

df_zones = (spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("Files/raw/taxi_zone_lookup.csv"))
print(f"Zones: {df_zones.count()} rows")

df_zones.write.format("delta").mode("overwrite").saveAsTable("zones_raw")

print("OK, Delta tabulky trips_raw + zones_raw uloženy")

Po runu ověř v SQL endpointu, jestli tpep_pickup_datetime jde projektovat. Pokud je NULL/missing, použij Cell 3.1 alternativu níže (s pickup_str parsováním přes TRY_CONVERT).

Očekáváno: Trips: ~3,000,000 rows, Zones: 265 rows.

✓

Verify: Lakehouse Explorer → Tables → Refresh → uvidíš trips_raw a zones_raw.

11.4

Wait + refresh SQL endpoint

Workspace → otevři SQL analytics endpoint LH_Taxi
Pravý klik na schema dbo → Refresh
Sanity: SELECT TOP 5 * FROM dbo.zones_raw; — musí vrátit data

11.5

Bronze + Silver + Gold (Warehouse SQL)

V Warehouse WH_Taxi → New SQL query.

Cell 3.1a — Bronze schema (vždy první)

SQL

CREATE SCHEMA bronze;
GO

CREATE TABLE bronze.trips (
    VendorID INT,
    tpep_pickup_datetime DATETIME2(6),
    tpep_dropoff_datetime DATETIME2(6),
    passenger_count INT,
    trip_distance DECIMAL(10,2),
    PULocationID INT,
    DOLocationID INT,
    payment_type INT,
    fare_amount DECIMAL(10,2),
    tip_amount DECIMAL(10,2),
    total_amount DECIMAL(10,2),
    _ingested_at DATETIME2(6)
);

CREATE TABLE bronze.zones (
    LocationID INT,
    Borough VARCHAR(50),
    Zone VARCHAR(100),
    service_zone VARCHAR(50),
    _ingested_at DATETIME2(6)
);
GO

Run. Vznikne prázdný bronze.trips + bronze.zones.

Cell 3.1b — INSERT do bronze.trips

ℹ️

Poznámka:

VYBER VARIANTU dle toho, co vidíš v SQL endpoint pro LH_Taxi.dbo.trips_raw:

Pokud má sloupce tpep_pickup_datetime (timestamp) → Varianta A
Pokud má jen pickup_str (string) → Varianta B (NYC TLC INT96 quirk fallback)

Varianta A — timestamp je čitelný (default):

SQL

INSERT INTO bronze.trips
(VendorID, tpep_pickup_datetime, tpep_dropoff_datetime, passenger_count,
 trip_distance, PULocationID, DOLocationID, payment_type,
 fare_amount, tip_amount, total_amount, _ingested_at)
SELECT
    VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
    CAST(passenger_count AS INT),
    CAST(trip_distance AS DECIMAL(10,2)),
    PULocationID, DOLocationID, payment_type,
    CAST(fare_amount AS DECIMAL(10,2)),
    CAST(tip_amount AS DECIMAL(10,2)),
    CAST(total_amount AS DECIMAL(10,2)),
    SYSDATETIME()
FROM LH_Taxi.dbo.trips_raw;

Varianta B — string fallback (kdyby timestamp nešel):

SQL

INSERT INTO bronze.trips
(VendorID, tpep_pickup_datetime, tpep_dropoff_datetime, passenger_count,
 trip_distance, PULocationID, DOLocationID, payment_type,
 fare_amount, tip_amount, total_amount, _ingested_at)
SELECT
    VendorID,
    TRY_CONVERT(DATETIME2(6), pickup_str),
    TRY_CONVERT(DATETIME2(6), dropoff_str),
    CAST(passenger_count AS INT),
    CAST(trip_distance AS DECIMAL(10,2)),
    PULocationID, DOLocationID, payment_type,
    CAST(fare_amount AS DECIMAL(10,2)),
    CAST(tip_amount AS DECIMAL(10,2)),
    CAST(total_amount AS DECIMAL(10,2)),
    SYSDATETIME()
FROM LH_Taxi.dbo.trips_raw;

Cell 3.1c — INSERT do bronze.zones (vždy stejný)

SQL

INSERT INTO bronze.zones (LocationID, Borough, Zone, service_zone, _ingested_at)
SELECT LocationID, Borough, Zone, service_zone, SYSDATETIME()
FROM LH_Taxi.dbo.zones_raw;

SELECT 'trips' AS tbl, COUNT(*) AS rows FROM bronze.trips
UNION ALL SELECT 'zones', COUNT(*) FROM bronze.zones;

Očekáváno: trips ~3M, zones 265.

Cell 3.2 — Silver (typed dim + cleaned fact)

SQL

CREATE SCHEMA silver;
GO

CREATE TABLE silver.dim_zone (
    LocationID INT NOT NULL,
    Borough VARCHAR(50),
    Zone VARCHAR(100),
    service_zone VARCHAR(50)
);

CREATE TABLE silver.dim_payment (
    payment_type INT NOT NULL,
    payment_name VARCHAR(30)
);

CREATE TABLE silver.fact_trip (
    pickup_ts DATETIME2(6),
    dropoff_ts DATETIME2(6),
    pickup_date DATE,
    pickup_hour SMALLINT,
    pickup_dow SMALLINT,                  -- day of week 1-7
    PULocationID INT,
    DOLocationID INT,
    pickup_borough VARCHAR(50),
    dropoff_borough VARCHAR(50),
    payment_type INT,
    payment_name VARCHAR(30),
    passenger_count INT,
    trip_distance DECIMAL(10,2),
    duration_min DECIMAL(10,2),
    fare_amount DECIMAL(10,2),
    tip_amount DECIMAL(10,2),
    total_amount DECIMAL(10,2),
    fare_per_mile DECIMAL(10,4),
    _silver_loaded_at DATETIME2(6)
);
GO

-- dim_zone
INSERT INTO silver.dim_zone
SELECT LocationID, Borough, Zone, service_zone FROM bronze.zones;

-- dim_payment (TLC číselník)
INSERT INTO silver.dim_payment VALUES
    (1, 'Credit card'),
    (2, 'Cash'),
    (3, 'No charge'),
    (4, 'Dispute'),
    (5, 'Unknown'),
    (6, 'Voided trip');

-- fact_trip s enriched joins
INSERT INTO silver.fact_trip
SELECT
    t.tpep_pickup_datetime,
    t.tpep_dropoff_datetime,
    CAST(t.tpep_pickup_datetime AS DATE),
    DATEPART(HOUR, t.tpep_pickup_datetime),
    DATEPART(WEEKDAY, t.tpep_pickup_datetime),
    t.PULocationID,
    t.DOLocationID,
    pu.Borough,
    dr.Borough,
    t.payment_type,
    p.payment_name,
    t.passenger_count,
    t.trip_distance,
    DATEDIFF(MINUTE, t.tpep_pickup_datetime, t.tpep_dropoff_datetime),
    t.fare_amount,
    t.tip_amount,
    t.total_amount,
    CASE WHEN t.trip_distance > 0
         THEN t.fare_amount / t.trip_distance
         ELSE NULL END,
    SYSDATETIME()
FROM bronze.trips t
LEFT JOIN silver.dim_zone pu ON pu.LocationID = t.PULocationID
LEFT JOIN silver.dim_zone dr ON dr.LocationID = t.DOLocationID
LEFT JOIN silver.dim_payment p ON p.payment_type = t.payment_type
WHERE t.trip_distance > 0
  AND t.fare_amount > 0
  AND t.tpep_pickup_datetime < t.tpep_dropoff_datetime;

SELECT 'dim_zone' AS tbl, COUNT(*) AS rows FROM silver.dim_zone
UNION ALL SELECT 'dim_payment', COUNT(*) FROM silver.dim_payment
UNION ALL SELECT 'fact_trip', COUNT(*) FROM silver.fact_trip;

Očekáváno: dim_zone=265, dim_payment=6, fact_trip ~2.8M (po filtraci nevalidních).

Cell 3.3 — Gold aggregates

SQL

CREATE SCHEMA gold;
GO

-- Gold 1: trips per borough × hour
CREATE TABLE gold.fact_trips_borough_hour (
    pickup_date DATE,
    pickup_hour SMALLINT,
    pickup_borough VARCHAR(50),
    pocet_jizd INT,
    avg_distance DECIMAL(10,2),
    avg_fare DECIMAL(10,2),
    avg_tip DECIMAL(10,2),
    total_revenue DECIMAL(18,2)
);

-- Gold 2: top zones (rev ranking)
CREATE TABLE gold.fact_top_zones (
    PULocationID INT,
    Borough VARCHAR(50),
    Zone VARCHAR(100),
    pocet_jizd INT,
    total_revenue DECIMAL(18,2),
    avg_fare DECIMAL(10,2)
);
GO

INSERT INTO gold.fact_trips_borough_hour
SELECT
    pickup_date, pickup_hour, pickup_borough,
    COUNT(*),
    AVG(trip_distance),
    AVG(fare_amount),
    AVG(tip_amount),
    SUM(total_amount)
FROM silver.fact_trip
WHERE pickup_borough IS NOT NULL
GROUP BY pickup_date, pickup_hour, pickup_borough;

INSERT INTO gold.fact_top_zones
SELECT
    PULocationID, pickup_borough,
    z.Zone,
    COUNT(*) AS pocet_jizd,
    SUM(total_amount) AS total_revenue,
    AVG(fare_amount) AS avg_fare
FROM silver.fact_trip f
JOIN silver.dim_zone z ON z.LocationID = f.PULocationID
WHERE pickup_borough IS NOT NULL
GROUP BY PULocationID, pickup_borough, z.Zone;

SELECT TOP 20 * FROM gold.fact_top_zones ORDER BY total_revenue DESC;

Očekáváno: top zone bude pravděpodobně Manhattan / Times Sq–Theatre District nebo JFK Airport.

11.6

Power BI report

PBI Desktop → Get Data → OneLake catalog → WH_Taxi → Direct Lake → vyber gold.fact_trips_borough_hour + gold.fact_top_zones.

Vizualizace:

Bar chart — top 10 zones podle revenue
Line chart — počet jízd per hodinu (peak hours)
Stacked column — revenue per borough × den v týdnu
Map — jízdy per borough (Manhattan, Brooklyn, Queens, Bronx, Staten Island, EWR)
Slicer: datum, borough, hodina

11.7

Story arc pro demo

"NYC Taxi je standard pro big data demo. 3 miliony jízd v jednom měsíci, 265 zón, 6 typů plateb. Bronze layer mirror parquet/CSV, Silver typing + enrichment + computed columns (duration, fare_per_mile), Gold ready pro BI.

Story: kde se v New Yorku nejvíc vydělává? Které zóny dominují? Manhattan je očekávaný leader, ale peak hours jsou spíš večerní (kdy Manhattan má noční život) než ranní commute."

11.8

Pre-kurz checklist

☐ Workspace + Lakehouse + Warehouse vytvořeny
☐ Cell 1-3 v notebooku otestován dopředu (trips download trvá 30-60 s)
☐ PBI report pre-built s Direct Lake
☐ Sample Warehouse v MS templatu zkoušen pro porovnání

11.9

Troubleshooting

Symptom	Fix
`urllib.error.HTTPError: 403`	CloudFront URL změnila — zkontroluj na https://www.nyc.gov/site/tlc/about/tlc-trip-record-data.page
Trips parquet má jiné sloupce než schema	TLC občas mění strukturu. Spustí `df_trips.printSchema()` a uprav SQL CREATE TABLE.
Bronze INSERT padne na DECIMAL overflow	Některé extremní fare_amount > 10M (chybná data). Přidej `WHERE fare_amount < 1000000` filter.
`Sample warehouse` v MS templatu má jiné názvy tabulek	Tyhle příklady spustí jen na tvém WH_Taxi, MS sample je read-only reference.

11.10

Bonus — multi-month rozšíření

Pro vícero měsíců změn Cell 1:

PYTHON

for MONTH in range(1, 13):  # celý rok 2024
    URL = f"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{YEAR}-{MONTH:02d}.parquet"
    urllib.request.urlretrieve(URL, f"/lakehouse/default/Files/raw/yellow_tripdata_{YEAR}_{MONTH:02d}.parquet")

A pak v Cell 2:

PYTHON

df_trips = spark.read.parquet("Files/raw/yellow_tripdata_2024_*.parquet")  # glob

Dostaneš ~35M řádků. Na F2 dlouhý ETL, ale demo "scaling" je pak hezké.