Add NYC taxi medallion Lakeflow declarative pipeline (serverless DAB)

[dvannoy]

Adds pipelines/nyctaxi-medallion/ — a Databricks Asset Bundle containing a serverless Lakeflow Spark Declarative Pipeline that builds a Bronze → Silver → Gold medallion from samples.nyctaxi.trips into catalog main across nyctaxi_bronze / nyctaxi_silver / nyctaxi_gold.

Source schema actually found on samples.nyctaxi.trips

databricks experimental aitools tools discover-schema samples.nyctaxi.trips returned exactly six columns, 21,932 rows, zero nulls in every column:

column	type
tpep_pickup_datetime	timestamp
tpep_dropoff_datetime	timestamp
trip_distance	double
fare_amount	double
pickup_zip	int
dropoff_zip	int

There is no primary key, no vendor id, no passenger count, no lat/lon, no PULocationID/DOLocationID, and no borough or zone column. Two consequences drove the design:

1. No natural key → the trip identity is sha2(concat_ws('|', pickup_ts, dropoff_ts, trip_distance, fare_amount, pickup_zip, dropoff_zip), 256), materialised as trip_key in bronze. Verified unique for this snapshot: COUNT(DISTINCT trip_key) = COUNT(*) = 21932. This uniqueness is empirical, not guaranteed by construction — the six columns carry no identity of their own, so two genuinely identical trips (same pickup and dropoff second, same distance, same fare, same ZIP pair) would collide and share a trip_key. It holds for this dataset; a different snapshot would need re-checking.
2. Location is ZIP-only → dim_zone is at ZIP grain, not taxi-zone grain.

Cross-borough proxy

pickup_zip / dropoff_zip are the only location columns, so the proxy is a ZIP-range → borough lookup (borough_of() in 03_silver_trips_enriched.py). NYC ZIP codes are borough-contiguous, which makes the range test a faithful, haversine-free borough assignment:

ZIP range	borough
10001–10282	Manhattan
10301–10314	Staten Island
10451–10475	Bronx
11201–11256	Brooklyn
11001–11109, 11351–11499, 11690–11697	Queens
11501–11599, 11701–11980	Long Island
10500–10999	Westchester
6000–6999 / 7000–8999	Connecticut / New Jersey
otherwise	Unknown

is_cross_borough is pickup_borough != dropoff_borough, but NULL rather than false when either end is unplaceable — 'Unknown' != 'Unknown' would otherwise read as a confident "not cross-borough" when the truth is "we don't know". A companion is_region_known boolean says whether the proxy applied at both ends at all.

The Westchester range (10500–10999) was added during review: it places the 7 trips whose dropoff ZIP is Yonkers (10703/10705/10710), Mount Vernon (10550), White Plains (10601) or Pelham (10803) and which previously fell through to Unknown. No Unknown regions remain in this dataset — is_region_known is true for all 21,847 rows and is_cross_borough is NULL for none of them, so the three-valued logic is defensive rather than load-bearing here.

is_cross_borough fires on 2,644 of 21,847 silver trips (12.1%) — a plausible rate for Manhattan-dominated 2016 yellow-cab data. Boroughs are also carried as columns on both silver trips and dim_zone, so the proxy is inspectable rather than a black-box flag.

Dataset type per layer, and why

Bronze — nyctaxi_bronze.trips_raw: streaming table

A streaming table is the only Lakeflow dataset type that is append-only by construction: each update processes only source rows that arrived since the last update (exactly-once, pipeline-managed checkpoint) and appends them. That is what makes _ingested_at = current_timestamp() meaningful — it records when we first saw a row. A materialized view would recompute the full result set on every refresh and reset _ingested_at on all 21,932 rows, which defeats the purpose of a raw layer. The source is a Delta table, read with spark.readStream.table(...), so an append-only batch source is streamed incrementally with no extra machinery.

Silver — nyctaxi_silver.trips_enriched: streaming table

Bronze is append-only and every transformation here is row-local: no aggregation, no join, no cross-row state. Streaming bronze's append feed means an update touches only new rows, and expectations are evaluated once per row at ingest instead of on every full recompute. An MV would rescan and re-derive all 21,847 rows each refresh for no benefit.

