如何用Flux查询聚合并统计连续重复的日志消息？

你可以通过以下Flux查询实现连续重复消息的分组，获取每组的起始时间、消息内容及重复次数：

from(bucket: "my_bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> sort(columns: ["_time"])
  // 标记当前消息与上一条是否不同，用于划分分组
  |> derive(column: "is_new_group", fn: (r) => r._value != previous(r._value))
  // 生成连续分组ID，每次消息变化时ID递增
  |> cumulativeSum(column: "group_id", fn: (r) => if r.is_new_group or exists(r.is_new_group) == false then 1 else 0)
  // 按分组ID和消息内容分组
  |> group(columns: ["group_id", "_value"])
  // 聚合每组的起始时间、消息内容和重复次数
  |> aggregateWindow(
    every: inf,
    fn: (table) => ({
      _time: min(table._time),
      _value: first(table._value),
      count: rowCount(table)
    })
  )
  // 清理辅助字段并按时间排序
  |> drop(columns: ["group_id", "is_new_group"])
  |> sort(columns: ["_time"])

关键步骤说明： #

1. 排序：sort(columns: ["_time"])确保数据按时间升序排列，这是后续判断连续重复的基础。
2. 标记分组边界：derive创建is_new_group字段，当当前消息与上一条不同时标记为true，第一条数据因无前置数据也会被标记为新分组。
3. 生成分组ID：cumulativeSum通过累加生成唯一分组ID，相同连续消息会被分配到同一ID下。
4. 分组聚合：按分组ID和消息内容分组后，用aggregateWindow聚合每组的最小时间（起始时间）、消息内容（同组一致）和行数（重复次数）。
5. 清理结果：移除辅助字段并重新按时间排序，得到目标格式的结果。

如果你的数据包含额外标签（如_measurement），可以在group步骤中加入对应标签，确保分组的准确性。

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