Configuring The MQTT Data Collector

Introduction

MQTT is a lightweight, publish-subscribe messaging protocol designed to move data efficiently across networks. Instead of having one system constantly ask devices for updates, MQTT lets devices push information only when something changes. This makes it fast, scalable, and easy to integrate across modern architectures.

You can use the Timebase MQTT Data Collector to log data from an MQTT broker. The MQTT Collector will work with MQTT version 5.0 (versions 3.0 and 3.11 coming soon).

To create and configure your MQTT logging session, follow these steps:

1. Launch your browser and connect to the Collector. The Collector browser config provides a front end editor for the MQTT Collector. By default, this is found at either localhost:4521 (unsecure) or localhost:4522 (secure).
2. Select the Config tab at the top of the browser, just to the right of the Timebase Collector logo.
3. Configure each of the primary sections, Logging Session, Target Historians, and MQTT Configuration.
4. Save your work between each section using the save button in the right hand corner.

Configuration Properties

The Logging Session Properties and Target Historian settings are universal to every Collector type in Timebase and define how the session behaves.

Logging Session Properties

Type: Ensure you have selected MQTT. Selecting Sparkplug will require a different guide for your configuration.

Dataset: The dataset name within the Timebase Historian you wish to target from this logging session. Datasets are automatically created for you if they do not exist. It is recommended you maintain a single logging session to dataset relationship.

Target Historians

You can create multiple connections to target more than one historian. Each historian targeted will receive it's own data stream providing for redundancy.

Identifier: The label used to identify this historian target within the session; does not affect network routing but helps in diagnostics.

Host: The DNS name or IP address of the target historian that receives samples.

Port: The port on which the historian accepts write requests; this must match the historian’s ingest configuration.

Use TLS: Enables encrypted communication between the collector and the historian; recommended for any environment where the collector and historian are not on the same protected network.

Authentication Enabled: Indicates whether the collector should authenticate using the Timebase Pulse Identity Provider before sending data.

IdP Host: The DNS name or IP address of the Pulse Identity Provider (IdP) used to obtain authentication tokens.

IdP Port: The port the IdP exposes for token requests.

IdP Use TLS: Enables TLS for communication with the Identity Provider.

Client ID: The client identity this collector uses when requesting tokens from the IdP.

Client Secret: The credential associated with the Client ID; it allows the collector to authenticate with Pulse.

MQTT Configuration

Connection

Host: The DNS name or IP address of the MQTT broker that publishes Sparkplug messages; the collector connects here to receive data.

Port: The MQTT port used by the broker; typically 1883 for non TLS and 8883 for TLS.

Use TLS: Enables encrypted MQTT communication with the broker; required when the broker enforces TLS for Sparkplug or MQTT connections.

When TLS Cert is Enabled:

Certificate File Path: The full path to the client certificate file required for mutual TLS; used when the broker requires the collector to authenticate with a certificate.

Certificate Password: The password that unlocks the encrypted certificate file.

Certificate Thumbprint: The unique fingerprint of a certificate installed in the system certificate store; used instead of a certificate file path.

Store Location: The certificate store location such as CurrentUser or LocalMachine to search when using a thumbprint.

Store Name: The specific certificate store such as My, Root, or CA where the certificate is located.

Allow Invalid: Allows the collector to proceed with the TLS handshake even if the certificate is expired, self signed, or invalid; suitable only for non production environments.

Authentication

Username: The MQTT username required by brokers that enforce user based authentication or ACLs.

Password: The password associated with the MQTT username.

Subscriptions

Type: Defines how the collector interprets the MQTT message payload, with three modes available depending on how your JSON is structured. Choose between Simple (Primatives), JSON (Simple), and JSON (Structured). 

Simple (Primatives)

When subscribing to a MQTT topic, you may receive messages that consist of only a simple value, such as a number or a string. “Simple (Primitives)”, also referred to as Type 1, are used when the payload is a flat JSON object of primitive values where each key represents a metric, such as:

{ 
"Temp": 72, 
"Speed": 1200,
"Pressure": 101.3 
}

A payload like this is a single-level JSON object containing only primitive types, and the collector will create Timebase metrics named Temperature, Speed, and Pressure directly from the keys without needing to flatten or interpret any structure.

JSON (Simple)

“JSON (simple)” is used when the payload contains nested JSON objects but you only need to extract primitive values, allowing the collector to flatten the structure using the Tagname Delimiter property.

Either examples of the following payloads can be parsed using the JSON (Simple) type.

Sample One, named objects with named children for each motor and no arrays:

{
    "payload": {
        "timestamp": "2024-07-12T12:15"
    },
    "motor1": {
        "rpm": 23.6,
        "location": {
            "long": 12.3,
            "lat": 34.7
        }
    },
    "motor2": {
        "rpm": 28.6,
        "location": {
            "long": 22.4,
            "lat": 36.8
        }
    }
}

