Multi-Tier Rollup Scheduling
A cascading rollup breaks in one of two silent ways, and neither shows up in a smoke test. Either a higher tier re-averages numbers that were already averages — computing a daily mean from twenty-four hourly means as if every hour carried equal weight — and the reported figure drifts away from the truth by a margin nobody can explain; or the tiers fire on offsets that do not cascade, so the daily task reads the hourly bucket a minute before the last hourly window has finished writing, and the day is quietly short one hour. Building a chain of downsampling tasks that step from raw to one-minute to one-hour to one-day means treating the schedule and the aggregation math as a single design problem: each tier must consume the tier below it, weight by the sample counts that tier carried, and start only after its parent has demonstrably finished. This page sits under downsampling & aggregation pipeline design and shows how to schedule that cascade so tiers feed each other without double-counting, gaps, or offset drift.
The failure scenario this solves
An EV-charging network operator runs 480 fast-charging stations with 3,200 connectors, each reporting instantaneous power draw in kilowatts every 10 seconds into a bucket named ev_charge_raw. That is roughly 8,640 samples per connector per day. To feed a grid-billing reconciliation and a capacity dashboard, the team built a three-hop cascade: a one-minute rollup, an hourly rollup, and a daily rollup, each reading from the tier below it and writing a mean of the field kw.
For the capacity dashboard it looked fine. Then finance ran the monthly demand-charge reconciliation against the utility and the numbers did not tie out. The daily job reported an average station power of 18.4 kW; the utility’s own interval meters put the true count-weighted daily mean at 22.1 kW — a 17 percent underreport worth five figures a month in disputed demand charges. The cause was in the daily task: it read the hourly kw_mean values and applied aggregateWindow(every: 1d, fn: mean). But charging sessions are bursty. Overnight hours had a handful of readings from idle connectors; midday hours had thousands from full stations. Averaging twenty-four hourly means gave a sleepy 03:00 hour — three readings of a trickle-charging car — exactly the same weight as a saturated 13:00 hour with 21,000 readings. The sparse hours dragged the daily figure down. The mean of the means was not the mean.
While they were fixing that, a second fault surfaced. The daily task ran at 00:01 UTC; the hourly task ran at the top of each hour with a two-minute offset, so the 23:00 window did not land in the hourly bucket until 00:02. The daily job had already read and closed its window — every day was silently missing its final hour. Two distinct bugs, one root disease: the tiers were scheduled and aggregated as if they were independent, when a cascade is a chain of dependencies. The rest of this page rebuilds it so counts propagate and offsets stagger.
Prerequisites
Core concept: propagate counts, stagger offsets
A multi-tier rollup has two invariants. The first is arithmetic: you cannot recover a mean from a set of means without their weights. The second is temporal: a child tier must not read a window its parent has not yet written. Break the first and every number above the one-minute tier is subtly wrong; break the second and the top tier develops a periodic gap exactly one parent-window wide.
The arithmetic invariant is why every tier writes two fields, not one. Each rollup stores the mean of its window and the count of underlying samples that produced it. The tier above then computes a count-weighted mean, never a plain mean of means. For a set of $k$ child windows with means $\bar{x}_i$ and sample counts $n_i$, the correct parent mean is:
$$\bar{x} = \frac{\sum_{i=1}^{k} n_i,\bar{x}i}{\sum^{k} n_i} ;\ne; \frac{1}{k}\sum_{i=1}^{k} \bar{x}_i$$
The right-hand expression — the naive mean of means — equals the true mean only in the special case where every $n_i$ is identical. The moment sample counts differ between windows (an idle connector versus a saturated one), the two diverge, and the direction of the error is toward whichever windows are sparsest. Carrying $n_i$ up the chain makes the weighted numerator a running sum(mean * count) and the denominator a running sum(count); their ratio reconstructs the exact mean the raw data would have produced in a single pass, at a fraction of the read cost. The same principle governs how many significant figures survive each hop, which is the province of precision mapping & rounding strategies — round too early and the weighted sum accumulates bias.
The temporal invariant is a scheduling constraint on the offset of each tier. A task with every: 1h, offset: 2m fires at two minutes past each hour and reads the window that just closed. If the tier above it fires before this write lands, it reads an incomplete parent. So each tier’s offset must clear its parent’s offset plus the parent’s execution time:
$$O_{child} \ge O_{parent} + D_{parent} + M$$
where $O$ is a tier’s offset, $D_{parent}$ is how long the parent task takes to run, and $M$ is a safety margin. Offsets therefore grow monotonically down the cascade: the one-minute tier waits 15 seconds for late packets, the hourly tier waits 2 minutes for the last one-minute window plus its own read, and the daily tier waits 10 minutes so the final hourly window is unambiguously written before the day closes. Getting that stagger right is the scheduling half of the problem, and the general mechanics of aligning every and offset to calendar boundaries are covered in cron & interval scheduling logic.
Step-by-step implementation
1. Roll raw into one-minute means and counts
The first hop reads 10-second raw power and writes two fields per window: kw_mean and kw_count. The count is not decoration — it is the weight every tier above depends on. The read range is anchored with tasks.lastSuccess() so a delayed run resumes from the last good window instead of skipping one, and concurrency: 1 guarantees runs never overlap and double-write. The retry-safe structure of these scripts is developed in flux scripting for task automation.
import "influxdata/influxdb/tasks"
option task = {
name: "rollup_power_1m",
every: 1m,
offset: 15s, // wait for late 10s packets to settle
concurrency: 1, // never overlap; no double-writes
}
src =
from(bucket: "ev_charge_raw")
|> range(start: tasks.lastSuccess(orTime: -10m))
|> filter(fn: (r) => r._measurement == "charge_power")
|> filter(fn: (r) => r._field == "kw")
src
|> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
|> set(key: "_field", value: "kw_mean")
|> to(bucket: "ev_charge_1m", org: "ev-cloud")
src
|> aggregateWindow(every: 1m, fn: count, createEmpty: false)
|> set(key: "_field", value: "kw_count")
|> to(bucket: "ev_charge_1m", org: "ev-cloud")
createEmpty: false keeps idle-connector windows from being padded with null rows that would otherwise inflate the count denominator upstream with zero-weight entries.
2. Roll one-minute into hourly with a count-weighted mean
The hourly tier must never take a plain mean of kw_mean. It pivots the mean and count of each one-minute window side by side, forms the weighted contribution kw_mean * kw_count per window, then reduces each hour to a weighted numerator and a count denominator. Their ratio is the exact hourly mean; the summed count is carried forward as the hour’s own kw_count so the daily tier can weight in turn.
import "influxdata/influxdb/tasks"
option task = {
name: "rollup_power_1h",
every: 1h,
offset: 2m, // clears the 1m tier's 15s offset + its runtime
concurrency: 1,
}
base =
from(bucket: "ev_charge_1m")
|> range(start: tasks.lastSuccess(orTime: -3h))
|> filter(fn: (r) => r._measurement == "charge_power")
|> filter(fn: (r) => r._field == "kw_mean" or r._field == "kw_count")
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|> window(every: 1h)
|> reduce(
identity: {wsum: 0.0, nsum: 0.0},
fn: (r, acc) => ({
wsum: acc.wsum + r.kw_mean * r.kw_count,
nsum: acc.nsum + r.kw_count,
}),
)
|> duplicate(column: "_stop", as: "_time")
base
|> map(fn: (r) => ({
_time: r._time, _measurement: "charge_power",
station_id: r.station_id, _field: "kw_mean",
_value: r.wsum / r.nsum,
}))
|> to(bucket: "ev_charge_1h", org: "ev-cloud")
base
|> map(fn: (r) => ({
_time: r._time, _measurement: "charge_power",
station_id: r.station_id, _field: "kw_count",
_value: r.nsum,
}))
|> to(bucket: "ev_charge_1h", org: "ev-cloud")
The offset: 2m is the temporal invariant in action: the last one-minute window of the hour lands at 15 seconds past, so two minutes leaves ample margin for it plus the hourly read itself.
3. Roll hourly into daily on a stagger that clears the parent
The daily tier is structurally identical to the hourly one — same weighted reduce, same carried count — but its offset is the critical parameter. At offset: 10m the daily task for a UTC day fires at 00:10, comfortably after the 23:00 hourly window (written by 23:02) has landed. This is exactly the hop the earlier failure got wrong, and the deeper double-counting traps specific to it are worked through in cascading hourly to daily rollups without double-counting.
import "influxdata/influxdb/tasks"
option task = {
name: "rollup_power_1d",
every: 1d,
offset: 10m, // clears the 1h tier's 2m offset + its runtime
concurrency: 1,
}
from(bucket: "ev_charge_1h")
|> range(start: tasks.lastSuccess(orTime: -50h))
|> filter(fn: (r) => r._measurement == "charge_power")
|> filter(fn: (r) => r._field == "kw_mean" or r._field == "kw_count")
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|> window(every: 1d)
|> reduce(
identity: {wsum: 0.0, nsum: 0.0},
fn: (r, acc) => ({
wsum: acc.wsum + r.kw_mean * r.kw_count,
nsum: acc.nsum + r.kw_count,
}),
)
|> duplicate(column: "_stop", as: "_time")
|> map(fn: (r) => ({
_time: r._time, _measurement: "charge_power",
station_id: r.station_id, _field: "kw_mean",
_value: r.wsum / r.nsum,
}))
|> to(bucket: "ev_charge_1d", org: "ev-cloud")
The -50h lookback on orTime is a resume cushion: if the daily task misses a run, the next execution still covers two full days rather than skipping the gap, and the pivot deduplicates by _time so a re-read of an already-written day overwrites rather than double-counts.
4. Decide which windows even need a rollup at all
Not every measurement deserves a full four-tier cascade. Fields that are only ever queried at coarse granularity can skip the one-minute tier, and fields whose values rarely cross a meaningful boundary can be filtered before they consume rollup budget. Choosing which windows are worth aggregating — and at what change threshold — is the subject of threshold tuning for aggregation; pruning here keeps the cascade’s series cardinality bounded so each tier’s task stays fast enough to finish inside its offset budget.
Configuration reference
| Tier | Source bucket | every |
offset |
Aggregation function |
|---|---|---|---|---|
| raw | (ingest write) | — | — | 10s samples of kw |
ev_charge_1m |
ev_charge_raw |
1m |
15s |
mean + count written as two fields |
ev_charge_1h |
ev_charge_1m |
1h |
2m |
count-weighted mean, count carried |
ev_charge_1d |
ev_charge_1h |
1d |
10m |
count-weighted mean, count carried |
The offsets are not arbitrary: each satisfies $O_{child} \ge O_{parent} + D_{parent} + M$. If a tier’s task ever runs long enough that its execution time $D$ eats into a child’s margin, widen the child’s offset rather than tightening the parent’s cadence. Every tier writes both kw_mean and kw_count except where a tier is terminal and nothing reads above it.
Common failure modes and fixes
1. Aggregating an average instead of re-aggregating from a lower tier with count weighting.
Symptom: higher-tier means drift systematically away from a direct raw computation, biased toward whichever windows are sparsest. Root cause: a tier applies fn: mean to a field that is itself a mean, giving every child window equal weight regardless of how many samples produced it. Fix: carry a count field at every tier and compute a count-weighted mean, never a mean of means.
// WRONG — mean of means; sparse idle hours skew the day down.
from(bucket: "ev_charge_1h")
|> range(start: -1d)
|> filter(fn: (r) => r._field == "kw_mean")
|> aggregateWindow(every: 1d, fn: mean, createEmpty: false)
The corrected form is the count-weighted reduce shown in step 3.
2. Offset cascade too tight — the daily reads an incomplete hourly. Symptom: the top tier is periodically short exactly one parent window; the most recent hour or day is missing on every run. Root cause: the child’s offset does not clear the parent’s offset plus its execution time, so the child reads before the parent writes. Fix: widen each offset monotonically down the chain so $O_{child} \ge O_{parent} + D_{parent} + M$.
// Daily must clear the hourly's 2m offset + runtime, not undercut it.
option task = {name: "rollup_power_1d", every: 1d, offset: 10m, concurrency: 1}
3. Overlapping windows double-write aggregates.
Symptom: an aggregate point has an inflated count or a doubled value after a retry or a manual backfill. Root cause: two runs of the same tier processed overlapping ranges and both called to(), or concurrency allowed a slow run to overlap the next. Fix: keep concurrency: 1, anchor reads with tasks.lastSuccess(), and rely on to() overwriting a point at an identical timestamp and tag set rather than appending — never widen a range so two runs share a window boundary.
4. Rounding applied before the weighted sum.
Symptom: the weighted mean is slightly but consistently off, worse the more tiers it passes through. Root cause: each tier rounds kw_mean before the tier above multiplies it by kw_count, so rounding bias compounds. Fix: keep full precision in the stored kw_mean and round only at the presentation layer, per precision mapping & rounding strategies.
5. Count lost between tiers.
Symptom: a tier works in isolation but the tier above it silently falls back to an unweighted mean. Root cause: a rollup wrote only kw_mean and dropped kw_count, so the parent has no weights to apply. Fix: assert both fields exist at every non-terminal tier; the verification query below catches a missing count.
Verification and testing
Verification answers three questions: does the weighted daily figure match a direct raw computation, is every tier carrying its count, and does a stalled tier raise an alert before the gap propagates upward?
First, reconcile the cascade against ground truth. Compute the same day’s mean directly from raw and compare it to the daily tier — they should agree to within floating-point noise, not by 17 percent:
direct = from(bucket: "ev_charge_raw")
|> range(start: -1d)
|> filter(fn: (r) => r._measurement == "charge_power" and r._field == "kw")
|> mean()
|> findColumn(fn: (key) => true, column: "_value")
rolled = from(bucket: "ev_charge_1d")
|> range(start: -1d)
|> filter(fn: (r) => r._measurement == "charge_power" and r._field == "kw_mean")
|> findColumn(fn: (key) => true, column: "_value")
Confirm every tier still carries its weight — a tier that returns no kw_count rows is silently un-weightable:
from(bucket: "ev_charge_1h")
|> range(start: -6h)
|> filter(fn: (r) => r._field == "kw_count")
|> count()
|> group()
|> sum()
Then add a deadman health check so a tier that stops producing pages an operator while the source data still exists to reprocess. If the hourly tier goes quiet, the daily tier is about to inherit a gap:
import "influxdata/influxdb/monitor"
import "experimental"
from(bucket: "ev_charge_1h")
|> range(start: -3h)
|> filter(fn: (r) => r._measurement == "charge_power" and r._field == "kw_mean")
|> monitor.deadman(t: experimental.subDuration(from: now(), d: 90m))
|> filter(fn: (r) => r.dead == true)
Finally, a scheduled Python check asserts the offset stagger itself — that each tier’s offset clears its parent’s — so a well-meaning edit that tightens an offset fails in CI rather than in production:
import os
from influxdb_client import InfluxDBClient
client = InfluxDBClient(
url=os.environ["INFLUX_URL"],
token=os.environ["INFLUX_TOKEN"],
org=os.environ["INFLUX_ORG"],
)
tasks_api = client.tasks_api()
# tier order and the parsed offset seconds pulled from each task's flux
chain = ["rollup_power_1m", "rollup_power_1h", "rollup_power_1d"]
offsets = {t.name: t.offset for t in tasks_api.find_tasks() if t.name in chain}
order = [offsets.get(name) for name in chain]
print("offsets (child must exceed parent):", list(zip(chain, order)))
for parent, child in zip(chain, chain[1:]):
assert offsets[child] > offsets[parent], f"{child} offset does not clear {parent}"
print("[OK] offset cascade is monotonic")
client.close()
Integration points
A cascade touches every neighbouring concern. The offsets that keep each tier reading a finished parent are an instance of the calendar-versus-interval reasoning in cron & interval scheduling logic, and the retry-safe tasks.lastSuccess() discipline that makes a delayed tier resume rather than skip comes from flux scripting for task automation. Whether each hop preserves enough significant figures for the weighted sum to stay unbiased is governed by precision mapping & rounding strategies, and deciding which windows earn a rollup at all — pruning idle or below-threshold series so tasks finish inside their offset budget — is threshold tuning for aggregation. The programmatic task-management endpoints used by the schedule audit are documented in the InfluxDB Task API reference.
FAQ
Why is a daily mean of hourly means wrong?
Because a plain mean gives every hour equal weight regardless of how many samples produced it. When some hours are sparse (idle connectors) and some are dense (saturated stations), the sparse hours pull the daily figure toward their values. The mean of means equals the true mean only when every window has the same sample count, which bursty telemetry never does — so you carry a count at each tier and compute a count-weighted mean.
How do I choose the offset for each tier?
Make each offset clear its parent’s offset plus the parent’s execution time, plus a margin: $O_{child} \ge O_{parent} + D_{parent} + M$. In practice that means offsets grow down the chain — 15 seconds at the one-minute tier, two minutes at the hourly, ten minutes at the daily. If a parent task starts running long, widen the child’s offset rather than speeding the parent up.
Do I have to store a count field at every tier?
At every tier that something reads above it, yes. The count is the weight the next tier needs; without it the parent silently falls back to an unweighted mean. A terminal tier that nothing consumes can drop the count, but it is cheap to keep and makes the chain extensible.
What stops a retried task from double-counting?
Three things together: concurrency: 1 so runs never overlap, tasks.lastSuccess() so a resumed run starts where the last one ended, and to() overwriting any point written at an identical timestamp and tag set. As long as no run widens its range to share a boundary with another, a retry re-writes a window rather than appending to it.
Can I skip the one-minute tier and roll raw straight to hourly?
Yes, if nothing queries minute-resolution data and the raw-to-hourly task still finishes inside its offset. The tiers exist to bound each task’s read cost and to serve intermediate query granularities; if neither applies to a measurement, a shorter cascade is legitimate. Just keep carrying the count so the hourly and daily tiers stay weightable.
Related
- Cascading hourly to daily rollups without double-counting — the top hop in depth: idempotent daily writes and gap-safe backfills.
- Precision Mapping & Rounding Strategies — keep enough significant figures for the weighted sum to stay unbiased through every hop.
- Threshold Tuning for Aggregation — decide which windows earn a rollup so tasks finish inside their offset budget.
- Cron & Interval Scheduling Logic — align each tier’s
everyandoffsetto calendar boundaries without drift. - Flux Scripting for Task Automation — the retry-safe patterns behind resumable, non-overlapping rollup tasks.
Up one level: Downsampling & Aggregation Pipeline Design