重新同步 InfluxDB 值的时间戳

Question

我有几个值被放入不同主题下的 InfluxDB 中。它们通过 MQTT/JSON 提供并由 Telegraf 处理。每条 JSON 记录都会生成不同系列条目的元组，这些条目具有稍微不同的时间戳。 Δ 通常低于 1 毫秒，而 JSON 事件相隔几秒，因此检测应该是可管理的。

为了进一步分析，有必要对条目进行重新分组，以便再次合并元组。在大多数情况下，简单地四舍五入时间戳就可以了，但当然不是一般情况下，因为边界可能就在这样的元组之间。

有什么想法吗？我正在通过 Python 访问数据，因此在 Python 中进行适当的 Influx 查询或处理都可以。当然，我可以循环扫描数据并决定如何将它们放在一起，但我想知道手头是否有一个优雅的解决方案，也许可以使用常见的 Python 库之一，例如 NumPy 或 Pandas。我想我不是唯一遇到这种问题的人。

Answer 1

您可以使用 Flux 处理数据，并在时间戳上使用日期命令将其截断到不同的粒度，例如 date.minute() 或 date.millisecond()。

也许创建一个新字段，例如 newg，然后使用 set 将 _time 字段替换为它，或者调整现有查询以使用新列？

|> map(fn: (r) => ({ r with newg: date.millisecond(t: r._time) }))

Answer 2

好的，我选择在 Python 中处理我的数据。这就是我想出的。它可以处理任意粒度间隔。对于我正在处理的数据量，它足够快；除了排序，它是 O(n)。

# input is res: dictionary of {timestamp: (topic, value)}
DELTA = timedelta(seconds=0.5) # granularity; adjust to needs
t = datetime(dt.MINYEAR, 1, 1, tzinfo=dt.timezone.utc)  # init running tstamp var
t_group = t # timestamp for sample group
outlist = [] # output list
group = None # current sample group
for key, (topic, val) in sorted(res.items()):
  if (key - t) > DELTA: # new group
    t_group = key
    if group: outlist.append(group) # save previous group to output list
    group = {"time": t_group} # init new group with 1st timestamp
  print(f"{t_group}\t{key}\t{topic} = {val}\t∆={key-t}")
  group[topic] = val # add to group
  t = key
print(f"\n{len(outlist)} entries extracted.\n")

重新同步 InfluxDB 值的时间戳

Re-synchronizing timestamps of InfluxDB values

python

pandas

numpy

influxdb

telegraf