4. Data Transformation¶
Points: 5 — At least one transformation step that is clearly justified, supports the end user's use case, and solves a defined problem.
Overview¶
Raw data lands as-is. dbt does the rest: 3 staging models (cleaning + small derivations) and 2 mart tables that the recommender reads at request time.
flowchart TD
R1["traffic_raw.hourly_counts"] --> S1["stg_traffic__counts<br/>(table)"]
S1 --> S2["stg_traffic__counters<br/>(view)"]
R2["weather_raw.hourly_history"] --> S3["stg_weather__history<br/>(view)"]
S1 --> M1["mart_traffic_baseline<br/>(table)"]
S1 --> M2["mart_weather_effect<br/>(table)"]
S3 --> M2Materialization:
| Layer | Materialized | Why |
|---|---|---|
stg_traffic__counts |
table | 4.1M rows; queried again by both marts. A view here pulled 28-minute runs and parallel-worker IPC stalls. |
| Other staging | view | cheap, always fresh |
| Marts | table | precomputed lookups, recommender hits them at request time |
Staging: traffic counts¶
File: stg_traffic__counts.sql
Adds the grouping keys the baseline needs:
| Column | How |
|---|---|
ts_hour |
cast(ts as timestamp) |
hour_of_day |
extract(hour from ts_hour) |
day_of_week |
{{ day_of_week('ts_hour') }} — macro that emits dow on Postgres and dayofweek on BigQuery |
month_of_year |
extract(month from ts_hour) |
4,130,262 rows after the 2-year backfill.
Staging: counters¶
File: stg_traffic__counters.sql
A DISTINCT over (zsid, direction) plus the latest known coordinates and
street name. 216 rows. The recommender uses this to find which counters sit
along an OSRM polyline.
Staging: weather history¶
File: stg_weather__history.sql
Buckets precipitation into 4 readable bands and turns the WMO code into a short label:
| Bucket | Range |
|---|---|
none |
< 0.1 mm |
light |
0.1 – 1 mm |
moderate |
1 – 4 mm |
heavy |
> 4 mm |
Mart: traffic baseline¶
File: mart_traffic_baseline.sql
The "expected count" for a counter at a given hour, day-of-week and month:
{{ config(
materialized='table',
partition_by={'field': 'month_of_year', 'data_type': 'int64',
'range': {'start': 1, 'end': 13, 'interval': 1}},
cluster_by=['zsid', 'day_of_week', 'hour_of_day']
) }}
select
zsid, direction,
month_of_year, day_of_week, hour_of_day,
avg(vehicle_count) * 1.0 as avg_vehicle_count,
{{ median('vehicle_count') }} as median_vehicle_count,
stddev_pop(vehicle_count) as stddev_vehicle_count,
count(*) as n_observations
from {{ ref('stg_traffic__counts') }}
group by 1, 2, 3, 4, 5
423,438 rows after the 2-year backfill (one per counter × month × dow × hour).
On BigQuery the table is partitioned by month_of_year (integer range
1..12) and clustered on (zsid, day_of_week, hour_of_day). The recommender
always filters on all four keys, so partition pruning drops 11/12 of the
table and cluster pruning narrows the surviving partition to one counter
and time-of-day block. Postgres ignores both configs silently.
Mart: weather effect¶
File: mart_weather_effect.sql
The coefficient for "how much does precipitation bucket X shift the count at this counter, dow, hour?"
It's computed in a single scan: join counts to weather, then aggregate twice
— once by (zsid, direction, dow, hour, bucket) and once by (zsid,
direction, dow, hour) — and divide. Joining back to mart_traffic_baseline
caused the 28-minute hang mentioned above, this is the rewrite.
{{ config(
materialized='table',
partition_by={'field': 'day_of_week', 'data_type': 'int64',
'range': {'start': 1, 'end': 8, 'interval': 1}},
cluster_by=['zsid', 'hour_of_day', 'precip_bucket']
) }}
with joined as (
select c.zsid, c.direction, c.day_of_week, c.hour_of_day,
c.vehicle_count, w.precip_bucket
from {{ ref('stg_traffic__counts') }} c
join {{ ref('stg_weather__history') }} w
on w.ts_hour = c.ts_hour
and w.latitude = nearest(c.ekoord, c.nkoord).lat -- pseudocode
)
select b.zsid, b.direction, b.day_of_week, b.hour_of_day, b.precip_bucket,
b.bucket_avg / o.baseline_avg as weather_coef,
b.observations
from by_bucket b
join overall o using (zsid, direction, day_of_week, hour_of_day)
where o.baseline_avg > 0
108,690 rows. The shape of the signal lines up with intuition:
| precip_bucket | rows | avg coef |
|---|---|---|
| none | 35,952 | 1.003 |
| light | 35,950 | 0.989 |
| moderate | 33,703 | 0.995 |
| heavy | 3,083 | 0.978 |
So heavy rain pulls the average down by ~2 %. Small, but consistent.
The where o.baseline_avg > 0 filter drops two zero-baseline edge cases
(Z060 at 06:00 on a Saturday) that would otherwise produce NULL coefficients.
Dual-target¶
All SQL is written against ANSI shapes plus two project macros for the bits that differ between engines:
{{ day_of_week(col) }}—extract(dow ...)on Postgres,extract(dayofweek ...)on BigQuery.{{ median(col) }}—percentile_cont(0.5) within group (order by ...)on Postgres,approx_quantiles(..., 100)[offset(50)]on BigQuery.
Same models run on PostgreSQL (--target dev) and BigQuery (--target prod)
without changes.
How to verify¶
$ docker compose run --rm dev uv run dbt build \
--project-dir pipelines/transformation/dbt \
--profiles-dir pipelines/transformation/dbt
...
Completed successfully
Done. PASS=N WARN=0 ERROR=0 SKIP=0 TOTAL=N
End-to-end build runs in about 8 seconds after the staging-table fix.