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

InfluxDB 2 Tasks
,
1

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.

2

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")
3 
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)
1 Like
4

@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
5

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

6

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
7

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:

8

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)

9

@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
10

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

11

@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)
12

Sounds good, thank you

13

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

14

Any update on timescales?

15

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

1 Like