• DROP via Lakeflow expectations — @dp.expect_all_or_drop({"positive_fare_amount": "fare_amount > 0", "positive_trip_distance": "trip_distance > 0"}). 85 source rows violate (10 with fare_amount <= 0, 76 with trip_distance <= 0, 1 overlapping), which is exactly the 21,932 → 21,847 delta.
• FLAG as a plain derived column, not an expectation — is_invalid_time_order = tpep_dropoff_datetime < tpep_pickup_datetime. Rows are kept. This sample happens to contain 0 reversed-timestamp rows (confirmed directly against the source), so the column is present and correct with 0 rows flagged.
• Derived: trip_duration_minutes, fare_per_mile (via try_divide — see below), pickup_borough/dropoff_borough, is_cross_borough, is_region_known, time_of_day (morning 6–11 / afternoon 12–17 / evening 18–22 / night 23–5), rider_id FK. Liquid-clustered on (pickup_date, pickup_zip).

Silver — nyctaxi_silver.riders: materialized view

A generated 500-row reference dimension. There is no incoming stream to read and nothing to append, so a streaming table is meaningless. Every value is a pure deterministic function of the row ordinal, which makes a full recompute idempotent — precisely MV semantics.

Built with deterministic hashing rather than Faker: xxhash64(row_ordinal, <per-column salt>) indexes fixed literal lists. This keeps output stable across refreshes with no extra dependency on serverless pipeline compute. No real PII: names/email-domains/home-ZIPs come from literal lists, every email domain is an example.* reserved domain, every phone number uses the reserved 555 fictional-use exchange, and card_last4 is a hash digit string.

Gold — all five datasets: materialized views

• dim_date, dim_zone — these are DISTINCT (i.e. aggregating/deduplicating) queries over the whole silver feed. Streaming tables are append-only and cannot produce a deduplicated result; MVs can, and incrementally refresh as silver grows. dim_zone unions pickup and dropoff ZIPs before deduplicating.
• dim_time_of_day — a fixed 4-row lookup built from literals. No stream exists. MV rather than a view so BI joins hit a stored table instead of re-evaluating the generator per query.
• dim_rider — a projection over a bounded 500-row source that is itself regenerated in place, not appended to; an MV keeps the two in lockstep.
• fact_trips — the fact is the join point of the star and must stay referentially consistent with four dimensions that are themselves recomputed MVs. An MV refreshes alongside them, so a dimension change (new attribute, re-bucketed time of day) is picked up atomically. A streaming table would freeze rows written under older dimension logic and need a full refresh — which destroys streaming state — to catch up. Liquid-clustered on (date_sk, pickup_zone_sk).

Divide-by-zero safety in the silver projection

fare_per_mile is computed with F.try_divide(fare_amount, trip_distance), not the raw / operator. The positive_trip_distance expectation and this projection are part of the same query, so a trip_distance == 0 row can be divided before the expectation drops it; under ANSI mode the raw operator would raise DIVIDE_BY_ZERO and fail the entire update instead of the row being cleanly dropped. try_divide yields NULL in that case, so the result is correct no matter what order the optimizer picks. Rounding to 4 places is unchanged. Verified: 0 NULL fare_per_mile values in silver, since all 76 zero-distance rows are dropped by the expectation as intended.

Surrogate keys are deterministic, not generated

Every gold SK is xxhash64(<natural key>) — never monotonically_increasing_id():

date_sk = xxhash64(cast(date as string)) · zone_sk = xxhash64(cast(zip as string)) · time_of_day_sk = xxhash64(label) · rider_sk = xxhash64(rider_id) · trip_sk = trip_key

fact_trips recomputes FKs with the same expressions the dimensions use, so FKs resolve without a join and can never drift. rider_id on silver trips is RDR- + lpad(pmod(xxhash64(trip_key), 500), 5, '0') — a stable hash of a stable natural key, so the same trip maps to the same rider forever. No rand(), no re-rolling.

