Skip to main content

SparkplugB

This section describes how to create a data ingestion task in taosExplorer that reads SparkplugB data into the current TDengine cluster.

Overview​

SparkplugB is an open messaging specification designed for Industrial Internet of Things (IIoT) applications and built on MQTT.

TDengine can subscribe to an MQTT broker through the SparkplugB connector and write the data into TDengine in real time.

Create a Task​

1. Add a Data Source​

On the Data In page, click Add Data Source.

2. Configure Basic Information​

Enter a task name, such as test_spb, and select SparkplugB as the type.

Optionally select an agent, or click Create New Agent. Select the target database, or click Create Database.

3. Configure the Connection and Authentication​

In Brokers, enter the MQTT broker address, such as localhost:1883. Separate multiple brokers with commas.

Select the MQTT protocol version. The default is 5.0.

Enter the client identifier used to connect to each broker and configure the keep-alive interval. If the broker receives no message from the client during this interval, it closes the connection.

Enter the MQTT username and password if required.

For TLS Verification, select one of the following modes:

  1. Disabled: Do not verify TLS certificates. The connector first attempts a TCP connection and, if that fails, attempts TLS without certificate verification.
  2. One-way authentication: Use TLS and verify the server certificate. Upload the CA certificate.
  3. Mutual authentication: Use mutual TLS. Upload the CA certificate, client certificate, and client private key.

Click Check Connectivity.

4. Configure the Subscription​

In Group ID, enter the SparkplugB group ID, which typically represents an organization, factory, or production line.

In Node/Device List, enter comma-separated nodes and devices. Specify a node by its ID and a device as {node-id}/{device-id}.

In Message Types, enter comma-separated SparkplugB message types. Supported values are NBIRTH, NDEATH, NDATA, NCMD, DBIRTH, DDEATH, DDATA, DCMD, and STATE. Node message types match nodes in Node/Device List, while device message types match devices.

When Send REBIRTH Command is enabled, taosX sends the Node Control/Rebirth NCMD command to retrieve all node and device metric metadata, including the mapping between metric names and aliases. You can leave this disabled when publishers do not use metric aliases.

5. Configure Payload Transformation​

5.1 Parse the Payload​

You can obtain sample data by retrieving it from the server, uploading a file, or entering the message body manually. SparkplugB messages use Protocol Buffers, so data retrieved from the server is decoded to JSON. The JSON parser can process fields such as SparkplugB metadata and properties.

Click the magnifying-glass icon to preview the parsed result.

5.2 Extract or Split Fields​

For example, to convert the value of datatype_str to a TDengine data type, select the mapping extractor, enter the following JSON in Rule, and enter td_datatype in Name:

{
  "Int8": "TINYINT",
  "UInt8": "TINYINT UNSIGNED",
  "Int16": "SMALLINT",
  "UInt16": "SMALLINT UNSIGNED",
  "Int32": "INT",
  "UInt32": "INT UNSIGNED",
  "Int64": "BIGINT",
  "UInt64": "BIGINT UNSIGNED",
  "Float": "FLOAT",
  "DOUBLE": "DOUBLE",
  "Boolean": "BOOL",
  "String": "VARCHAR(128)",
  "DateTime": "TIMESTAMP"
}

For example, this converts Int8 in datatype_str to TINYINT in the new td_datatype column.

You can add or delete extraction rules and preview their results.

5.3 Filter Data​

Enter a filter expression. For example, datatype_str != "Int8" writes only rows whose datatype_str value is not Int8.

You can delete the rule or preview its result.

5.4 Map Tables​

Select a target supertable or click Create Supertable.

If the supertable must be generated dynamically from each message, select Create Template. The supertable name, column names, and column types can contain template variables. When data arrives, taosX evaluates the variables, creates a missing supertable, and adds missing columns to an existing supertable.

Configure the target subtable name, such as t_{id}, and map source fields to columns and tags. Mapping rules support default values.

Click Preview to inspect the mapping result.

6. Configure Advanced Options​

The Advanced Options section is collapsed by default. Click > to expand it. MQTT and SparkplugB data sources commonly provide the following options; field names can vary by connector.

Message Queue Size specifies the receive buffer size. If the queue is full and Cache Realtime Data is disabled, newly arrived data is discarded. Set this option to 0 to disable buffering.

Maximum In-Process Batches specifies how many batches can be processed concurrently. Once this limit is reached, the connector stops taking messages from the receive queue, causing them to accumulate. The minimum value is 1.

Batch Size specifies how many messages are sent to the processing pipeline at a time. It works with Batch Delay: a full batch is sent immediately even if the delay has not elapsed. The minimum value is 1.

Batch Delay specifies the timeout for each batch in milliseconds, measured from its first message. When the timeout expires, the batch is sent even if it has not reached Batch Size. The minimum value is 1.

Write Concurrency specifies how many tasks can write to TDengine concurrently.

When Cache Realtime Data is enabled, consumed data is first written to a local file and a background task forwards it downstream. This provides traffic shaping when downstream processing cannot keep up. Cache files are removed after the backlog is consumed. This option is disabled by default. For details, see Store and Forward.

Cache Storage Directory overrides the directory used for cache files. It applies only when Cache Realtime Data is enabled and otherwise defaults to the data directory configured at taosX startup.

When Save Raw Data is enabled, you can also configure Maximum Retention Days and Raw Data Storage Directory.

For health-monitoring options, see Health Status.

7. Configure Exception Handling​

The Exception Handling Strategy section is collapsed by default. Click > to expand it.

The common strategies are:

  • Archive: Write the invalid data to an archive file, by default under ${data_dir}/tasks/<id>/<datetime>, without writing it to the target database.
  • Discard: Ignore the invalid data.
  • Error: Report an error.
  • Cache: For a target connection failure or resource shortage, write data to a cache file and ingest it after the target recovers.

You can configure strategies for these conditions:

  • Target connection timeout: Archive, discard, error, or cache.
  • Target database does not exist: Archive, discard, or error.
  • Table does not exist: Archive, discard, error, or create the table automatically and retry.
  • Primary timestamp out of range (now - keep1 through now + 100y): Archive, discard, or error.
  • Primary timestamp is null: Archive, discard, error, or use the current time.
  • Composite primary key is null: Archive, discard, or error.
  • Table name exceeds 192 characters: Archive, discard, error, truncate, or truncate and archive.
  • Invalid table-name characters, such as .: Archive, discard, error, or replace invalid characters with a configured string.
  • Null table-name template variable: Discard, leave the variable empty, or replace it with a configured string.
  • Column does not exist: Archive, discard, error, or add the missing column automatically and retry.
  • Column name exceeds 64 characters: Archive, discard, or error.
  • Column value exceeds its defined length: Archive, discard, error, truncate, or truncate and archive. Automatic Column Expansion can alter the table and retry instead.
  • Other data errors: Archive, discard, or error.

Additional settings include:

  • Connection Timeout: Target connection timeout in seconds, from 1 through 600.
  • Temporary Storage Location: A path relative to ${data_dir}/tasks/<id>/.
  • Archive Retention Days: A nonnegative integer; 0 means unlimited.
  • Archive Available Space: From 0 through 65535; 0 means unlimited.
  • Archive Location: A path relative to ${data_dir}/tasks/<id>/.
  • Archive Write Failure Strategy: Delete old files, discard the data, or report an error and stop the task.

8. Complete the Task​

Click Submit. The task status is displayed on the Data Source List page.