Telegraf-json_v2-Parser und MQTT-Topic-Parsing erklärt

Mit dem Telegraf json_v2 Parser bestimmen Sie genau, welche Teile einer MQTT-Payload zu Measurements, Tags und Fields werden, während das Topic-Parsing Bezeichner aus dem Topic selbst extrahiert. In diesem Tutorial konfigurieren Sie beides am mqtt_consumer-Input von Telegraf, ausgehend von einer realistischen verschachtelten Payload, und prüfen die Ausgabe Zeile für Zeile, bevor irgendetwas InfluxDB erreicht.

Warum der alte JSON-Parser nicht ausreicht

Der ursprüngliche json-Parser von Telegraf glättet alles, was er findet, zu Fields, verbindet verschachtelte Schlüssel mit Unterstrichen, ignoriert Strings, sofern Sie sie nicht auflisten, und bietet nur begrenzte Kontrolle über Typen. Für flache Payloads wie {"temperature": 21.5} ist das akzeptabel. Industrie- und IoT-Payloads sind selten so einfach. Ein Maschinen-Gateway veröffentlicht typischerweise etwas wie das hier:

{
  "ts": 1760600000123,
  "device": { "id": "cnc1", "fw": "2.4.1", "serial": "SN-88231" },
  "status": "running",
  "metrics": { "spindle_rpm": 11800, "temp_c": 41.7, "vibration_mm_s": 2.3 },
  "alarms": []
}

Sie möchten die Firmware als Tag, den Status als String-Field, die drei Metriken als korrekt typisierte Fields, die Seriennummer ignoriert und den Gerätezeitstempel als Zeitpunkt des Punkts. Der json_v2-Parser drückt jede dieser Entscheidungen explizit mit GJSON-Pfadsyntax aus, sodass nichts versehentlich in der Datenbank landet.

Das Beispiel-Setup

Nachrichten treffen auf einem Topic im ISA-95-Stil ein:

connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry

Das sind sieben Ebenen, innerhalb der Grenze des öffentlichen MQTT-Brokers von acht. Das Topic enthält bereits Standort, Linie und Maschine, die Konfiguration übernimmt diese daher aus dem Topic, statt sich auf die Payload zu verlassen.

Dieses Muster stammt aus dem Unified Namespace (UNS): Die ISA-95-Equipment-Hierarchie (Enterprise → Site → Area → Line → Cell) wird zur Topic-Struktur auf dem MQTT-Broker, im Werk etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Ist der UNS konsequent aufgebaut, liefert topic_parsing die Anlagenhierarchie als Tags gratis mit, und Grafana-Dashboards lassen sich nach Werk, Linie oder Maschine filtern. Den Hintergrund erklärt Sparkplug B und Unified Namespace.

Veröffentlichen Sie eine Testnachricht an broker.connectmq.com, mit Ihrer eigenen zufälligen ID anstelle von a1b2c3:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry \
  -m "{\"ts\":$(date +%s%3N),\"device\":{\"id\":\"cnc1\",\"fw\":\"2.4.1\",\"serial\":\"SN-88231\"},\"status\":\"running\",\"metrics\":{\"spindle_rpm\":11800,\"temp_c\":41.7,\"vibration_mm_s\":2.3},\"alarms\":[]}"

Konfiguration des Telegraf json_v2 Parsers

Speichern Sie dies als json_v2.conf:

[agent]
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/telemetry"]
  client_id = "telegraf-json-a1b2c3"
  qos = 0
  topic_tag = ""
  data_format = "json_v2"

  # Standort, Linie und Maschine kommen aus dem Topic
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/telemetry"
    tags = "_/_/_/site/line/asset/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "machine"
    timestamp_path = "ts"
    timestamp_format = "unix_ms"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "device.fw"
      rename = "firmware"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "status"
      type = "string"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.spindle_rpm"
      rename = "spindle_rpm"
      type = "int"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.temp_c"
      rename = "temp_c"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.vibration_mm_s"
      rename = "vibration_mm_s"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "alarms.#"
      rename = "alarm_count"
      type = "int"

