Sparkplug B

Overview

Sparkplug B is an industrial IoT data transfer specification built on MQTT 3.1.1. Sparkplug B provides a unified way for device manufacturers and software providers to share data by making MQTT networks state-aware and interoperable while ensuring flexibility and efficiency.

Data collected by Neuron from devices can be transferred from the edge to the Sparkplug B application via the Sparkplug B protocol, and users can send data modification commands to Neuron from the application. sparkplug B is an application-based protocol running on top of MQTT, so the setup in Neuron is similar to the MQTT driver.

Example

Here, the actual point data of the device is collected through the Neuron southbound, and the data is reported to EMQX through the northbound Sparkplug B plug-in, and the correct and complete data results are obtained after decoding through the codec function. The process is shown in the figure:

Sparkplug B

Neuron

Southbound device

Collect the Modbus TCP simulator point value through the southbound drive to simulate the actual device point value, the configuration is as follows：

Add device

image-20230421141232191

Device Configuration

image-20230421141356669

image-20230421141441471

Create group

image-20230421141538315

Create Tag

image-20230421141639201

Northbound application

Add application

image-20230421141821812

Application configuration

image-20230421141912355

image-20230421142020855

add subscription

image-20230421142109283

EMQX

If you directly subscribe to the EMQX data reported by Neuron northbound to SparkPlugB, there will be garbled strings, as shown in the figure:

image-20230421151918685

Therefore, through the encoding and decoding capabilities of the EMQX rule engine, a corresponding proto file can be written and combined with the rule engine to decode the reported data and obtain correct and complete data results.

Create Schema Registry

image-20230421142801332

image-20230421142920299

  1. // Complete proto file
  2. syntax = "proto2";
  4. //
  5. // To compile:
  6. // cd client_libraries/java
  7. // protoc --proto_path=../../ --java_out=src/main/java ../../sparkplug_b.proto 
  8. //
  10. message Payload {
  11.     /*
  12.         // Indexes of Data Types
  13.         // Unknown placeholder for future expansion.
  14.         Unknown         = 0;
  15.         // Basic Types
  16.         Int8            = 1;
  17.         Int16           = 2;
  18.         Int32           = 3;
  19.         Int64           = 4;
  20.         UInt8           = 5;
  21.         UInt16          = 6;
  22.         UInt32          = 7;
  23.         UInt64          = 8;
  24.         Float           = 9;
  25.         Double          = 10;
  26.         Boolean         = 11;
  27.         String          = 12;
  28.         DateTime        = 13;
  29.         Text            = 14;
  30.         // Additional Metric Types
  31.         UUID            = 15;
  32.         DataSet         = 16;
  33.         Bytes           = 17;
  34.         File            = 18;
  35.         Template        = 19;
  37.         // Additional PropertyValue Types
  38.         PropertySet     = 20;
  39.         PropertySetList = 21;
  40.     */
  42.     message Template {
  44.         message Parameter {
  45.             optional string name        = 1;
  46.             optional uint32 type        = 2;
  48.             oneof value {
  49.                 uint32 int_value        = 3;
  50.                 uint64 long_value       = 4;
  51.                 float  float_value      = 5;
  52.                 double double_value     = 6;
  53.                 bool   boolean_value    = 7;
  54.                 string string_value     = 8;
  55.                 ParameterValueExtension extension_value = 9;
  56.             }
  58.             message ParameterValueExtension {
  59.                 extensions              1 to max;
  60.             }
  61.         }
  63.         optional string version         = 1;          // The version of the Template to prevent mismatches
  64.         repeated Metric metrics         = 2;          // Each metric is the name of the metric and the datatype of the member but does not contain a value
  65.         repeated Parameter parameters   = 3;
  66.         optional string template_ref    = 4;          // Reference to a template if this is extending a Template or an instance - must exist if an instance
  67.         optional bool is_definition     = 5;
  68.         extensions                      6 to max;
  69.     }
  71.     message DataSet {
  73.         message DataSetValue {
  75.             oneof value {
  76.                 uint32 int_value                        = 1;
  77.                 uint64 long_value                       = 2;
  78.                 float  float_value                      = 3;
  79.                 double double_value                     = 4;
  80.                 bool   boolean_value                    = 5;
  81.                 string string_value                     = 6;
  82.                 DataSetValueExtension extension_value   = 7;
  83.             }
  85.             message DataSetValueExtension {
  86.                 extensions  1 to max;
  87.             }
  88.         }
  90.         message Row {
  91.             repeated DataSetValue elements  = 1;
  92.             extensions                      2 to max;   // For third party extensions
  93.         }
  95.         optional uint64   num_of_columns    = 1;
  96.         repeated string   columns           = 2;
  97.         repeated uint32   types             = 3;
  98.         repeated Row      rows              = 4;
  99.         extensions                          5 to max;   // For third party extensions
  100.     }
  102.     message PropertyValue {
  104.         optional uint32     type                    = 1;
  105.         optional bool       is_null                 = 2; 
  107.         oneof value {
  108.             uint32          int_value               = 3;
  109.             uint64          long_value              = 4;
  110.             float           float_value             = 5;
  111.             double          double_value            = 6;
  112.             bool            boolean_value           = 7;
  113.             string          string_value            = 8;
  114.             PropertySet     propertyset_value       = 9;
  115.             PropertySetList propertysets_value      = 10;      // List of Property Values
  116.             PropertyValueExtension extension_value  = 11;
  117.         }
  119.         message PropertyValueExtension {
  120.             extensions                             1 to max;
  121.         }
  122.     }
  124.     message PropertySet {
  125.         repeated string        keys     = 1;         // Names of the properties
  126.         repeated PropertyValue values   = 2;
  127.         extensions                      3 to max;
  128.     }
  130.     message PropertySetList {
  131.         repeated PropertySet propertyset = 1;
  132.         extensions                       2 to max;
  133.     }
  135.     message MetaData {
  136.         // Bytes specific metadata
  137.         optional bool   is_multi_part   = 1;
  139.         // General metadata
  140.         optional string content_type    = 2;        // Content/Media type
  141.         optional uint64 size            = 3;        // File size, String size, Multi-part size, etc
  142.         optional uint64 seq             = 4;        // Sequence number for multi-part messages
  144.         // File metadata
  145.         optional string file_name       = 5;        // File name
  146.         optional string file_type       = 6;        // File type (i.e. xml, json, txt, cpp, etc)
  147.         optional string md5             = 7;        // md5 of data
  149.         // Catchalls and future expansion
  150.         optional string description     = 8;        // Could be anything such as json or xml of custom properties
  151.         extensions                      9 to max;
  152.     }
  154.     message Metric {
  156.         optional string   name          = 1;        // Metric name - should only be included on birth
  157.         optional uint64   alias         = 2;        // Metric alias - tied to name on birth and included in all later DATA messages
  158.         optional uint64   timestamp     = 3;        // Timestamp associated with data acquisition time
  159.         optional uint32   datatype      = 4;        // DataType of the metric/tag value
  160.         optional bool     is_historical = 5;        // If this is historical data and should not update real time tag
  161.         optional bool     is_transient  = 6;        // Tells consuming clients such as MQTT Engine to not store this as a tag
  162.         optional bool     is_null       = 7;        // If this is null - explicitly say so rather than using -1, false, etc for some datatypes.
  163.         optional MetaData metadata      = 8;        // Metadata for the payload
  164.         optional PropertySet properties = 9;
  166.         oneof value {
  167.             uint32   int_value                      = 10;
  168.             uint64   long_value                     = 11;
  169.             float    float_value                    = 12;
  170.             double   double_value                   = 13;
  171.             bool     boolean_value                  = 14;
  172.             string   string_value                   = 15;
  173.             bytes    bytes_value                    = 16;       // Bytes, File
  174.             DataSet  dataset_value                  = 17;
  175.             Template template_value                 = 18;
  176.             MetricValueExtension extension_value    = 19;
  177.         }
  179.         message MetricValueExtension {
  180.             extensions  1 to max;
  181.         }
  182.     }
  184.     optional uint64   timestamp     = 1;        // Timestamp at message sending time
  185.     repeated Metric   metrics       = 2;        // Repeated forever - no limit in Google Protobufs
  186.     optional uint64   seq           = 3;        // Sequence number
  187.     optional string   uuid          = 4;        // UUID to track message type in terms of schema definitions
  188.     optional bytes    body          = 5;        // To optionally bypass the whole definition above
  189.     extensions                      6 to max;   // For third party extensions
  190. }

