Couldn't reduce with pivot and group

[Duck]

Below code is perfectly working and separating users table with “pivot” and “group”.

from(bucket:"statistics:daily")
    |> range(start:-1d)
    |> filter(fn: (r) => r["_measurement"] == "players")
    |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
    |> group(columns: ["user"], mode: "by")

but whenever I try to “reduce” it, Flux says the fields are empty.

Latest code

from(bucket:"statistics:daily")
    |> range(start:-1d)
    |> filter(fn: (r) => r["_measurement"] == "players")
    |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
    |> group(columns: ["user"], mode: "by")
    |> reduce(fn: (r, accumulator) => ({
                    user: r.user,
                    _time: r._time,
                    _measurement: r._measurement,
                    _field: r._field,
                    _value: float(v: r._value) + accumulator._value
    }), identity: {_time: now(), _measurement: "", user: "", _field: "", _value: 0.0})

Latest code error (NOTE: There are no empty fields or values, I’m sure about it. If you use this code without reducing, it’d work as expected.)

runtime error @6:8-12:87: reduce: null values are not supported for “_field” in the reduce() function

Even I check and set the default “field”, Flux continues to say “null for _value”. I guess it drops all the things whenever I switch to “reduce” aggregate.

Is it impossible to use reduce with pivot and group?

[Jay_Clifford]

Hi @Duck,
This is due to the fact that _field and _value are considered null when you pivot by field. let’s take a look at my current data:

Before the pivot you can see _field and _value are represented accordingly and would work against your reduce. After the pivot toy can see we drop the _field and _vaue field. These are replaced by the column jetson_CPU1 with the values represented in this field.

Here is an example of using reduce with a pivot before.

raw = from(bucket: "Jetson")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "exec_jetson_stats")
  |> filter(fn: (r) => r["_field"] == "jetson_CPU1")
  |> last()
  |> yield(name: "before_pivot")
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> yield(name: "After_pivot")

raw
    |> reduce(fn: (r, accumulator) => ({
                    _time: r._time,
                    _measurement: r._measurement,
                    jetson_CPU1: r.jetson_CPU1,
                    _value: float(v: r.jetson_CPU1) + accumulator._value
    }), identity: {_time: now(), _measurement: "", jetson_CPU1: 0.0, _value: 0.0})
      |> yield(name: "After_reduce")

Note if you wanted to do it your way which is perfectly reasonable then you should pivot after reduce.

[Duck]

Thank you so much for the fast answer, it helped me. I’ve figured out how to deal with it and write an aggregation in Java language.

The problem I am facing now is to use to to transfer my data to another bucket. After using pivot, group, and reduce, it is almost impossible to transfer my grouped and reduced data(Example: ‘_result, user=1’ and 'result, user=2).

Is there a way to transfer aggregated data to another bucket? I can do it but it just gets the latest field. For example, if I have 3 users grouped by the “user” field, it writes “user3” only, not “user1” and “user2”.

BEFORE (bucket: daily)

AFTER (bucket: overall)

[Jay_Clifford]

No problem at all. Could you try ungrouping before the to() function? So essentially do reduce() then

|> group() //ungroup 

then do to().

[Duck]

Ungrouping works but the result is the same. It just wrote the last user instead of all users.

BEFORE

AFTER

[Duck]

I’ve kind of solved with using experimental.to without regrouping.

[Duck]

After taking a look at it, it is not solved. It is working as I want but user field becomes tag. If I regroup with just measurement, it just resulted as same as first time(getting only last player, u_2)