InfluxDB: Flux - Aggregation Function reduce()

Example

Because the raw data in my example measurement "asset_hyperv_local" has a lot of tags I use "keep()" to decrease the number of columns.

Flux

bucket = "<YOUR_BUCKET>"

from(bucket: bucket)
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => 
    r._measurement == "asset_hyperv_local" and
    r._field == "physicalmemory" and
    r.cluster =~ /C1TN7.*/
  )
  |> last()
  |> group(columns: ["cluster", "manufacturer"])
  |> keep(columns: ["_time", "_value", "cluster", "manufacturer", "_field"])

Flux with the function "reduce()" in action

bucket = "<YOUR_BUCKET>"

from(bucket: bucket)
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => 
    r._measurement == "asset_hyperv_local" and
    r._field == "physicalmemory" and
    r.cluster =~ /C1TN7.*/
  )
  |> last()
  |> group(columns: ["cluster", "manufacturer"])
  |> keep(columns: ["_time", "_value", "cluster", "manufacturer", "_field"])
  |> reduce(
    fn: (r, accumulator) => ({ 
      NumberOfNodes: accumulator.NumberOfNodes + 1.0,
      SumOfPhysicalMemory: r._value + accumulator.SumOfPhysicalMemory,
    }), 
    identity: {NumberOfNodes: 0.0, SumOfPhysicalMemory: 0.0}
  )