InfluxDB降采样时保留最值及对应时间戳的实现方法

合并min/max序列的方法 #

你当前的查询生成两个独立序列，可以通过以下两种方式合并为单表：

方法1：用join关联窗口数据 #

将min和max结果按窗口时间、测量值等维度关联，合并到同一行：

data = from(bucket: "test-bucket")
  |> range(start: 0, stop: now())
  |> filter(fn: (r) => r["_field"] == "sensorValue")

minData = data
  |> aggregateWindow(every: 1s, fn: min, createEmpty: false)
  |> rename(columns: {_value: "min_value", _time: "window_time"})

maxData = data
  |> aggregateWindow(every: 1s, fn: max, createEmpty: false)
  |> rename(columns: {_value: "max_value"})

join(
  tables: {min: minData, max: maxData},
  on: ["window_time", "_measurement", "_field", "_start", "_stop"]
)
|> yield(name: "min_max_combined")

方法2：用union+pivot转换结构 #

先合并两个序列，再将聚合类型转为列：

data = from(bucket: "test-bucket")
  |> range(start: 0, stop: now())
  |> filter(fn: (r) => r["_field"] == "sensorValue")

minMaxCombined = union(tables: [
  data |> aggregateWindow(every: 1s, fn: min, createEmpty: false) |> set(key: "agg_type", value: "min"),
  data |> aggregateWindow(every: 1s, fn: max, createEmpty: false) |> set(key: "agg_type", value: "max")
])
|> pivot(rowKey: ["_time"], columnKey: ["agg_type"], valueColumn: "_value")
|> rename(columns: {min: "min_value", max: "max_value"})

minMaxCombined |> yield(name: "min_max_pivoted")

获取极值对应的原始时间戳 #

注意：上面的方法仅返回窗口时间，而非极值点的具体时间。要保留你示例中(27,5)、(2,7)这种极值+对应时间的结果，需要自定义聚合逻辑：

data = from(bucket: "test-bucket")
  |> range(start: 0, stop: now())
  |> filter(fn: (r) => r["_field"] == "sensorValue")
  |> aggregateWindow(
    every: 1s,
    fn: (tables=<-) =>
      tables
      |> reduce(
        identity: {min_val: 0.0, min_time: 0, max_val: 0.0, max_time: 0},
        fn: (r, accumulator) => ({
          min_val: if r._value < accumulator.min_val then r._value else accumulator.min_val,
          min_time: if r._value < accumulator.min_val then r._time else accumulator.min_time,
          max_val: if r._value > accumulator.max_val then r._value else accumulator.max_val,
          max_time: if r._value > accumulator.max_val then r._time else accumulator.max_time,
        })
      ),
    createEmpty: false
  )

data |> yield(name: "min_max_with_timestamps")

这个自定义reduce会遍历窗口内的每条数据，记录最小值、最小值对应的时间，以及最大值、最大值对应的时间，完全匹配你的需求。

优化建议 #

• 20kHz采样频率下，1s窗口包含2万条数据，自定义聚合性能略低于内置函数，但InfluxDB可轻松处理；若追求极致性能，可考虑写入时提前做局部聚合。
• 若仅需极值及对应时间，直接使用自定义聚合方案即可，无需合并多个序列。

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