Sample Two, root object with a single “data” array, and each array element holds one motor:

{
    "payload": {
        "timestamp": "2024-07-12T12:15"
    },
    "data": [
        {
            "motor1": {
                "rpm": 23.6,
                "location": {
                    "long": 12.3,
                    "lat": 34.7
                }
            }
        },
        {
            "motor2": {
                "rpm": 28.6,
                "location": {
                    "long": 22.4,
                    "lat": 36.8
                }
            }
        }
    ]
}

With your Tagname Delimiter property set to a / either payload will be parsed as follows:

JSON (Structured)

“JSON (Structured)” is used for complex or hierarchical JSON payloads where both structure and field relationships matter, allowing the collector to parse nested objects, arrays, or multi-level element groups while still mapping values into Timebase metrics in a controlled and predictable way.

As an example, look at the payload below. 

• • The 'name' property contains the tag name.
  • The 'value' property contains the tag's value.
  • The 'timestamp' property contains the timestamp.

{
    "payload": {
        "timestamp": 1638223073192,
        "metrics": [
            {
                "name": "Tag 1",
                "timestamp": 1638223073192,
                "dataType": "Int32",
                "metaData": {},
                "properties": {
                    "Quality": {
                        "type": "Int32",
                        "value": 192
                    }
                },
                "value": 1
            },
            {
                "name": "Tag 2",
                "timestamp": 1638223073011,
                "dataType": "Boolean",
                "metaData": {},
                "properties": {
                    "Quality": {
                        "type": "Int32",
                        "value": 192
                    }
                },
                "value": false
            },
            {
                "name": "Tag 3",
                "timestamp": 1638223073041,
                "dataType": "Float",
                "metaData": {},
                "properties": {
                    "Quality": {
                        "type": "Int32",
                        "value": 192
                    }
                },
                "value": 1.23
            }
        ],
        "seq": 1
    }
}

Tagname Field: (JSON Structured only) Specifies one or more JSON field paths whose values will be concatenated (using Tagname Delimiter) in order to form the full metric name, allowing the collector to build a meaningful tag name directly from the payload by joining the returned values instead of relying on the JSON hierarchy or topic structure.

With a Tag Delimiter of / the example would finalize with a tag name that might look like Machine01/TemperatureTransmitter/101

Example: If your JSON payload contains identifying fields such as machine.name, sensor.type, and sensor.id, and you want the tag name to include all of them, you would create three tag name Fields: 

1. machine.name
2. sensor.type
3. sensor.id

With a Tag Delimiter of / the example would finalize with a tag name such as: Machine01/TemperatureTransmitter/101

Tagname Includes Topic: (JSON Simple / JSON Structured only) When enabled, the MQTT topic string is prepended to the generated tag name for this subscription, so topic levels such as site, area, or machine become part of the tag name.

Tagname Delimiter: The character used to join nested JSON fields into a single flattened tag name, so a payload like:

{
   "motor1": {
       "rpm": 23.6,
       "location": {
           "long": 12.3,
           "lat": 34.7
        }
    }
}

becomes motor/location/lat when the delimiter is /, or motor.location.lat when the tag delimiter is set to . thus enabling readable and predictable names for historian metrics.

Timestamp Fields: An ordered list of JSON field paths that may contain timestamp information; when multiple fields are specified, the collector will use the deepest matching field found in the payload as the sample timestamp.

Timestamp Type: Specifies the expected format of the timestamp value in the payload (e.g., UNIX seconds, UNIX milliseconds, or a date/time string) so the collector can correctly convert it to a Timebase timestamp.

Value Field: (JSON (Structured) only) Defines the JSON field path that contains the numeric or value data for each metric in a structured payload; required for Json (Structured) subscriptions to correctly map the metric value.

Quality Field: (JSON (Structured) only) Defines the JSON field path that contains the quality or status information for each metric in a structured payload; optionally used to record metric quality alongside values.

UOM Field: An array of JSON paths used to retrieve a tag’s unit of measure. Each path is tried in order; the last one found is used. If multiple return they may be concatenated.

Quality Of Service: Defines the MQTT delivery guarantee for subscribed messages. 

