InfluxDB 1.8 Flux查询时如何修改异常传感器数据？

一、直接替换为固定值 #

如果只需要把超出阈值的异常值替换成固定合理值（比如35℃），用map函数结合条件判断即可，完全不会修改数据库原始数据：

from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "sensor" and r._field == "temperature")
  |> map(fn: (r) => ({
      r with
      _value: if r._value > 100.0 or r._value < -20.0 then 35.0 else r._value
    })
  )

这段代码会遍历每一条数据，判断温度值是否超出[-20, 100]的合理范围，超出则替换为35℃，否则保留原始值。

二、用前后正常数据生成替换值 #

InfluxDB 1.8没有内置插值函数，要生成更贴合实际的替换值，需要手动实现逻辑，以下两种常用方式：

1. 用前一个正常值填充异常点 #

适合异常点是短暂波动的场景，用state函数追踪上一个有效的数据值：

from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "sensor" and r._field == "temperature")
  |> sort(columns: ["_time"]) // 必须按时间排序，保证state追踪顺序正确
  |> state(fn: (r, state) => {
      // 异常值用上一个有效值替换，正常值则更新追踪的有效值
      currentValue = if r._value > 100.0 or r._value < -20.0 then state.lastValid else r._value
      return {
        lastValid: if r._value <= 100.0 and r._value >= -20.0 then r._value else state.lastValid,
        newValue: currentValue
      }
    }, initial: {lastValid: 35.0, newValue: 35.0}) // 初始有效值设为35℃，可按需调整
  |> map(fn: (r) => ({r with _value: r.newValue})) // 将新值赋值给原始字段
  |> drop(columns: ["lastValid", "newValue"]) // 清理中间状态列

2. 用前后最近正常值的平均值填充 #

适合需要更平滑过渡的场景，先筛选正常数据，再为每个异常点匹配前后最近的有效值计算平均值：

// 第一步：获取原始数据并标记是否有效
rawData = from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "sensor" and r._field == "temperature")
  |> map(fn: (r) => ({
      r with
      is_valid: r._value <= 100.0 and r._value >= -20.0
    })
  )

// 第二步：提取所有正常数据，用于后续匹配
validData = rawData 
  |> filter(fn: (r) => r.is_valid) 
  |> keep(columns: ["_time", "_value"])

// 第三步：遍历原始数据，替换异常值
rawData
  |> map(fn: (r) => {
      // 正常数据直接返回
      if r.is_valid then return r
      
      // 查找当前异常点之前最近的正常数据
      prevValid = validData 
        |> range(start: -inf, stop: r._time) 
        |> last() 
        |> findRecord(fn: (key) => true, idx: 0)
      
      // 查找当前异常点之后最近的正常数据
      nextValid = validData 
        |> range(start: r._time, stop: +inf) 
        |> first() 
        |> findRecord(fn: (key) => true, idx: 0)
      
      // 计算替换值：优先取前后平均值，无前后数据则用固定值35℃
      avgValue = if prevValid == null and nextValid == null then 35.0 
                else if prevValid == null then nextValid._value 
                else if nextValid == null then prevValid._value 
                else (prevValid._value + nextValid._value) / 2.0
      
      return {r with _value: avgValue}
    }
  )
  |> drop(columns: ["is_valid"]) // 清理标记列

注意：这种方式对大数据量的查询性能有一定影响，可通过缩小range范围优化。

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