InfluxDB 1.6/2实现仅填充<10分钟间隙的线性插值查询

InfluxDB 1.6（InfluxQL）实现 #

InfluxQL没有原生支持带间隙阈值的填充，需要通过子查询筛选出间隙符合要求的数据段，再进行分组填充：

SELECT mean(value) AS value
FROM (
  -- 子查询：计算每个数据点与前一个点的时间间隙
  SELECT value, time,
    LAG(time) OVER (ORDER BY time) AS prev_time,
    time - LAG(time) OVER (ORDER BY time) AS gap
  FROM your_measurement
  WHERE time >= '2024-01-01T00:00:00Z' AND time <= '2024-01-01T12:00:00Z'
)
-- 筛选间隙≤10分钟（600000毫秒）的点，保留第一个数据点（gap为NULL）
WHERE gap IS NULL OR gap <= 600000
GROUP BY time(10s)
-- 对筛选后的分组数据应用线性填充
FILL(linear)

说明 #

1. 子查询通过LAG(time)获取前一个数据点的时间，计算当前点与前点的时间差gap；
2. 外层查询只保留间隙≤10分钟的点，以及序列的第一个点；
3. 按10秒分组后，FILL(linear)仅对筛选范围内的间隙进行填充，超过10分钟的间隙段会保留NULL或无数据返回。

InfluxDB 2.x（Flux）实现 #

Flux支持更灵活的条件填充，可以直接在fill()函数中设置间隙阈值判断：

// 1. 获取原始数据并按10秒窗口聚合，保留空窗口（createEmpty: true）
aggregatedData = from(bucket: "your-bucket")
  |> range(start: 2024-01-01T00:00:00Z, stop: 2024-01-01T12:00:00Z)
  |> filter(fn: (r) => r._measurement == "your_measurement" and r._field == "value")
  |> aggregateWindow(every: 10s, fn: mean, createEmpty: true)
  |> sort(columns: ["_time"])

// 2. 带条件的线性填充：仅间隙≤10分钟时填充
aggregatedData
  |> fill(
      fn: linearFill,
      columns: ["_value"],
      condition: (r, idx) => {
        // 查找当前点之前最近的非空值
        prevNonNull = (idx..0)
          |> map(fn: (i) => aggregatedData[i])
          |> find(fn: (v) => exists v._value)
          |> get(idx: 0, default: null)
        // 查找当前点之后最近的非空值
        nextNonNull = (idx..length(arr: aggregatedData)-1)
          |> map(fn: (i) => aggregatedData[i])
          |> find(fn: (v) => exists v._value)
          |> get(idx: 0, default: null)
        
        // 仅当前后都有非空值且间隙≤10分钟时，执行填充
        if exists prevNonNull and exists nextNonNull then
          duration(v: nextNonNull._time - prevNonNull._time) <= 10m
        else
          false
      }
    )
  |> yield(name: "conditional_filled_data")

说明 #

1. aggregateWindow按10秒生成时间窗口，createEmpty: true确保缺失数据的窗口保留为NULL；
2. fill()的condition参数判断当前NULL点是否处于两个非空点之间，且两点间隙≤10分钟；
3. 仅满足条件的NULL点会被线性填充，超过10分钟的间隙段将保留NULL。

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