Help Needed: Creating an Influx Query to Aggregate Data in 10s Intervals

InfluxDB 2
1

Hello everyone,

I’m currently working on a project where I need to aggregate data from an InfluxDB database in 10-second intervals. I have some experience with InfluxDB, but I’m facing some challenges with this specific requirement.

import "date"
from(bucket: "main")
    |> range(start: -10000ms)
    |> filter(fn: (r) => r["_measurement"] == "measurements")
    |> aggregateWindow(every: 10000ms, fn: mean, createEmpty: true)

I want to get 1 aggregated value with 10second interval, but usually I get 2 values, not sure why. Can anyone point out what I’m missing?
image

2

@CDRW When aggregateWindow determines interval boundaries, it does so using logical boundary steps. For example, 10s interval will generate window boundaries on 00:00:00, 00:00:10, 00:00:20, … etc. However, in range(), the start time is relative to now. So if right now is 00:00:23, the range of the query covers 00:00:13 to 00:00:23. This actually spans across two logical window boundaries, so you get two results.

When it comes to the output timestamps, by default, aggregateWindow() uses 1) the stop boundary for the window or 2) if the stop boundary is outside of the query range, the stop boundary of the query. That is why the two output timestamps are 2023-12-05T02:22:20 and 2023-12-05T02:22:23.

So really anytime you manually run this query, it will return two results (unless you somehow manage to run it on the exact nanosecond of a logical time boundary). However, if you’re going to be running this query as a task, the task engine schedules tasks on logical time boundaries, so you shouldn’t have this issue.

But if you want to protect against this and make sure your query range always falls on logical time boundaries (even when you manually execute it), you can override the start and stop query boundaries to use timestamps truncated to the latest 10s time.

import "date"

interval = 10s
truncatedNow = date.truncate(t: now(), unit: interval)
start = date.sub(d: interval, from: truncatedNow)
stop = truncatedNow

from(bucket: "main")
    |> range(start: start, stop: stop)
    |> filter(fn: (r) => r["_measurement"] == "measurements")
    |> aggregateWindow(every: interval, fn: mean, createEmpty: true)
1 Like