Multiple aggregation windows for power aggregation

[RandGH]

Hi,
I run Grafana (8, with an influx 1.8 backend) to visualize data from a power meter. After searching for a while I found an InfluxQL query that works with my data to aggregate power consumption over time:

SELECT cumulative_sum(SUM(“power”::float)) / 60.0
FROM (
SELECT mean(“value”) AS power FROM “javascript.0.Volkszaehler.AKTUELLER_Bezug” WHERE $timeFilter GROUP BY time(1m) tz(‘Europe/Berlin’)
)
GROUP BY time(1h) tz(‘Europe/Berlin’)

I can modify this query to graph various timeframes, but due to the fact that influxql is incapable to run Month or Year times for longer timeframes this is alway off.

Google provided the information that i should implement my query in flux instead. I have no knowledge of flux , but further searching enabled me to at least come up with the following flux query that covers the inner sql statement:

FinalOutput = ["_field", “_time”, “_value”]

RawSeries = from(bucket: “iobroker/autogen”)
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == “javascript.0.Volkszaehler.AKTUELLER_Bezug”)
|> filter(fn: (r) => r["_field"] == “value”)
|> aggregateWindow (every:1m, fn: mean, createEmpty: true)

NamedSeries = RawSeries
|> map(fn: (r) => ({_value:r._value, _time:r._time, _field:“AKTUELLER_Bezug”}))
|> keep(columns:FinalOutput)

NamedSeries |> yield()

Now this is off by timezone (which is missing in flux), but else its identical to influxql.
What i cant seem to wrap my head around is how to implement the outer statement with either a second statement or within the same…

I’d appreciate pointers on how to do this with flux.

Thanks,
cheers

[RandGH]

Sorry for replying to myself, but couldn’t see how to edit my post.

Just wanted to mention that I had seen Migrate query to calculate energy (kWh) - #29 by pawelgl (and others to import timezone), but i think its not supported in 1.8x. At least its not working. So if anyone has a neat trick to do that on top of the double aggregation…

Since i saw the timezone comment I o/c also saw the idea to use integral, but that is a solution to calculate power usage based on a difference of totals (ever increasing consumption counter) vs my goal to sum up current power draw averages/sums over time…

I could do the same o/c but i don’t have the historical data of the energy counter (in the correct format / database) while i do have the historical consumption (+ solar generation), at both high level (total) and also (some) device levels, so its much nicer data although its more difficult to display (since I will have to dynamically sum up 2 years of data in the extreme case, but its fine to wait a few secs for these dashboards).

[RandGH]

So it seems like nobody has an idea here - anybody got an idea where else I might find help?

[Anaisdg]

Hello @RandGH,
Sorry for the delay. Feel free to tag me in the future.
First the following resources might be useful to you:

%%title%%

TL;DR InfluxDB Tech Tips — From Subqueries to Flux!

Now let’s convert that InfluxQL query!

SELECT cumulative_sum(SUM(“power”::float)) / 60.0
FROM (
SELECT mean(“value”) AS power FROM “javascript.0.Volkszaehler.AKTUELLER_Bezug” WHERE $timeFilter GROUP BY time(1m) tz(‘Europe/Berlin’)
)
GROUP BY time(1h) tz(‘Europe/Berlin’)

import "timezone"

option location = timezone.location(name: "Europe/Berlin")

from(bucket: "noaa")
  |> range(start: startTime, stop: stopTime)
  |> filter(fn: (r) => r["_measurement"] == "average_temperature")
  |> filter(fn: (r) => r["_field"] == "degrees")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
  |> toFloat()
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> map(fn: (r) => ({ r with _value: r._value/60.0}))
  |> cumulativeSum()
  |> yield(name: "cumulative sum")

We use the following functions:

map() function | Flux 0.x Documentation

aggregateWindow() function | Flux 0.x Documentation

cumulativeSum() function | Flux 0.x Documentation

toFloat() function | Flux 0.x Documentation

Please let me know if that works for you!

[RandGH]

Hi @Anaisdg,

thanks for picking this up

I adapted your example to

from(bucket: “iobroker/autogen”)
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == “javascript.0.Volkszaehler.AKTUELLER_Bezug”)
|> filter(fn: (r) => r["_field"] == “value”)
|> aggregateWindow (every:1m, fn: mean, createEmpty: true)
|> toFloat()
|> aggregateWindow(every: 1h, fn: sum, createEmpty: true)
|> map(fn: (r) => ({ r with _value: r._value/60.0}))
|> cumulativeSum()
|> yield(name: “cumulative sum”)


