InfluxDB：如何用Flux添加连续行/时间戳差值列？

要计算G_PAED列连续行的差值，不能直接用map()（它仅能处理当前行数据），需要使用**statefulMap()**函数——它可以保留上一行的状态值，用来和当前行完成差值计算。

完整查询示例 #

假设你获取初始数据的基础查询如下：

from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "your-measurement")
  |> filter(fn: (r) => r._field == "G_PAED")

在此基础上添加statefulMap()计算差值：

from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "your-measurement")
  |> filter(fn: (r) => r._field == "G_PAED")
  |> statefulMap(fn: (r, state) => {
      // 计算当前值与上一行值的差值，第一行无前置值时设为0.0（可按需改为null）
      diff = if exists state.prevValue then r._value - state.prevValue else 0.0
      // 更新状态，将当前值传给下一行作为前置参考值
      return {r with _diff: diff}, {prevValue: r._value}
    }, initialState: {prevValue: 0.0})

关键说明 #

• statefulMap()包含两个核心参数：行处理函数，以及初始状态initialState
• 处理函数中的state会保留上一次返回的状态值，首次执行时使用initialState定义的初始值
• 最终会新增_diff列，存储G_PAED列连续行的差值
• 确保数据按时间顺序排列（Flux的range查询默认升序，若需强制排序可追加|> sort(columns: ["_time"])）

内容的提问来源于stack exchange，提问作者Ahmad Mujtaba