Telegraf json_v2 Parser and MQTT Topic Parsing Explained

The Telegraf json_v2 parser lets you decide exactly which parts of an MQTT payload become measurements, tags and fields, while topic parsing extracts identifiers from the topic itself. In this tutorial you will configure both on Telegraf's mqtt_consumer input, starting from a realistic nested payload, and verify the output line by line before anything reaches InfluxDB.

Why the Legacy JSON Parser Is Not Enough

Telegraf's original json parser flattens everything it finds into fields, joins nested keys with underscores, ignores strings unless you list them, and offers limited control over types. That is acceptable for flat payloads like {"temperature": 21.5}. Industrial and IoT payloads are rarely that simple. A machine gateway typically publishes something like this:

{
  "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": []
}

You want firmware as a tag, status as a string field, the three metrics as properly typed fields, the serial number ignored and the device timestamp used as the point time. The json_v2 parser expresses each of those decisions explicitly using GJSON path syntax, so nothing ends up in the database by accident.

The Example Setup

Messages arrive on an ISA-95 style topic:

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

That is seven levels, within the public broker's limit of eight. The topic already carries site, line and machine, so the configuration will take those from the topic rather than relying on the payload.

Publish a test message to broker.connectmq.com with your own random ID in place of 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\":[]}"

Telegraf json_v2 Parser Configuration

Save this as 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"

  # Site, line and machine come from the 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"

Run it in test mode. mqtt_consumer is a service input, so give it time to receive a message, and publish the test message while it waits:

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

The output is a single line of 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

How the Mapping Works

Paths use GJSON syntax

Every path is a GJSON expression evaluated against the payload. A dot walks into nested objects, so metrics.temp_c reads the temperature inside metrics. The # modifier returns the length of an array, which is why alarms.# produces an alarm count without storing the array itself. GJSON also supports array indexes such as readings.0.value and queries such as readings.#(sensor=="t1").value when you need to pick one element.

Tags, fields and types

Declaring types explicitly prevents a classic InfluxDB problem: a field written as an integer once and as a float later is rejected as a type conflict. Choose a type per field and keep it forever.

Without rename, a nested path becomes a field name with the path flattened, which tends to produce long, awkward names. Renaming keeps the schema readable in queries and dashboards.

Topic parsing fills in the hierarchy

The topic_parsing block splits the topic on / and lines it up with the pattern in tags. Each _ skips a level and each name creates a tag from that position. The pattern must have exactly as many segments as the topic, and the topic filter inside the block decides which messages the rule applies to, with + and # wildcards allowed.

Topic parsing can also set the measurement and create fields. Here is the same rule taking the measurement from the last topic level instead of the fixed measurement_name:

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

With that rule, messages on .../cnc1/telemetry land in a telemetry measurement and messages on .../cnc1/energy in an energy measurement. Topic parsing runs after the payload parser, so the measurement from the topic replaces the one from measurement_name. For numeric values carried in the topic itself, use fields together with a [inputs.mqtt_consumer.topic_parsing.types] table to set their types, but this is rarely a good topic design.

Parsing Arrays into Multiple Metrics

Gateways often batch several readings in one message:

{
  "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 }
  ]
}

An object block turns each array element into its own metric. Put it in a separate mqtt_consumer input that listens on the batch topic, so it never runs against payloads without a readings array:

[[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"

Each element produces a line such as reading,gateway=gw-07,line=line1,sensor=t1,site=factory1 value=21.3 1760600000100000000. The tags list promotes keys to tags, timestamp_key reads the per-element time and the fields table fixes types. Note that the gateway tag comes from the topic, which keeps the object block focused on the array itself.

Troubleshooting Checklist

1. No output in test mode. Confirm the topic with mosquitto_sub -h broker.connectmq.com -p 1883 -i dbg-a1b2c3 -t "connectmq/demo/a1b2c3/#" -v and make sure you publish during the --test-wait window.
2. Field missing from output. Check the GJSON path against the actual payload; a typo in a path silently yields nothing. Newer Telegraf versions accept optional = true on fields that are not always present.
3. Error about topic parsing segments. Count the levels in the topic filter and in the tags pattern; they must match.
4. Timestamps in 1970. The payload uses seconds but the config says unix_ms, or the reverse.
5. Type conflict errors in InfluxDB. An earlier configuration wrote the same field with another type. Use a new field name or a new bucket when changing types.
6. Every value is a string. A field type was set to string or the device sends numbers in quotes, such as "41.7". Setting type = "float" converts quoted numbers.
7. Too many series. A high-cardinality value such as a serial number or message ID was mapped to a tag. Make it a field or drop it.

Running It in Production

Once the mapping is right, remove the outputs.file block, add an [[outputs.influxdb_v2]] output, and switch the input to a private broker. With a fully managed MQTT service like ConnectMQ, that means servers = ["ssl://broker.connectmq.com:8883"], username, password and the client ID issued in the portal. Before writing parsing rules for a new device type, the portal's live topic explorer is a convenient way to copy a real payload. ConnectMQ also decodes Sparkplug B and can republish it into ISA-95 topics, so Telegraf can subscribe to clean ISA-95 topics; inspect a republished payload in the portal before writing paths for it.

For the basics of the input plugin, start with the Telegraf MQTT consumer tutorial. To run the full pipeline with Grafana, use the TIG stack Docker Compose setup, and to choose device-side libraries, see the MQTT client libraries guide. The official json_v2 parser reference lists every option.

Get Started with ConnectMQ

Test your parsing rules against the free public MQTT broker with sample payloads, then move to private credentials for real machines. ConnectMQ runs a managed MQTT 5 broker with TLS and a portal that shows live topics and payloads. Create a free Serverless broker at app.connectmq.com with no credit card required. For a wider view of hosting choices, see the cloud MQTT broker guide.

Frequently Asked Questions

What is the difference between the json and json_v2 parsers in Telegraf?

The legacy json parser flattens payloads automatically with limited control. The json_v2 parser uses explicit GJSON paths for tags, fields, types and timestamps, which suits nested IoT payloads much better.

Can Telegraf take tags from the MQTT topic?

Yes. A topic_parsing block maps topic levels to tags, fields or the measurement name. Use _ for levels you want to skip.

How do I parse a JSON array into separate metrics?

Use an object block with the array's path. Each element becomes its own metric, and you can promote keys to tags and set a per-element timestamp.

Do I need json_v2 for flat payloads?

No. For flat numeric payloads the legacy parser works fine. Switch to json_v2 when you need strings, explicit types, renamed fields or nested structures.