Cascading hourly-to-daily rollups without double-counting
A district-heating operator runs 640 substations, each with a flow meter that reports a monotonically increasing energy_kwh counter and a supply_temp_c reading every ten seconds. Finance wants a daily energy total per substation; operations wants a daily mean supply temperature. The obvious build — one task that reads raw and sums, another that reads raw and averages — is wrong on both fields: summing a cumulative counter multiplies the true energy several times over, and a plain mean of a day’s raw points quietly re-weights hours that reported fewer samples. The correct pattern is a cascade: an hourly task computes per-hour deltas and per-hour means from raw, and a daily task derives its totals from the hourly tier, not from raw again. This page builds that two-tier rollup — count-weighted means, per-interval energy deltas, and an offset that makes the daily task fire only after the last hour is complete and stay idempotent across re-runs. It is a concrete instance of the scheduling model in multi-tier rollup scheduling.
Prerequisites
Solution walkthrough
The cascade has three moving parts: what the hourly tier must carry so the daily tier can aggregate correctly, how the daily tier combines those carried values, and how the two schedules are offset so nothing runs early.
1. Compute per-hour deltas and carry the sample count
The hourly task is where the two anti-patterns are neutralised. Energy must be a per-hour delta, not a raw counter value, so it uses increase() — which sums only the positive jumps in the monotonic counter within each window and survives meter resets. Temperature is a mean, but the daily tier cannot re-weight it later unless the hourly tier records how many samples produced each mean, so the task also writes sample_count.
option task = {name: "heat-hourly-rollup", every: 1h, offset: 5m}
src =
from(bucket: "heat_raw")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "substation")
|> group(columns: ["substation_id"])
// per-hour energy consumed = increase of the cumulative counter
src
|> filter(fn: (r) => r._field == "energy_kwh")
|> increase()
|> aggregateWindow(every: 1h, fn: last, createEmpty: false)
|> set(key: "_field", value: "energy_sum")
|> set(key: "_measurement", value: "substation_hourly")
|> to(bucket: "heat_hourly")
// per-hour mean temperature AND the count that produced it
tmp =
src
|> filter(fn: (r) => r._field == "supply_temp_c")
tmp
|> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
|> set(key: "_field", value: "temp_mean")
|> set(key: "_measurement", value: "substation_hourly")
|> to(bucket: "heat_hourly")
tmp
|> aggregateWindow(every: 1h, fn: count, createEmpty: false)
|> set(key: "_field", value: "sample_count")
|> set(key: "_measurement", value: "substation_hourly")
|> to(bucket: "heat_hourly")
The offset: 5m gives late ten-second batches time to land before the window closes; because InfluxDB tasks keep now() aligned to the scheduled boundary while only delaying execution, range(start: -task.every) still captures exactly the clock hour just ended. Rounding of the mean is deliberately deferred to the higher tier — the reasoning is in precision mapping and rounding strategies.
2. Derive the daily tier from hourly with a count-weighted mean
The daily task reads heat_hourly, never heat_raw. Energy is trivially additive: the day’s total is the sum of 24 hourly deltas. Temperature is not — the mean of the hourly means is only correct when every hour reported the same number of samples, which never holds when a substation drops offline for part of an hour. The day’s mean must weight each hour by its own count:
$$ \bar{T}{\text{day}} = \frac{\sum^{23} n_h,\bar{T}h}{\sum^{23} n_h} ;\neq; \frac{1}{24}\sum_{h=0}^{23} \bar{T}_h $$
To evaluate the numerator in Flux, pivot temp_mean and sample_count into one row per hour, multiply them, then reduce:
option task = {name: "heat-daily-rollup", every: 1d, offset: 45m}
// energy: additive sum of the per-hour deltas
from(bucket: "heat_hourly")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "substation_hourly" and r._field == "energy_sum")
|> group(columns: ["substation_id"])
|> aggregateWindow(every: 1d, fn: sum, createEmpty: false)
|> set(key: "_field", value: "energy_sum")
|> set(key: "_measurement", value: "substation_daily")
|> to(bucket: "heat_daily")
// temperature: count-weighted mean over the hourly tier
from(bucket: "heat_hourly")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "substation_hourly")
|> filter(fn: (r) => r._field == "temp_mean" or r._field == "sample_count")
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|> group(columns: ["substation_id"])
|> reduce(
identity: {wsum: 0.0, nsum: 0.0},
fn: (r, acc) => ({
wsum: acc.wsum + r.temp_mean * float(v: r.sample_count),
nsum: acc.nsum + float(v: r.sample_count),
}),
)
|> map(fn: (r) => ({
_time: date.truncate(t: now(), unit: 1d),
substation_id: r.substation_id,
_measurement: "substation_daily",
_field: "temp_mean",
_value: r.wsum / r.nsum,
}))
|> to(bucket: "heat_daily")
Every written point carries a fixed _time of the day boundary (date.truncate) and the stable substation_id tag. Because InfluxDB overwrites any point sharing the same measurement, tag set, field, and timestamp, re-running this task produces byte-identical points — the rollup is idempotent, so a backfill or a retried run never adds a second copy.
3. Offset the schedules so daily never fires early
The daily task runs at every: 1d, offset: 45m, i.e. at 00:45. The last hourly window (23:00–00:00) is scheduled at 00:00 but executes at 00:05 because of the hourly offset; the 45-minute daily offset leaves a 40-minute margin for that final hourly write to complete and settle. Get this ordering wrong and the daily total is missing its last hour with no error raised. If your fleet’s late data or task queue depth needs a wider margin, widen the daily offset rather than the hourly one — the offset-versus-cadence trade-off is worked through in cron and interval scheduling logic, and the retry-safe structure of these tasks follows writing robust Flux scripts for automated data rollups.
Gotchas and edge cases
Summing a cumulative counter double-counts energy. District-heating meters report a lifetime energy_kwh total, not per-interval consumption. Running sum() on that raw field adds every reading — a value near 4,000,000 kWh repeated thousands of times a day — producing a daily “total” orders of magnitude too large. The fix is increase() at the hourly tier so each hour carries only the delta; the daily tier then sums those deltas, which are genuinely additive.
Mean-of-means silently re-weights sparse hours. If a substation reports 360 samples in a healthy hour but only 30 during a comms outage, averaging the two hourly means treats a 30-sample hour as equal to a 360-sample hour. Over a day with several degraded hours the daily mean drifts by tenths of a degree — enough to fail a compliance check. Carrying sample_count and computing $\sum n_h \bar{T}_h / \sum n_h$ restores the exact weighting a raw-tier mean would have given.
A daily task without offset alignment reads a partial final hour. Firing the daily rollup at exactly 00:00 races the 23:00 hourly window, which has not been written yet, so the daily point is built from 23 hours and looks plausibly wrong forever after. Always set the daily offset larger than the hourly offset plus the hourly task’s worst-case execution time, and verify the margin holds under peak load, not just at idle.
Verification
Prove the cascade agrees with a direct raw computation for one substation and one day — if the tiers are consistent, the daily energy equals increase() over raw, and the weighted mean matches a raw mean:
raw_energy = from(bucket: "heat_raw")
|> range(start: -1d)
|> filter(fn: (r) => r._measurement == "substation" and r._field == "energy_kwh")
|> filter(fn: (r) => r.substation_id == "sub-0142")
|> increase()
|> last()
|> findColumn(fn: (key) => true, column: "_value")
daily_energy = from(bucket: "heat_daily")
|> range(start: -1d)
|> filter(fn: (r) => r._measurement == "substation_daily" and r._field == "energy_sum")
|> filter(fn: (r) => r.substation_id == "sub-0142")
|> last()
|> findColumn(fn: (key) => true, column: "_value")
// these two totals should match within floating-point tolerance
A mismatch on energy points to a counter reset the hourly increase() handled differently, or a raw window that expired before the hourly task read it. A mismatch on temperature almost always means a plain mean crept in somewhere the weighted mean belongs.
FAQ
Why not compute the daily rollup directly from raw and skip the hourly tier?
Reading raw twice doubles the scan cost and re-exposes the daily task to raw retention: if heat_raw expires a window before the daily task reads it, the day is silently short. Deriving daily from hourly means the expensive raw scan happens once, the hourly tier acts as a durable checkpoint, and the daily task processes 24 rows per substation instead of thousands.
Is the weighted mean still needed if every hour has the same sample count?
No — when all n_h are equal the weighted mean collapses to the plain average, so the two agree exactly. The problem is that you cannot guarantee equal counts in an IoT fleet: outages, maintenance, and backhaul gaps make counts uneven, and the weighted form is correct in both cases, so there is no reason to risk the unweighted one.
How do I safely backfill a month of daily rollups?
Because each daily point is keyed on a truncated day timestamp and stable tags, re-running the daily task over a past range overwrites rather than appends. Drive it with a Python loop that sets the range per day and reuses the same task logic; the idempotent to() write makes repeated backfills safe.
Related
- Multi-Tier Rollup Scheduling — the tiered-cadence model this hourly-to-daily cascade instantiates.
- Precision Mapping & Rounding Strategies — where to round means and totals across tiers without accumulating error.
- Cron & Interval Scheduling Logic — choosing offsets and cadences so a downstream tier never fires before its source is complete.