Downsampling Task fail with error (unsupported input type for mean aggregate)

What I want
I want to downsample all of my data (with mean) from my raw bucket “telegraf” (frequency = 30s, retention 30 days) like this:

  • telegraf → telegraf_90d (frequency = 1h, retention 90 days)
  • telegraf_90d → telegraf_365d (frequency = 12h, retention 365 days)

What I get / Error
Unfortunately I get following error:
could not execute task run: unsupported input type for mean aggregate: string

What is the cause?
I have some collectors like NetApp Harvest that unfortunately writes sometimes strings or booleans in “_value” and that’s why I get the error above.

What I want to prevent / What is my goal

  • I want a working downsampling task that works with all supported data types without specifying them hardcoded.
  • I don’t want to exclude all the _measurements or _fields that have strings in “_value”.

Acceptable Workaround (when we don’t find any solution)
Exclude all other data types that not contains a numeric value in “_value” from downsampling.
I thought as a QuickFix exclude all non numeric data with regex, but this won’t work and I don’t get any help (Bug? Strange regex behaviour in Flux · Issue #3804 · influxdata/flux · GitHub)

→ But I would be very happy If we find a solution for my issue and not a workaround.

My downsampling task (one of them)

option task = {name: "task_telegraf_90d", every: 1h}

data = from(bucket: "telegraf")
	|> range(start: -duration(v: int(v: task.every) * 2))
	|> filter(fn: (r) =>
		(r._measurement =~ /.*/))

data
	|> aggregateWindow(fn: mean, every: 1h)
	|> filter(fn: (r) =>
		(exists r._value))
	|> to(bucket: "telegraf_90d", org: "MYORG")

Additional Informations
InfluxDB: Version 2.0.5
VM: 8 vCores & 128GB Memory

This is extremely important for us! I am happy about any help.

Hi @fluxator,
Could you try something like this using a custom function and conditions:
(Based on my data)

import "types"
aggregate_type = (tables=<-) =>
{
extract = tables
    |> findRecord(
      fn: (key) => true,
      idx: 0
    )
 
 temp = if types.isType(v: extract["_value"], type: "int") then  tables |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  else if types.isType(v: extract["_value"], type: "float") then tables |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  else tables |> aggregateWindow(every: v.windowPeriod, fn: last, createEmpty: false)
 
return temp

}

from(bucket: "Jetson")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "exec_jetson_stats")
  |> aggregate_type()
  |> yield(name: "result")
  |> to(bucket: "telegraf_90d", org: "MYORG")
data = from(bucket: "Jetson")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "exec_jetson_stats")

int_data = data
  |> filter(fn: (r) => types.isType(v: r._value, type: "int"))
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)

float_data = data
  |> filter(fn: (r) => types.isType(v: r._value, type: "float"))
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)

string_data = data
  |> filter(fn: (r) => types.isType(v: r._value, type: "string"))
  |> aggregateWindow(every: v.windowPeriod, fn: last, createEmpty: false)

@Jay_Clifford & @Anaisdg
Thanks, that is exactly what I need. I will test this and give feedback here. :slight_smile:

But first in which InfluxDB version is this supported?
image

Because I in the release notes of the InfluxDB GitHub repo the Flux version 0.140.0+ is not mentioned.
The last mentioned Flux Update is the InfluxDB 2.1.0 to Flux v0.139.0.

1 Like

Arghhh was just looking at this, could do with knowing when this will be available to use

If you’re using InfluxDB Cloud, it’s already available. If you’re using InfluxDB OSS 2.x, support for types.isType() will ship with InfluxDB 2.2, which should be later this week or early next week.

2 Likes

Unfortunately we are not allowed to use InfluxDB Cloud. So I have to wait for version 2.2 and then I give you feedback from my testing. Thanks anyway already for any help. :blush::+1:

I know you probably cannot commit to it 100% but if you can do we know if this is still likely to be released soon? (next week or so)

@jon_griffiths There was a minor defect found in one of the new 2.2 features that has pushed the release back a bit. I don’t know the exact timeline, but it is still expected soon.

2 Likes

@scott is there another way of checking the type of a variable?

@deb, without Flux 0.140.0+, no, it’s not possible. You have three options to use a version of InfluxDB with a version of Flux that includes types.isType():

  1. Use InfluxDB Cloud
  2. Build InfluxDB OSS from source
  3. Wait for the InfluxDB OSS 2.2 release (should be coming soon, working through a defect in a new feature)

Sounds good, thank you

can we track the defect in an issue in Git? so we can see current status?

Any update on timescales?

@jon_griffiths @fluxator @deb InfluxDB 2.2 just dropped. It shipped with Flux 0.150.0 and includes support for types.isType().

1 Like