# Daily city temperatures Source: https://mostlyright.md/docs/recipes/city-temperatures/ Build one daily temperature table from two weather stations, and refresh a few days at a time. ## The question What was the highest air temperature reported at New York Central Park and Denver International on each calendar day? One row per city per day. The readings arrive hourly, so the recipe reduces them. The archive grows every hour, so the recipe declares which parameters of its own address carry time. This is the hosted version of the harness `city-temperature` example. That example reads two checked-in CSV files on the retired local path. A `user_file` locator has no hosted route, so these sources are public addresses. ## The sources Two entries, one per station, on the same publisher and the same endpoint. | | | | --- | --- | | Publisher | Iowa Environmental Mesonet, Iowa State University | | Addresses | `…/request/asos.py?station=NYC&data=tmpf&year1=…` and the same with `station=DEN` | | Format | comma-separated text, served as `text/plain; charset=UTF-8` | | Cadence | about one reading per station per hour | | Rights | United States federal weather observations, redistributed by IEM, recorded as `public_domain_asserted` | One address serves one station, so two stations are two sources. They share a publisher, a rights claim, a limit block and a window, and the transform puts them together. ## The recipe ```json title="recipe.json" { "dataset": { "name": "New York and Denver temperature history" }, "question": { "text": "What was the highest temperature reported at New York Central Park and Denver International on each day?" }, "table": { "name": "daily_maximum_temperature", "description": "One row per city and calendar day, from June 2026 onward.", "grain": [ "city", "valid" ], "columns": [ { "name": "city", "type": "string", "nullable": false, "description": "The city the reporting station serves.", "presentation": { "chart": "top_values" } }, { "name": "station", "type": "string", "nullable": false, "description": "The ASOS station identifier the observations came from.", "presentation": { "chart": "top_values" } }, { "name": "valid", "type": "date", "nullable": false, "description": "The UTC calendar day the observations were taken on. The acquired relation names its timestamp column valid, and the table declares the day under that name so a refresh can partition by it.", "presentation": { "chart": "timeline", "bucket": "month", "story": "The archive runs unbroken from the day the recipe was registered." } }, { "name": "max_temp_f", "type": "decimal", "nullable": true, "description": "The highest air temperature the station reported that day.", "presentation": { "chart": "histogram", "bins": 10, "story": "Most days sit in a narrow band and the heatwave days sit well to the right." } } ] }, "sources": [ { "name": "nyc_asos", "description": "Hourly air temperature readings at New York Central Park.", "source_class": "user_api", "data_classification": "public", "locator": { "kind": "https_url", "display_locator": "https://mesonet.agron.iastate.edu/cgi-bin/request/asos.py?station=NYC&data=tmpf&year1=2026&month1=6&day1=1&year2=2026&month2=6&day2=2&format=onlycomma&tz=Etc/UTC&report_type=3&latlon=no&missing=empty&trace=empty" }, "rights_claim": { "claimed_basis": "public_domain_asserted", "claim_evidence_digest": "sha256:1111111111111111111111111111111111111111111111111111111111111111", "claim_note": "United States federal weather observations, redistributed by the Iowa Environmental Mesonet." }, "connector": { "adapter_id": "public.https", "credential_mode": "none", "origin": "https://mesonet.agron.iastate.edu" }, "limits": { "max_source_bytes": 4194304, "max_rows": 2000000, "max_requests": 2 }, "window": { "start_at": "2026-06-01T00:00:00Z", "granularity": "day", "timezone": "UTC", "lookback_seconds": 172800, "max_span_seconds": 34560000, "request": { "start": { "encoding": "date_parts", "bound": "inclusive", "pad": "none", "parameters": { "year": "year1", "month": "month1", "day": "day1" } }, "end": { "encoding": "date_parts", "bound": "exclusive", "pad": "none", "parameters": { "year": "year2", "month": "month2", "day": "day2" } } }, "merge": { "materialization": "partition_replace", "partition": { "column": "valid", "key": "iso_date_prefix" }, "row_identity": [ "station", "valid" ] } } }, { "name": "den_asos", "description": "Hourly air temperature readings at Denver International Airport.", "source_class": "user_api", "data_classification": "public", "locator": { "kind": "https_url", "display_locator": "https://mesonet.agron.iastate.edu/cgi-bin/request/asos.py?station=DEN&data=tmpf&year1=2026&month1=6&day1=1&year2=2026&month2=6&day2=2&format=onlycomma&tz=Etc/UTC&report_type=3&latlon=no&missing=empty&trace=empty" }, "rights_claim": { "claimed_basis": "public_domain_asserted", "claim_evidence_digest": "sha256:1111111111111111111111111111111111111111111111111111111111111111", "claim_note": "United States federal weather observations, redistributed by the Iowa Environmental Mesonet." }, "connector": { "adapter_id": "public.https", "credential_mode": "none", "origin": "https://mesonet.agron.iastate.edu" }, "limits": { "max_source_bytes": 4194304, "max_rows": 2000000, "max_requests": 2 }, "window": { "start_at": "2026-06-01T00:00:00Z", "granularity": "day", "timezone": "UTC", "lookback_seconds": 172800, "max_span_seconds": 34560000, "request": { "start": { "encoding": "date_parts", "bound": "inclusive", "pad": "none", "parameters": { "year": "year1", "month": "month1", "day": "day1" } }, "end": { "encoding": "date_parts", "bound": "exclusive", "pad": "none", "parameters": { "year": "year2", "month": "month2", "day": "day2" } } }, "merge": { "materialization": "partition_replace", "partition": { "column": "valid", "key": "iso_date_prefix" }, "row_identity": [ "station", "valid" ] } } } ], "transform": { "engine": "duckdb_sql", "steps": [ { "step_id": "observations", "sql": "select 'New York' as city, station, cast(substr(valid, 1, 10) as date) as observed_day, tmpf from nyc_asos union all select 'Denver' as city, station, cast(substr(valid, 1, 10) as date) as observed_day, tmpf from den_asos", "description": "Put the two stations into one relation, name the city each one serves, and read each reading down to its UTC day." }, { "step_id": "daily_maximum_temperature", "sql": "select city, station, observed_day as valid, cast(max(try_cast(tmpf as decimal(6, 2))) as decimal(6, 2)) as max_temp_f from observations group by city, station, observed_day", "description": "Reduce the hourly readings to one row per city and day, and return the day under the name the window partitions on. The outer cast is what makes the result DECIMAL rather than DOUBLE, which is what the declared type admits." } ] }, "checks": [ { "check_id": "one_row_per_city_day", "kind": "key_uniqueness", "enforcement": "required", "columns": [ "city", "valid" ], "description": "One row per city per calendar day." }, { "check_id": "temperature_present", "kind": "null_ceiling", "enforcement": "required", "columns": [ "max_temp_f" ], "max_null_ppm": 20000 }, { "check_id": "temperature_plausible", "kind": "value_range", "enforcement": "required", "columns": [ "max_temp_f" ], "min_value": "-60", "max_value": "140" }, { "check_id": "days_present", "kind": "row_expectation", "enforcement": "required", "min_rows": 1 } ], "units": [ { "column": "max_temp_f", "unit": "[degF]" } ], "timezone": "UTC" } ``` ## Part by part ### Sources Each source names `data=tmpf` alone, so the decoded relation is `station,valid,tmpf`. Asking for five measurements and using one costs bytes on every refresh. Both sources carry the same `window`, because both addresses spell time the same way. ### The window ```json "window": { "start_at": "2026-06-01T00:00:00Z", "granularity": "day", "timezone": "UTC", "lookback_seconds": 172800, "max_span_seconds": 34560000, "request": { "start": {"encoding": "date_parts", "bound": "inclusive", "pad": "none", "parameters": {"year": "year1", "month": "month1", "day": "day1"}}, "end": {"encoding": "date_parts", "bound": "exclusive", "pad": "none", "parameters": {"year": "year2", "month": "month2", "day": "day2"}} }, "merge": {"materialization": "partition_replace", "partition": {"column": "valid", "key": "iso_date_prefix"}, "row_identity": ["station", "valid"]} } ``` `start_at` is the first UTC midnight of this corpus. A full run renders its start there, and the engine clamps every derived window to it. The registered address needs query literals that render the same instant, so it says `year1=2026&month1=6&day1=1`. `pad: "none"` renders `month1=6` rather than `month1=06`, which is how this publisher writes it. The year is four digits either way. `bound` describes the publisher, not the engine. The engine's window is half-open `[start, end)` in UTC and the renderer converts. To learn which bound a publisher means, run one windowed pull against a full pull over the same days and compare row for row. Their documentation is not evidence. `lookback_seconds: 172800` re-requests the two days before the last committed boundary, so a late or corrected observation lands. `max_span_seconds: 34560000` is 400 days. The engine clips a run that fell further behind rather than refusing it, and the next tick continues from what it covered. `merge.partition.column` is `valid`, a column of the acquired relation. `iso_date_prefix` takes the first ten ISO characters of that cell. `row_identity` is what the merged relation must be unique on. The table declares a column named `valid` too, and that is what makes this two-source plan refreshable. A refresh plan naming more than one source needs every window's partition column on the table, under the name the acquired relation uses. The transform carries the name through rather than deriving an `observed_on` from it. The first windowed refresh writes a version even if nothing changed. A source's first stored relation carries the publisher's bytes, and a merged relation the engine's canonical CSV. The first refresh after a window arrives normalises the spelling and mints a version. Later refreshes end `unchanged`. ### Transform Two steps. The first names the city each station serves, reads each reading down to its UTC day, and puts both relations together: ```sql select 'New York' as city, station, cast(substr(valid, 1, 10) as date) as observed_day, tmpf from nyc_asos union all select 'Denver' as city, station, cast(substr(valid, 1, 10) as date) as observed_day, tmpf from den_asos ``` A per-source constant carries a source's own identity into the table without a lookup file. The second reduces to one row per city per day: ```sql select city, station, observed_day as valid, cast(max(try_cast(tmpf as decimal(6, 2))) as decimal(6, 2)) as max_temp_f from observations group by city, station, observed_day ``` The day gets its table name in the last step, where nothing else reads a column called `valid`. Aliasing an input column to its own name inside a grouped statement asks the binder which one a bare `valid` means, and a worked recipe should not depend on that answer. The outer `cast` sets the result type. `max()` over a decimal returns a decimal here. An arithmetic expression usually produces `DOUBLE`, and a column declared `decimal` takes `DECIMAL(p,s)` and nothing else. ### Checks `key_uniqueness` on `["city", "valid"]` is the grain, and it catches a merge that duplicated a day. `value_range` bounds `max_temp_f` between `"-60"` and `"140"`, as decimal strings, because the document carries no fractional JSON number. `null_ceiling` tolerates two per cent missing. `row_expectation` guards against an empty answer. ### Cadence `mr-data promote TABLE_ID` records how often this table refreshes. It does not make the table readable. The first succeeded run does that on its own. A daily table needs a daily schedule or faster. A window whose granularity is `day` never runs slower than a day. Two windowed sources are one more condition. Where a refresh plan names more than one source, every window's `merge.partition.column` has to be a column the table declares, and all of them have to agree on the materialization and the partition key. Both windows here name `valid` with `partition_replace` and `iso_date_prefix`, and the table declares `valid`. So the two windows refresh together, and each refresh replaces the day partitions it re-requested and keeps the rest. Drop that column from the table, or rename it on the way out, and the table stops being partitioned by anything. Studio then answers `RESYNC_REQUIRED` before acquisition and the recipe advances only by explicit resync. ## Register, run, inspect Use the real table for one full build. For a bounded test first, follow the [scratch-table workflow](/docs/build/run-and-inspect/#test-before-the-full-build). A successful sample on the real table can publish truncated data. Stop if the forecast refuses; size the recipe before running. ```sh mr-data dataset create --name "New York and Denver temperature history" --json mr-data dataset set DATASET_ID --category climate-environment \ --topics "weather,temperature,asos,new york,denver" --license "CC0-1.0" --json mr-data recipe recipe.json --json mr-data recipe explain RECIPE_ID --json mr-data recipe forecast RECIPE_ID --digest RECIPE_DIGEST --json mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --json mr-data watch RUN_ID --json mr-data peek RUN_ID --json mr-data checks RUN_ID --json mr-data query RUN_ID "select city, count(*) from run_table group by city" --json mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --refresh --json ``` ## What the run reports `mr-data receipt RUN_ID --json` carries one entry per source, so two here. Each entry names the address the run rendered rather than the one in the document, the bytes fetched, the digest, and a coverage disposition. On a strict refresh, only a source the recipe declares `closed: true` can reuse its predecessor. A publisher's `304` is an unchanged reply after acquisition, and the legacy settled-source heuristic is not a reuse authorization. A source without an incremental strategy is refused before Studio acquires it. A refresh that gained nothing produces bytes identical to the previous run's. It writes nothing, uploads nothing, and ends `unchanged`. ## Traps `--max-rows` applies per source. Two sources cut at 2,000 rows each deliver up to 4,000 rows. The engine stitches the two partial spans into one table, with seams that exist nowhere in the data. A continuity or completeness check that passes over the whole table fails over that sample. Measure the sources, then set the ceiling over the largest. The engine refuses a publisher answer that carries rows outside the window, with `WINDOW_ROWS_OUT_OF_RANGE`. It does not clip them. Replacing a partition the run never asked for would refill it from a slice that is not authoritative for it. A source that declares a `window` never counts as closed. Its end parameters are a placeholder the engine rewrites on every run. The range its address carries today says nothing about whether the corpus has finished. A source states at most one of `window` and `closed`. Each declared parameter name must occur once in the query. The engine refuses a query where `year1` appears twice or not at all. The rest of the address stays fixed: the scheme, the host, the path, every unrelated query pair. Host allowlisting and attribution keep using the registered origin. A grouped column must appear in `group by` in the same spelling. `substr(valid, 1, 10)` runs in the union step, so the grouped step names the plain column `observed_day` in both the select list and the grouping. Deriving the day inside the grouped step instead would mean repeating the whole `substr(…)` expression in `group by`, because the output alias does not stand in for it.