Daily city temperatures
Build one daily temperature table from two weather stations, and refresh a few days at a time.
The question
Section titled “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
Section titled “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
Section titled “The recipe”{ "dataset": { "name": "New York and Denver temperature history", "description": "Each row is one calendar day at one airport station, carrying the highest air temperature the station reported that day, from the Iowa Environmental Mesonet ASOS archive." }, "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
Section titled “Part by part”Sources
Section titled “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
Section titled “The window”"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
Section titled “Transform”Two steps. The first names the city each station serves, reads each reading down to its UTC day, and puts both relations together:
select 'New York' as city, station, cast(substr(valid, 1, 10) as date) as observed_day, tmpffrom nyc_asosunion allselect 'Denver' as city, station, cast(substr(valid, 1, 10) as date) as observed_day, tmpffrom den_asosA 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:
select city, station, observed_day as valid, cast(max(try_cast(tmpf as decimal(6, 2))) as decimal(6, 2)) as max_temp_ffrom observationsgroup by city, station, observed_dayThe 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
Section titled “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
Section titled “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
Section titled “Register, run, inspect”mr-data dataset create --name "New York and Denver temperature history" --jsonmr-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 --jsonmr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --sample --max-rows 2000 --jsonmr-data peek RUN_ID --jsonmr-data checks RUN_ID --json
mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --jsonmr-data query RUN_ID "select city, count(*) from run_table group by city" --json
mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --refresh --jsonWhat the run reports
Section titled “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.
--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.