Topics: Defines one or more MQTT topic filters, each entered on its own line, to determine which published messages the collector will process, with support for MQTT wildcards (+ for a single topic level and # for all remaining levels) so you can target individual devices or subscribe broadly without manually listing every topic.

More About Topic Structure and Wildcards

MQTT does not define or enforce any payload structure, so devices and applications are free to publish any JSON shape they want: flat objects, deeply nested objects, arrays, mixed schemas, or even non-JSON formats such as CSV or binary. Because MQTT does not prescribe where values, timestamps, identifiers, or units should live inside the payload, Timebase relies on topic filters and subscription settings to determine which messages to listen to and how to interpret them, which is why MQTT wildcards are essential for scaling: they allow you to subscribe to entire namespaces of interest regardless of how many devices exist or how many topics they publish.

Wildcards help you subscribe to exactly the data you want without having to list every device or node manually. Wildcards let you capture entire groups of devices, simplify onboarding of new equipment, and reduce the need to constantly update topic lists.

The + Wildcard: The + wildcard matches exactly one level in a topic path. If a segment may vary but stays within a single level, use +.

Example: Plant1/+/Filler01 matches all nodes in Plant1 that contain a device named Filler01.

Example: ACME Juice Co/+/Node01/ matches Node01 in all groups held within the ACME Juice Co topic.

The # Wildcard: The # wildcard matches zero or more levels and must always be at the end of the topic pattern. Use # when you want to subscribe to everything under a certain point in the hierarchy.

Example: Plant1/LineA/# captures every device, metric, and event published from LineA.

Example: ACME Juice Co/+/+/# captures all traffic across all Groups and Nodes within the ACME Juice Co topic, which is helpful for initial discovery or troubleshooting

Topic Definition: A slash-delimited list of semantic keys that tells the MQTT Collector how to interpret each segment of the MQTT topic so those values can be used as dynamic fields on the generated tags.

How Topic Definition Works

MQTT topics often contain meaningful context such as enterprise, site, line, equipment, or subsystem, but the MQTT Collector cannot know which segment represents what unless it is defined.

Topic Definition allows you to map each topic segment to a specific semantic role by listing the keys in order, and the MQTT Collector then extracts those values from every incoming topic to populate fields on the metric.

Example: With a Topic Definition defined as enterprise/site/area/machine and an MQTT topic of ACME Juice Co/Austin/Packaging/Capper1/Temp, the collector resolves

• enterprise = ACME Juice Co
• site = Austin
• area = Packaging
• machine = Capper1

Segments beyond those defined are ignored, and the resolved values can be added to tag names or stored as metadata using the Fields configuration.

This approach lets Timebase derive rich contextual information directly from the topic structure without requiring that data to be repeated inside the payload.

Filter: A .NET regular expression (regex) used to omit metrics from logging; the collector evaluates the regex against each tag name and excludes any tag that matches the pattern, allowing broad subscriptions while preventing unwanted tags from being written to the historian.

How .NET Regex Filtering Works

The .NET regex engine matches your pattern against the full metric name provided in each Sparkplug message; if the name matches the pattern, the collector does not log that metric, which helps keep the historian focused on meaningful operational data.

Simple Contains Match

To exclude any metric containing the word Debug, simply enter Debug into the filter.

Prefix Based Filtering

To exclude all metrics beginning with a specific device prefix, use: ^Filler02_.* which removes Filler02_Speed, Filler02_Temp, and all other metrics starting with Filler02_.

Excluding Classes of Metrics

If alarm metrics follow a known prefix such as Alarm_, you can exclude them using: ^Alarm_.* which keeps the historian from filling with high-frequency alarm chatter.

Matching Multiple Patterns

To exclude metrics belonging to either of two prefixes, use: ^(Filler02_|Capper01_).* which uses .NET grouping and alternation.

Using Negative Lookahead

To keep all metrics except those containing the word Test, use: ^(?!.*Test).* which is a .NET negative lookahead pattern and excludes any name containing Test while allowing everything else.

For a online tutorial in regex or to learn more, we recommend the website RegExr - https://regexr.com/

Fields: Metadata key/value pairs applied to all tags in this subscription. Defining these key/value pairs will store custom metadata fields and their corresponding values to tags within the dataset.

This feature is helpful for mining or extracting metadata directly from your source systems where available. Within Timebase, and for applications consuming data from Timebase, Fields are useful to classify, group, and filter tags based on attributes like enterprise, site, area, machine number, process step, or many other useful associations.

Varying Types of Fields

Static Fields

Fields can be defined a static text:

"Fields": {
   "Enterprise": "ACME",
   "Site": "Austin",
   "Area": "Packaging"
 }

Topic Fields

Together with the Topic Definition property (configuration above), fields can be defined to extract values from the topic structure:

"Fields": {
  "Enterprise": "[Topic:enterprise]",
  "Site": "[Topic:site]",
  "Area": "[Topic:area]",
  "Machine": "[Topic:Machine]"
 }

Payload Fields

Fields can be defined to extract values from the payload:

 "Fields": {
   "Device": "[Payload:topic.deviceId]"
 }

"Fields": {
  "Id": "Id-[Topic:area][Topic:machine][Payload:topic.deviceId]"
 }