InfluxDB 2 Flux 自定义窗口聚合函数查询及差值计算实现咨询

原代码问题说明 #

你的写法存在两个核心错误：

• 传入aggregateWindow的自定义函数入参tables是表流对象，不能直接做算术运算，需要先提取对应列的数值
• 自定义聚合函数的返回值必须是符合Flux规范的表流，需要包含_time、_value等聚合后的标准字段，不能直接返回数值

正确实现代码 #

// 自定义聚合函数：计算窗口内指定列首尾值的差值
difference = (column = "_value", tables = <-) =>
  tables
    // 提取窗口内第一条记录的目标值
    |> first(column: column)
    |> rename(columns: {[column]: "first_val"})
    // 关联同窗口内最后一条记录的目标值
    |> join(
        tables: tables |> last(column: column) |> rename(columns: {[column]: "last_val"}),
        // 此处填写你实际用到的所有标签键，无额外标签可保留公共字段即可
        on: ["_start", "_stop", "_measurement", "_field"]
    )
    // 计算差值并生成标准聚合输出字段
    |> map(fn: (r) => ({
        _time: r._stop,
        _value: r.last_val - r.first_val,
        _measurement: r._measurement,
        _field: r._field
        // 若有自定义标签可在此处补充返回
      }))

// 完整查询逻辑
from(bucket: "my-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "simple")
  |> filter(fn: (r) => r["_field"] == "value")
  |> aggregateWindow(
      every: 1mo,
      fn: difference,
      column: "_value",
      timeSrc: "_stop",
      timeDst: "_time",
      createEmpty: true
    )
  |> yield(name: "diff")

自定义聚合函数入参规范说明 #

所有可传入aggregateWindow的自定义聚合函数都需要满足以下要求：

• 必须包含tables = <-作为管道参数，接收上游传入的窗口表流
• 必须包含column参数，指定要聚合的列名
• 返回值必须是包含_time、_value两个必填字段的表流，可额外携带标签、measurement、field等元信息

内容的提问来源于stack exchange，提问作者Sergey Shcherbakov