InfluxDB中如何筛选正值并拆分计量数据计算进出口电费？

InfluxDB 1.8（使用InfluxQL） #

通过筛选正负值分别处理购电（功率为负）和售电（功率为正）数据，再用integral()函数计算电量，最后转换为kWh单位即可。

1. 计算购电量（importing） #

筛选所有功率小于0的数据，取绝对值后积分（将瓦时转换为千瓦时）：

SELECT integral(abs("power"), 1h) / 1000 AS "import_kWh"
FROM "grid_measurement" -- 替换为你的测量表名
WHERE "power" < 0 AND time >= now() - 7d -- 替换为你的时间范围
GROUP BY time(1d) -- 按天统计，可按需调整时间间隔

2. 计算售电量（exporting） #

筛选所有功率大于0的数据直接积分：

SELECT integral("power", 1h) / 1000 AS "export_kWh"
FROM "grid_measurement"
WHERE "power" > 0 AND time >= now() - 7d
GROUP BY time(1d)

3. 合并购售电数据查询 #

如果需要在一个结果中同时看到购售电数据，可使用UNION ALL：

SELECT time, "import_kWh", 0 AS "export_kWh"
FROM (
    SELECT time, integral(abs("power"), 1h)/1000 AS "import_kWh"
    FROM "grid_measurement"
    WHERE "power" < 0 AND time >= now() - 7d
    GROUP BY time(1d)
)
UNION ALL
SELECT time, 0 AS "import_kWh", "export_kWh"
FROM (
    SELECT time, integral("power", 1h)/1000 AS "export_kWh"
    FROM "grid_measurement"
    WHERE "power" > 0 AND time >= now() - 7d
    GROUP BY time(1d)
)
ORDER BY time

InfluxDB 2.x（使用Flux） #

Flux提供了更灵活的数据处理能力，通过filter和integrate函数实现拆分计算：

1. 计算购电量 #

from(bucket: "your_bucket") -- 替换为你的桶名
  |> range(start: -7d) -- 替换为时间范围
  |> filter(fn: (r) => r._measurement == "grid_measurement" and r._field == "power" and r._value < 0)
  |> map(fn: (r) => ({ r with _value: abs(r._value) })) -- 取绝对值转换为正数值
  |> aggregateWindow(every: 1d, fn: integrate) -- 按天积分
  |> map(fn: (r) => ({ r with _value: r._value / 1000, _field: "import_kWh" })) -- 转换为kWh

2. 计算售电量 #

from(bucket: "your_bucket")
  |> range(start: -7d)
  |> filter(fn: (r) => r._measurement == "grid_measurement" and r._field == "power" and r._value > 0)
  |> aggregateWindow(every: 1d, fn: integrate)
  |> map(fn: (r) => ({ r with _value: r._value / 1000, _field: "export_kWh" }))

3. 合并购售电数据查询 #

通过experimental.join合并两个数据集：

import "experimental"

// 加载原始数据
rawData = from(bucket: "your_bucket")
  |> range(start: -7d)
  |> filter(fn: (r) => r._measurement == "grid_measurement" and r._field == "power")

// 计算购电量
importKwh = rawData
  |> filter(fn: (r) => r._value < 0)
  |> map(fn: (r) => ({ r with _value: abs(r._value) }))
  |> aggregateWindow(every: 1d, fn: integrate)
  |> map(fn: (r) => ({ r with _value: r._value / 1000, _field: "import_kWh" }))

// 计算售电量
exportKwh = rawData
  |> filter(fn: (r) => r._value > 0)
  |> aggregateWindow(every: 1d, fn: integrate)
  |> map(fn: (r) => ({ r with _value: r._value / 1000, _field: "export_kWh" }))

// 合并结果
experimental.join(
  tables: {import: importKwh, export: exportKwh},
  on: ["_time"]
)

关键说明 #

• integral()函数的结果单位是瓦时（Wh），除以1000转换为千瓦时（kWh），这是电费计算的标准单位。
• 得到kWh数据后，直接乘以对应费率即可算出购电/售电成本。
• 你之前尝试的min()/max()是取时间区间内的极值，确实不适合拆分连续的功率数据，筛选正负值才是正确的方向。

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