Verified stable: after a full --full-refresh of the entire pipeline (bronze reprocessed from scratch), the md5 fingerprint of the trip→rider assignment ordered by trip_key was byte-identical (1ff9a8f524201c4ad7dcc74e57290c90 before and after), as were all row counts and the rider dimension fingerprint (0cf2b1191a8f96397d7803d91f241ba0).

Deploy + run confirmation

databricks bundle validate / deploy -t dev -p DEFAULT → OK. Pipeline nyctaxi_medallion_etl, ID 2e6386e7-1a11-4766-8af4-ded34e54eb06, serverless: true — the spec contains no cluster, node type, or autoscale config.

update	kind	state
0afed5db-010e-4b82-bf03-afe93fce9b45	first attempt	FAILED — element_at needs an INT position, pmod(xxhash64(...)) is BIGINT. Fixed with an explicit .cast("int").
e1ab1919-47a1-463b-8eb8-7510482ccba8	incremental	COMPLETED ✅
0bee0c6f-801b-45f1-8de2-d29048e75d68	full refresh (determinism proof)	COMPLETED ✅
737d4e65-b97e-463e-94fa-bb54419bb007	full refresh after the try_divide / borough fixes	COMPLETED ✅

The last update is a full refresh because is_region_known is a new column on a streaming table: already-written rows are not reprocessed by an incremental update, so they would have kept NULL for the new column and the old is_cross_borough semantics. A full refresh is safe here precisely because everything is deterministic — the fingerprint check below confirms it reproduced the identical trip→rider assignment.

Final row counts

table	dataset type	rows
main.nyctaxi_bronze.trips_raw	streaming table	21,932 (= source, nothing lost)
main.nyctaxi_silver.trips_enriched	streaming table	21,847 (21,932 − 85 dropped by expectations)
main.nyctaxi_silver.riders	materialized view	500
main.nyctaxi_gold.dim_date	materialized view	60 (2016-01-01 → 2016-02-29)
main.nyctaxi_gold.dim_zone	materialized view	200 distinct ZIPs
main.nyctaxi_gold.dim_time_of_day	materialized view	4
main.nyctaxi_gold.dim_rider	materialized view	500
main.nyctaxi_gold.fact_trips	materialized view	21,847 (trip grain, 1:1 with silver)

Counts are unchanged from before the review fixes, and the trip→rider fingerprint is still 1ff9a8f524201c4ad7dcc74e57290c90.

Referential integrity — LEFT ANTI JOIN from fact_trips to each dimension returns 0 orphans on all five FKs (date_sk, pickup_zone_sk, dropoff_zone_sk, time_of_day_sk, rider_sk), and all 500 riders are used by at least one trip.

Metric view: main.nyctaxi_gold.trips_metrics

pipelines/nyctaxi-medallion/resources/metric_views/trips_metrics.sql defines a Unity Catalog metric view (YAML version: 1.1) over the gold star schema, sourced from fact_trips with declarative joins to dim_date, dim_zone twice (pickup_zone / dropoff_zone, off pickup_zone_sk / dropoff_zone_sk), dim_time_of_day and dim_rider.

Deployment is DDL, not a bundle resource. DABs on CLI v1.9.0 has no metric_views resource type, so resources/metric_views/deploy.sh <WAREHOUSE_ID> DEFAULT submits the CREATE OR REPLACE VIEW ... WITH METRICS LANGUAGE YAML statement. It deliberately does not use aitools statement submit --file: that path strips the YAML block's leading indentation and the server rejects it with METRIC_VIEW_INVALID_VIEW_DEFINITION ... expected <block end>, but found '-'. JSON-encoding the statement to /api/2.0/sql/statements/ preserves whitespace exactly. Both the trap and the fix are documented in the file header.

Measures

measure	expression	notes
total_revenue	SUM(fare_amount)
trip_count	COUNT(1)
average_fare	AVG(fare_amount)
total_miles	SUM(trip_distance)
revenue_per_mile	try_divide(SUM(fare_amount), SUM(trip_distance))	ratio of sums, so it re-aggregates correctly at any grouping (not a mean of per-trip ratios)
average_trip_duration	AVG(trip_duration_minutes)
cross_borough_share	try_divide(COUNT(1) FILTER (WHERE is_cross_borough), COUNT(1) FILTER (WHERE is_cross_borough IS NOT NULL))	see NULL handling below
trailing_7day_revenue	SUM(fare_amount) + window: [{order: date, range: trailing 7 day, semiadditive: last}]	rolling window measure