[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Führen Sie es im Testmodus aus. mqtt_consumer ist ein Service-Input, geben Sie ihm also Zeit, eine Nachricht zu empfangen, und veröffentlichen Sie die Testnachricht, während er wartet:

telegraf --config json_v2.conf --test --test-wait 20

Die Ausgabe ist eine einzelne Zeile Line Protocol:

machine,asset=cnc1,firmware=2.4.1,line=line1,site=factory1 status="running",spindle_rpm=11800i,temp_c=41.7,vibration_mm_s=2.3,alarm_count=0i 1760600000123000000

So funktioniert das Mapping

Pfade verwenden GJSON-Syntax

Jeder path ist ein GJSON-Ausdruck, der gegen die Payload ausgewertet wird. Ein Punkt geht in verschachtelte Objekte hinein, sodass metrics.temp_c die Temperatur innerhalb von metrics liest. Der Modifikator # liefert die Länge eines Arrays, weshalb alarms.# eine Alarmanzahl erzeugt, ohne das Array selbst zu speichern. GJSON unterstützt außerdem Array-Indizes wie readings.0.value und Abfragen wie readings.#(sensor=="t1").value, wenn Sie ein einzelnes Element auswählen müssen.

Tags, Fields und Typen

Explizit deklarierte Typen verhindern ein klassisches InfluxDB-Problem: Ein Field, das einmal als Integer und später als Float geschrieben wird, wird als Typkonflikt abgelehnt. Wählen Sie pro Field einen Typ und behalten Sie ihn für immer.

Ohne rename wird ein verschachtelter Pfad zu einem Field-Namen mit geglättetem Pfad, was meist lange, sperrige Namen ergibt. Umbenennen hält das Schema in Abfragen und Dashboards lesbar.

Topic-Parsing füllt die Hierarchie

Der Block topic_parsing teilt das Topic an / auf und legt es neben das Muster in tags. Jedes _ überspringt eine Ebene, und jeder Name erzeugt an dieser Position ein Tag. Das Muster muss genau so viele Segmente haben wie das Topic, und der topic-Filter innerhalb des Blocks entscheidet, auf welche Nachrichten die Regel angewendet wird, wobei die Wildcards + und # erlaubt sind.

Topic-Parsing kann auch das Measurement setzen und Fields erzeugen. Hier dieselbe Regel, die das Measurement aus der letzten Topic-Ebene statt aus dem festen measurement_name übernimmt:

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/+"
    measurement = "_/_/_/_/_/_/measurement"
    tags = "_/_/_/site/line/asset/_"

Mit dieser Regel landen Nachrichten auf .../cnc1/telemetry in einem Measurement telemetry und Nachrichten auf .../cnc1/energy in einem Measurement energy. Topic-Parsing läuft nach dem Payload-Parser, sodass das Measurement aus dem Topic dasjenige aus measurement_name ersetzt. Für numerische Werte, die im Topic selbst stecken, verwenden Sie fields zusammen mit einer Tabelle [inputs.mqtt_consumer.topic_parsing.types], um ihre Typen festzulegen; das ist aber selten ein gutes Topic-Design.

Arrays in mehrere Metriken parsen

Gateways bündeln oft mehrere Messwerte in einer Nachricht:

{
  "gateway": "gw-07",
  "readings": [
    { "sensor": "t1", "ts": 1760600000100, "value": 21.3 },
    { "sensor": "t2", "ts": 1760600000105, "value": 22.0 },
    { "sensor": "h1", "ts": 1760600000110, "value": 48.5 }
  ]
}

Ein object-Block macht aus jedem Array-Element eine eigene Metrik. Legen Sie ihn in einen separaten mqtt_consumer-Input, der auf dem Batch-Topic lauscht, damit er nie gegen Payloads ohne readings-Array läuft:

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/batch"]
  client_id = "telegraf-batch-a1b2c3"
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/batch"
    tags = "_/_/_/site/line/gateway/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "reading"

    [[inputs.mqtt_consumer.json_v2.object]]
      path = "readings"
      tags = ["sensor"]
      timestamp_key = "ts"
      timestamp_format = "unix_ms"
      [inputs.mqtt_consumer.json_v2.object.fields]
        value = "float"

Jedes Element erzeugt eine Zeile wie reading,gateway=gw-07,line=line1,sensor=t1,site=factory1 value=21.3 1760600000100000000. Die tags-Liste macht Schlüssel zu Tags, timestamp_key liest die Zeit pro Element, und die fields-Tabelle legt die Typen fest. Beachten Sie, dass das Gateway-Tag aus dem Topic stammt, wodurch sich der Object-Block auf das Array selbst konzentriert.

Checkliste zur Fehlerbehebung

1. Keine Ausgabe im Testmodus. Prüfen Sie das Topic mit mosquitto_sub -h broker.connectmq.com -p 1883 -i dbg-a1b2c3 -t "connectmq/demo/a1b2c3/#" -v und stellen Sie sicher, dass Sie innerhalb des --test-wait-Fensters veröffentlichen.
2. Field fehlt in der Ausgabe. Prüfen Sie den GJSON-Pfad gegen die tatsächliche Payload; ein Tippfehler in einem Pfad liefert stillschweigend nichts. Neuere Telegraf-Versionen akzeptieren optional = true für Fields, die nicht immer vorhanden sind.
3. Fehler zu Topic-Parsing-Segmenten. Zählen Sie die Ebenen im topic-Filter und im tags-Muster; sie müssen übereinstimmen.
4. Zeitstempel in 1970. Die Payload verwendet Sekunden, die Konfiguration aber unix_ms, oder umgekehrt.
5. Typkonflikt-Fehler in InfluxDB. Eine frühere Konfiguration hat dasselbe Field mit einem anderen Typ geschrieben. Verwenden Sie beim Ändern von Typen einen neuen Field-Namen oder einen neuen Bucket.
6. Jeder Wert ist ein String. Ein Field-type wurde auf string gesetzt, oder das Gerät sendet Zahlen in Anführungszeichen, etwa "41.7". type = "float" wandelt Zahlen in Anführungszeichen um.
7. Zu viele Serien. Ein Wert mit hoher Kardinalität wie eine Seriennummer oder Nachrichten-ID wurde auf ein Tag abgebildet. Machen Sie ihn zu einem Field oder verwerfen Sie ihn.

Betrieb in der Produktion

Sobald das Mapping stimmt, entfernen Sie den outputs.file-Block, ergänzen einen [[outputs.influxdb_v2]]-Output und stellen den Input auf einen privaten Broker um. Bei einem verwalteten MQTT-Broker wie ConnectMQ bedeutet das servers = ["ssl://broker.connectmq.com:8883"], username, password und die im Portal ausgestellte Client-ID. Bevor Sie Parsing-Regeln für einen neuen Gerätetyp schreiben, ist der Live-Topic-Explorer im Portal ein bequemer Weg, eine echte Payload zu kopieren. ConnectMQ dekodiert außerdem Sparkplug B und kann es in ISA-95-Topics neu veröffentlichen, sodass Telegraf saubere ISA-95-Topics abonnieren kann; prüfen Sie eine neu veröffentlichte Payload im Portal, bevor Sie Pfade dafür schreiben.

Für die Grundlagen des Input-Plugins beginnen Sie mit dem Tutorial zum Telegraf MQTT Consumer. Um die gesamte Pipeline mit Grafana zu betreiben, nutzen Sie das TIG-Stack-Setup mit Docker Compose, und zur Auswahl gerätseitiger Bibliotheken lesen Sie den Leitfaden zu MQTT-Client-Bibliotheken. Die offizielle Referenz zum json_v2-Parser listet jede Option auf.

Starten Sie mit ConnectMQ

Testen Sie Ihre Parsing-Regeln mit Beispiel-Payloads gegen den kostenlosen öffentlichen MQTT-Broker, der für Tests gedacht und für jeden lesbar ist, und wechseln Sie für echte Maschinendaten zu privaten Zugangsdaten. ConnectMQ betreibt einen verwalteten MQTT-5-Broker mit TLS und einem Portal, das Live-Topics und Payloads zeigt. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ohne Kreditkarte. Einen breiteren Blick auf Hosting-Optionen bietet der Leitfaden zum Cloud-MQTT-Broker.

Häufig gestellte Fragen

Was ist der Unterschied zwischen dem json- und dem json_v2-Parser in Telegraf?

Der alte json-Parser glättet Payloads automatisch, mit begrenzter Kontrolle. Der json_v2-Parser verwendet explizite GJSON-Pfade für Tags, Fields, Typen und Zeitstempel, was zu verschachtelten IoT-Payloads deutlich besser passt.

Kann Telegraf Tags aus dem MQTT-Topic übernehmen?

Ja. Ein topic_parsing-Block bildet Topic-Ebenen auf Tags, Fields oder den Measurement-Namen ab. Verwenden Sie _ für Ebenen, die Sie überspringen möchten.

Wie parse ich ein JSON-Array in separate Metriken?

Verwenden Sie einen object-Block mit dem Pfad des Arrays. Jedes Element wird zu einer eigenen Metrik, und Sie können Schlüssel zu Tags machen und einen Zeitstempel pro Element festlegen.

Brauche ich json_v2 für flache Payloads?

Nein. Für flache numerische Payloads funktioniert der alte Parser einwandfrei. Wechseln Sie zu json_v2, wenn Sie Strings, explizite Typen, umbenannte Fields oder verschachtelte Strukturen brauchen.