InfluxDB与Grafana Geomap：多Topic间时间戳同步问题

一、在Telegraf阶段避免时间戳差异 #

由于Telegraf逐个处理MQTT消息，lat和lon会被分别打上到达时的时间戳，导致微小差异。可以通过以下两种方式将二者合并为同一时间戳的单点：

方法1：用Starlark处理器缓存并合并坐标 #

配置Telegraf的starlark处理器，临时缓存收到的lat或lon值，当两个值都收集完成后，生成包含lat和lon字段的统一指标再写入InfluxDB，确保时间戳完全一致。

示例Telegraf配置片段：

[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt-broker:1883"]
  topics = [
    "foo/bar/lat",
    "foo/bar/lon"
  ]
  data_format = "value"
  data_type = "float"

[[processors.starlark]]
  source = '''
cache = {}

def apply(metric):
    # 从主题提取坐标类型（lat/lon）
    topic_parts = metric.tags["topic"].split("/")
    coord_type = topic_parts[-1]
    # 若有设备ID，建议加入key中避免不同设备坐标混淆，这里用固定key示例
    key = "location"
    # 缓存当前值与时间戳
    if key not in cache:
        cache[key] = {}
    cache[key][coord_type] = metric.fields["value"]
    cache[key]["time"] = metric.time
    # 凑齐两个坐标值后生成新指标
    if "lat" in cache[key] and "lon" in cache[key]:
        new_metric = Metric("location")
        new_metric.time = cache[key]["time"]
        new_metric.fields["lat"] = cache[key]["lat"]
        new_metric.fields["lon"] = cache[key]["lon"]
        del cache[key]  # 清除缓存准备下一次事件
        return [new_metric]
    # 仅收到单个值时不写入，返回空列表
    return []
'''

[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "your-token"
  organization = "your-org"
  bucket = "your-bucket"

方法2：用Aggregate处理器按时间窗口合并 #

设置极小时间窗口（1ms），将同一窗口内的lat和lon合并为单点，时间戳统一为窗口结束时间。

示例配置片段：

[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt-broker:1883"]
  topics = [
    "foo/bar/lat",
    "foo/bar/lon"
  ]
  data_format = "value"
  data_type = "float"
  # 添加标签区分lat和lon
  [inputs.mqtt_consumer.tags]
    coord_type = "${topic}"

[[processors.aggregate]]
  period = "1ms"
  delay = "1ms"
  [[processors.aggregate.metrics]]
    namepass = ["mqtt_consumer"]
    fields = ["value"]
    tags = ["coord_type"]
    [[processors.aggregate.metrics.transform]]
      tag = "coord_type"
      dest = "field"
      value = "value"

[[outputs.influxdb_v2]]
  # 输出配置同上

二、在InfluxDB查询层面匹配时间戳 #

若无法修改Telegraf配置，可通过查询对齐lat和lon的时间戳：

方法1：Flux查询 - 匹配最近坐标对 #

lat_data = from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "mqtt_consumer" and r.topic == "foo/bar/lat")
  |> keep(columns: ["_time", "_value"])
  |> rename(columns: {_value: "lat"})

lon_data = from(bucket: "your-bucket")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "mqtt_consumer" and r.topic == "foo/bar/lon")
  |> keep(columns: ["_time", "_value"])
  |> rename(columns: {_value: "lon"})

join(
  tables: {lat: lat_data, lon: lon_data},
  on: (l, r) => abs(duration(v: l._time - r._time)) < 1ms,
  as: (l, r) => ({_time: l._time, lat: l.lat, lon: r.lon})
)

该查询会将时间差小于1ms的lat和lon配对，生成Geomap可识别的有效坐标对。

方法2：InfluxQL查询 - 时间分组聚合 #

SELECT 
  LAST(lat) AS lat, 
  LAST(lon) AS lon 
FROM (
  SELECT value AS lat FROM mqtt_consumer WHERE topic = 'foo/bar/lat'
  UNION ALL
  SELECT value AS lon FROM mqtt_consumer WHERE topic = 'foo/bar/lon'
) 
GROUP BY time(1ms)

通过1ms时间窗口分组，取窗口内lat和lon的最后值合并为坐标对，由于你的事件间隔远大于1ms，不会出现跨事件混淆。

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