Divide-by-zero handling — both ratio measures use try_divide, not /. A slice with zero total miles (or zero classifiable trips) yields NULL instead of raising DIVIDE_BY_ZERO under ANSI mode and killing the dashboard query. Same reasoning as the fare_per_mile fix in the silver layer.

NULL handling in cross_borough_share — is_cross_borough is NULL when the ZIP-range proxy cannot place an end of the trip. Those rows are excluded from both numerator and denominator, so the measure reads as "share of trips we could classify that crossed a borough." The alternative — counting unknowns in the denominator — would silently dilute the ratio toward zero and present an unclassified trip as a same-borough trip. COUNT(1) FILTER (WHERE is_cross_borough) counts only TRUE, and the denominator counts only non-NULL. On the current data this is a no-op (0 NULLs), so it is defensive rather than load-bearing.

Trailing window semantics, verified empirically — trailing 7 day covers the 7 days preceding each date and excludes the current date. Confirmed against the data: trailing_7day_revenue on 2016-01-12 is 30,551.5, exactly the sum of daily revenue for 2016-01-05 through 2016-01-11. semiadditive: last returns the last value when date is not in the GROUP BY.

Dimensions

• Date hierarchy — date (the window-ordering dimension), month = DATE_TRUNC('MONTH', ...), year, plus a day_of_week rollup (day_name) with day_of_week_number so day names sort chronologically instead of alphabetically, and is_weekend.
• Location — pickup_zip / pickup_borough from the pickup_zone join, and dropoff_zip / dropoff_borough from the dropoff_zone join.
• Time of day — time_of_day plus time_of_day_order (so morning → afternoon → evening → night sorts chronologically, not alphabetically).
• Rider — rider_id as the grain, plus rider_home_zip as coarse geography. full_name, email, phone and card_last4 are reachable through the join but deliberately not exposed as dimension attributes — a metric view is a governed, broadly-shared surface, and per-rider PII is not needed for any KPI here. (The data is synthetic either way; the point is that the pattern is right.)

Dashboard-read performance (measured)

Grain is kept at fact_trips row level with the joins declared in the YAML and pushed down, not pre-joined into a wide denormalized table: the joins are many-to-one on surrogate keys, so the engine prunes any dimension a query does not group by, whereas one wide table would force every query to scan the widest possible row.

rely: {at_most_one_match: true} on all five joins. Every join now carries the planner hint, and it is truthful rather than decorative — each dimension's surrogate key is verified unique, so each join really is at most 1:1 and the planner may skip the duplicate-match handling it would otherwise insert:

SELECT (SELECT COUNT(*)-COUNT(DISTINCT date_sk)         FROM main.nyctaxi_gold.dim_date)        dup_date,
       (SELECT COUNT(*)-COUNT(DISTINCT zone_sk)         FROM main.nyctaxi_gold.dim_zone)        dup_zone,
       (SELECT COUNT(*)-COUNT(DISTINCT time_of_day_sk)  FROM main.nyctaxi_gold.dim_time_of_day) dup_tod,
       (SELECT COUNT(*)-COUNT(DISTINCT rider_sk)        FROM main.nyctaxi_gold.dim_rider)       dup_rider
-- dup_date 0 | dup_zone 0 | dup_tod 0 | dup_rider 0   → no xxhash64 collisions anywhere

The field is accepted and round-trips in the stored definition — SHOW CREATE TABLE main.nyctaxi_gold.trips_metrics echoes rely: {at_most_one_match: true} under each of the five joins. Worth flagging: it is unenforced. If a dimension ever gained a duplicate SK the results would be silently wrong, so that uniqueness check belongs in any future gold-layer test suite. (It is also not documented in the installed databricks-metric-views skill — server-side acceptance is the confirmation.)