Create rule

SQL statement

  1. SELECT
  2.   schema_decode('neuron', payload, 'Payload') as SparkPlugB
  3. FROM
  4.   "spBv1.0/group1/DDATA/node1/modbus"

The key point here is schema_decode('neuron', payload, 'Payload'):

  • schema_decode The function decodes the content of the payload field according to the Schema ‘protobuf_person’;
  • as SparkPlugB Store the decoded value in the variable “SparkPlugB”;
  • The last parameter Payload indicates that the message type in the payload is the ‘Payload’ type defined in the protobuf schema.

image-20230421143952695

Then add the action with the following parameters:

  • Action Type: Message Repost
  • Purpose topic: SparkPlugB/test

This action sends the decoded “Payload” to the SparkPlugB/test topic in JSON format.

image-20230421143453011

Verify

Here, the MQTTX tool is used to subscribe to the data decoded by the codec function of the EMQX rule engine, as shown in the figure:

image-20230421151817047

As shown in the figure above, it can be seen that the original data before decoding is garbled, and the complete and correct data result is obtained after decoding; so far, the point value of the device is collected in the south direction of Neuron, and reported to EMQX in the north direction of SparkPlugB, and the complete data is obtained by decoding through the codec function Data results are complete.

Appendix

The topic that Neuron reports data to EMQX is namespace/group_id/DDATA/edge_node_id/device_id defined according to the Sparkplug B protocol specification, as shown in the figure:

image-20230419143059088

As for how to define more Neuron northbound Sparkplug B plug-in related standards, you can refer to the Sparkplug B protocol specification🔗Sparkplug Application - 图18 (opens new window)。