Normalized measurement starting at value 1

Fluxlang
1

I want to normalize a measurement such that it starts by value 1. For a fixed starting point this is no problem using the code:

start_value = 12930.0
from(bucket: "finance")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "price")
  |> filter(fn: (r) => r["Stock_Name"] == "DAX 30")
  |> filter(fn: (r) => r["ISIN"] == "DE0008469008")
  |> map(fn: (r) => ({ r with _value: r._value / start_value}))

But I don’t now how to code a dynamical range. For this I need to inject a calculated start_value (depending on the range) into the code. The following line:

  |> map(fn: (r) => ({ r with _value: r._value / first()._value}))

does not work. I get "type error @8:50-8:57: missing required argument tables".

I would be happy if someone can help. Thank you very much.

2

@jennerwein first() operates on a stream of tables, which is why you’re getting the error you’re getting. You need to extract the first value as a scalar value to use it in the map operation.

data = from(bucket: "finance")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "price")
  |> filter(fn: (r) => r["Stock_Name"] == "DAX 30")
  |> filter(fn: (r) => r["ISIN"] == "DE0008469008")

start_value = {
  values = data |> first() |> findColumn(fn: (key) => true, column: "_value")
  return values[0]
}

data
  |> map(fn: (r) => ({ r with _value: r._value / start_value}))
3

@scott Thanks for the answer. I understand that I have to extract the first value as a scalar value.
But testing your code with the latest version of InfluxDB 2.0 I got an error at the assignement start_value = ....

Then I tested the code with the assignement start_value = 1000.0 which ran smoothly.

But even the simple assignment

start_value = { return 1000.0 }

threw the error compilation failed: error at @7:17-7:29: unexpected token for property key: RETURN (return).

4

Try this:

data = from(bucket: "finance")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "price")
  |> filter(fn: (r) => r["Stock_Name"] == "DAX 30")
  |> filter(fn: (r) => r["ISIN"] == "DE0008469008")

values = data |> first() |> findColumn(fn: (key) => true, column: "_value")
start_value = values[0]

data
  |> map(fn: (r) => ({ r with _value: r._value / start_value}))
5

@scott Now I get the error An internal error has occurred.

When I set values = [1000.0, 200.0] it works fine. But the line

values = data |> first() |> findColumn(fn: (key) => true, column: "_value")

throws the error An internal error has occurred.
Is it possible that this is not even supported in Influxdb 2.0?
(cp. Extract scalar values in Flux | InfluxDB OSS 2.6 Documentation)

6

What beta version of 2.0 are you using? I just tried this method on my own data set and it worked.

7

@scott I use docker Version 2.0.0 (c8af0f3). ( v2.0.0-beta.14 [2020-07-08])

8

Looks like you’re running beta-14. It’s definitely supported. Does data |> first() return any data?

9

@scott values = data |> first() returns one row.

11

@scott I found a partial solution:
The line

 values = data |> first()

as well as the line

 values = data |> findColumn(fn: (key) => true, column: "_value")

work, but not the line

values = data |> first() |> findColumn(fn: (key) => true, column: "_value")

Do you have an explanation for this? Thanks!

12

I don’t have an explanation, but that essentially does the same thing. The only difference is that without first(), findColumn() returns an array of all the values in the _value column (which will be multiple). first() reduces the table to a single row, but findColumn() should still be able to return an array with that single value. I’m not sure why it’s failing.

13

Thanks, now it works for me!
Interestingly, the program also works for a table with only one row.
Only if you have first() in the table flow (before findColumns) you get the internal error.