Measured timings. Warehouse 592a9f85708fccd4 (datakickstart_xs, PRO serverless), durations read from /api/2.0/sql/history/queries?include_metrics=true:

query	total	compile	execute	fetch
month × time_of_day KPI (8 rows) — run 1	1127 ms	603 ms	484 ms	74 ms
— run 2	1073 ms	579 ms	457 ms	53 ms
— run 3	1055 ms	538 ms	478 ms	44 ms
pickup_borough × pickup_zip slice (25 rows)	977–1049 ms	~0.55–0.60 s	378–414 ms	34–59 ms
rider_id slice (25 rows)	885–970 ms	~0.51 s	338–408 ms	34–49 ms

Straight answer: execution is comfortably sub-second — ~0.34–0.48 s — but end-to-end is ~0.9–1.1 s, not sub-second. Roughly half of every run is metric-view/YAML query compilation (~0.5–0.6 s), which is a fixed cost independent of data volume. result_from_cache was false on every run, so these are real executions rather than cache hits; the KPI query reads 3 files / 1.38 MB.

For comparison, the same three KPI runs before adding rely were 1297 / 1355 / 1511 ms total. The improvement is real but within noise for a warehouse this size and a fact table this small — the hint's value is at scale, not here.

EXPLAIN substantiates the join-pruning claim. On the KPI query the physical plan contains only 2 joins — dim_date and dim_time_of_day — with pickup_zone, dropoff_zone and dim_rider eliminated entirely, and both survivors are PhotonBroadcastHashJoin:

+- PhotonBroadcastHashJoin [time_of_day_sk], [time_of_day_sk], LeftOuter, BuildRight
:  +- PhotonBroadcastHashJoin [date_sk], [date_sk], LeftOuter, BuildRight

Zero SortMergeJoin and zero shuffle joins. That is the concrete evidence the "don't pre-join into a wide table" choice was right: an unused dimension costs nothing.

Other supporting facts — fact_trips is liquid-clustered on (date_sk, pickup_zone_sk), the two highest-cardinality join keys this view groups by, so date and pickup-zone filters get file skipping; the four dimensions (60 / 200 / 4 / 500 rows) are broadcast-sized, which the plan above confirms.

materialization: is still left off deliberately, documented as a commented-out block in the SQL file. Given the numbers, materializing would attack the ~0.4 s execution, not the ~0.6 s compilation that dominates — so it cannot get this under half a second, and it would stand up an extra hidden Lakeflow pipeline to own and pay for. The file shows the exact aggregated block (dimensions [month, time_of_day], measures [total_revenue, trip_count]) to enable if the fact table grows by orders of magnitude.

KPI query: revenue and trip count by month and time of day

SELECT
  month,
  time_of_day,
  MEASURE(total_revenue) AS total_revenue,
  MEASURE(trip_count)    AS trip_count
FROM main.nyctaxi_gold.trips_metrics
GROUP BY month, time_of_day, time_of_day_order
ORDER BY month, time_of_day_order

time_of_day_order is in the GROUP BY purely so ORDER BY can sort the buckets chronologically rather than alphabetically. Actual output (all 8 rows):

month	time_of_day	total_revenue	trip_count
2016-01-01	morning	31,011.50	2,612
2016-01-01	afternoon	40,565.50	3,254
2016-01-01	evening	37,538.00	3,196
2016-01-01	night	24,062.01	1,767
2016-02-01	morning	32,358.50	2,704
2016-02-01	afternoon	40,550.51	3,318
2016-02-01	evening	40,308.50	3,290
2016-02-01	night	22,627.51	1,706

Re-run after adding rely: {at_most_one_match: true} to all five joins: byte-identical output — same 8 rows, same revenue to the cent, same counts. Trip counts sum to 21,847, reconciling exactly with fact_trips. A cross-check by time_of_day alone also returns 21,847 trips with cross_borough_share ranging from 0.086 (morning) to 0.219 (night) and revenue_per_mile from 3.77 (night) to 4.53 (afternoon) — night trips are longer-haul and cross boroughs far more often, which is the kind of thing this layer exists to surface.

🤖 Generated with Claude Code