and that seems to work as expected, thank you very much. I was so close with my last attempts, but couldn’t make it work;)

Couple of issues remain -

1. as mentioned timezone is not working on Influx 1.8 - is there a workaround?
2. Naming the yield is not working either, which is the reason i had the Named series workaround - is this also an issue with the 1.x version
3. Slightly different issue - why is there no button for the code tags? They are working as you showed me, just not represented

Overall its nice to see that its straight forward after all, just add one manipulation after the other, I like it:)

Thanks again,
cheers

[Anaisdg]

@RandGH,
Yes learning Flux definitely takes an adjustment in thinking and approach but after you get used to It I generally think that’s its easier. It makes me happy to hear ya’ll are getting along.

1. Hm not all Flux functions are available in 1.x. Obviously you could upgrade.
   Otherwise you could maybe use the timeShift() function?
   timeShift() function | Flux 0.x Documentation

2. Hmm I’m not sure. The yield() function should be working. What error are you getting? Could it perhaps be the quotes? I see:

|> yield(name: “cumulative sum”)

instead of

|> yield(name: "cumulative sum")

3. What do you mean by button for the code tags? I the influxDB ui in the query builder?

of course!

[RandGH]

Hi,
Re 1 - Migrating to v2 is not possible at this point due to some legacy stuff i run. I would need to go dual stack and I didn’t want to go there (yet).
Timeshift seems to be static, so while it probably would work for shorter periods, its going to cause errors when crossing DST borders… doesnt sound ideal either… which is probably why you implemented timezone. Are there any plans to backport this to 1.8?

Re 2 - The new quotes cause an error (invalid expression), same as single quotes. With double quotes there is no error, but the series is not renamed at all, it stays at " ". This seems to be a frequently encountered issue since i found various posts providing workarounds.

re 3 - no I was referring to the editor for the forum here which has no button for a code tag, or i have no found the option to enable it;)

Edit - is flux more expensive than influxql in terms of processing cost?
If i run
SELECT cumulative_sum(SUM(“power”::float)) / 60.0 as “AktuellerBezug_influx”
FROM (
SELECT mean(“value”) AS power FROM “javascript.0.Volkszaehler.AKTUELLER_Bezug” WHERE $timeFilter GROUP BY time(1m) tz(‘Europe/Berlin’)
)
GROUP BY time(30d) tz(‘Europe/Berlin’)
over one year this is working perfectly fine.
However if i run
FinalOutput = ["_field", “_time”, “_value”]

RawSeries=from(bucket: “iobroker/autogen”)
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> filter(fn: (r) => r["_measurement"] == “javascript.0.Volkszaehler.AKTUELLER_Bezug”)
|> filter(fn: (r) => r["_field"] == “value”)
|> aggregateWindow (every:1m, fn: mean, createEmpty: true)
|> toFloat()
|> aggregateWindow(every: 30d, fn: sum, createEmpty: true)
|> map(fn: (r) => ({ r with _value: r._value/60.0}))
|> cumulativeSum()

NamedSeries = RawSeries
|> map(fn: (r) => ({_value:r._value, _time:r._time, _field:“AktuellerBezug_flux”}))
|> keep(columns:FinalOutput)

NamedSeries |> yield()

I get a timeout error

Thanks,
cheers

[RandGH]

Any idea re the timeout issue @Anaisdg ?

[RandGH]

@Anaisdg No idea at all ?

[Anaisdg]

Hello @RandGH,
A ton of work on optimizing flux has been performed on 2.x. I don’t know that I would recommend using Flux on 1.8 if you’re looking for good performance.
As for this snippet:

NamedSeries = RawSeries
|> map(fn: (r) => ({_value:r._value, _time:r._time, _field:“AktuellerBezug_flux”}))
|> keep(columns:FinalOutput)

You could just use set for the _field instead of mapping the _value and _time column again (that part seems redundant)

  |> set(key: "_field", value: "AktuellerBezug_flux")

[RandGH]

Well I would not have said I’d be looking for “good” performance, but not crashing would have been appreciated…

But ok, if the takeaway is that Flux @ 1.8 is not capable to handle large amounts of data then I guess I’ll have to stay away for the time being until I can move to 2.x

Thanks for your help.

Cheers