重新同步 InfluxDB 值的时间戳

如何解决重新同步 InfluxDB 值的时间戳

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

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

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

解决方法

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

也许创建一个新字段，例如 newg 并使用 set 用它替换 _time 字段或调整您现有的查询以使用新列？

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

,

好的，我选择在 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,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")