MariaDB Integration

How to integrate MariaDB with Connectware, including writing machine state changes into a MariaDB table and polling an aggregated machine state summary back into an MQTT topic.

This guide describes how to integrate MariaDB with Connectware. You configure a service commissioning file that writes machine state changes into a MariaDB table and polls an aggregated state summary back into an MQTT topic on a fixed interval. A complete example file is available at the end of this guide.

Objectives

• Establishing a connection between Connectware and MariaDB.

• Writing machine state changes from an ISA-95-style topic hierarchy into a machine states table.

• Polling an aggregated machine state summary, the raw input for Overall Equipment Effectiveness (OEE) calculations, into an MQTT topic on a fixed interval.

Prerequisites

To follow this guide, you will need the following:

• A running instance of Cybus Connectware.

• A MariaDB server that is reachable from Connectware.

• A database user with INSERT and SELECT permissions on the target database.

• Access to the Admin UI with sufficient user permissions.

• Basic knowledge of MQTT and the Connectware services concept (for example, service commissioning files, connections, and endpoints).

Connectware and MariaDB Integration

Connectware communicates with MariaDB through the SQL connector. The connector works by defining SQL queries or query templates on endpoints:

• Write endpoints define a query template, typically an INSERT statement. The template contains placeholders in the form $identifier, which the connector replaces with the values from the JSON payload of each incoming MQTT message (see Placeholder Syntax).

• Subscribe endpoints define a query and a polling interval. The connector executes the query on a regular basis and publishes the result rows as a JSON array to the MQTT broker. The query can be any SELECT statement the database can run, including aggregations, so the database does the heavy lifting and Connectware publishes a precomputed summary instead of raw rows.

The MQTT topics in this guide follow an ISA-95-style equipment hierarchy (<enterprise>/<site>/<area>/<line>/<cell>). The mapping subscribes with named wildcards across all levels, so any machine in the hierarchy is picked up without changing the integration, and the topic levels become column values in the database.

Machine States Table

The integration writes into a single table that stores one row per completed machine state interval. Whenever a machine changes state, it reports the state it is leaving and how long it was in that state. Create the table in your target database:

CREATE TABLE machine_states (
  id BIGINT AUTO_INCREMENT PRIMARY KEY,
  state_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  site VARCHAR(64) NOT NULL,
  area VARCHAR(64) NOT NULL,
  line VARCHAR(64) NOT NULL,
  cell VARCHAR(64) NOT NULL,
  machine_state VARCHAR(32) NOT NULL,
  duration_seconds INT NOT NULL,
  INDEX idx_state_time (state_time)
);

The TIMESTAMP column defaults to the insert time, so the messages do not need to carry a timestamp. If you want to store machine timestamps instead, add a column for them and extend the INSERT statement with another placeholder. The index on state_time keeps the time-windowed polling query fast as the table grows.

MariaDB Connection Properties

We add the connection values as parameters to the service commissioning file, so you can set them when you install the service.

Do not worry about copying the service commissioning file snippets together into one, the complete example file is available at the end of this guide.

• mariadbHost and mariadbPort: The hostname and port of your MariaDB server. The default port is 3306.

• mariadbUsername and mariadbPassword: The database user that Connectware uses to connect.

• mariadbDatabase: The database that contains the machine_states table.

• pollingInterval: The interval in milliseconds between polling queries.

• topicRoot: The root of the MQTT topic hierarchy. Defaults to enterprise.

parameters:
  mariadbHost:
    description: Hostname or IP address of the MariaDB server
    type: string
    default: example.com

  mariadbPort:
    description: Port of the MariaDB server
    type: integer
    default: 3306

  mariadbUsername:
    description: Database user that Connectware uses to connect
    type: string
    default: connectware

  mariadbPassword:
    description: Password of the database user
    type: string

  mariadbDatabase:
    description: Database that contains the machine_states table
    type: string
    default: factory

  pollingInterval:
    description: Interval in milliseconds between polling queries
    type: integer
    default: 60000

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

MariaDB Connection

The SQL connector addresses the database with a single connection URL of the form mariadb://<user>:<password>@<host>:<port>/<database> (see Connection). We set up a Cybus::Connection resource and compose the URL from the parameters with !sub. All endpoints in this guide share this connection.

resources:
  mariadbConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'mariadb://${mariadbUsername}:${mariadbPassword}@${mariadbHost}:${mariadbPort}/${mariadbDatabase}'

If the password contains characters that are not allowed in URLs, such as @ or /, percent-encode them in the parameter value.

For all available connection properties, including the certificate handling, the query log level, and the reconnection strategy, see SQL Connection Properties.

Writing Machine State Changes

The write endpoint defines the INSERT statement as a query template. Each placeholder, for example $site, is replaced with the value of the matching key in the JSON payload of the incoming message.

The mapping feeds the endpoint from the topic hierarchy. It uses named wildcards, so the topic levels are available in the $context.vars object of the transform rule. The rule builds one flat object whose keys match the query placeholders, combining the topic levels with the machine_state and duration_seconds fields of the machine payload.

machineStateWrite:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref mariadbConnection
    write:
      query: >-
        INSERT INTO machine_states
        (site, area, line, cell, machine_state, duration_seconds)
        VALUES ($site, $area, $line, $cell, $machine_state, $duration_seconds)

machineStateMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-state'
        publish:
          endpoint: !ref machineStateWrite
        rules:
          - transform:
              expression: >-
                {
                  "site": $context.vars.site,
                  "area": $context.vars.area,
                  "line": $context.vars.line,
                  "cell": $context.vars.cell,
                  "machine_state": machine_state,
                  "duration_seconds": duration_seconds
                }

Any message published to a matching topic, for example enterprise/hamburg/assembly/line-1/press-01/machine-state, now inserts one row into the table. The machine payload only needs the state fields, the topic provides the rest:

{
  "machine_state": "RUNNING",
  "duration_seconds": 118
}

All placeholders defined in the query must exist in the resulting message payload. If one is missing, the connector logs an error and ignores the message (see Writing Data).

The SQL connection on the Connectware side does not perform any data validation against the database schema. The senders of the MQTT messages must ensure that the values match the column types, for example an integer for the duration_seconds column.

Polling a Machine State Summary

For the opposite direction, a subscribe endpoint executes a query on a fixed interval and publishes the result rows to the MQTT broker. Polled queries are not limited to reading raw rows back. This example aggregates directly in the database: it groups the state intervals of the last hour by line, cell, and state, counts the state changes, and sums the time spent in each state. The result is the time-in-state distribution per machine, the raw input for OEE availability. For example, the summed RUNNING seconds divided by 3,600 is the availability of the last hour.

stateSummaryQuery:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref mariadbConnection
    subscribe:
      query: >-
        SELECT line, cell, machine_state,
        COUNT(*) AS state_changes,
        SUM(duration_seconds) AS total_seconds
        FROM machine_states
        WHERE state_time > NOW() - INTERVAL 1 HOUR
        GROUP BY line, cell, machine_state
        ORDER BY line, cell, machine_state
      interval: !ref pollingInterval

stateSummaryMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref stateSummaryQuery
        publish:
          topic: !sub '${topicRoot}/oee/state-summary'

Choose the polling interval carefully. A low value can overload the database. Instead of a fixed interval, you can also poll on a schedule with the cronExpression property (see SQL Endpoint Properties).

Every query execution publishes a message with a timestamp property and a value property that contains the result rows (see Output Format on Read). If the query returns no rows, value is an empty array ([]).

{
  "timestamp": 1784160000000,
  "value": [
    {
      "line": "line-1",
      "cell": "press-01",
      "machine_state": "BLOCKED",
      "state_changes": 13,
      "total_seconds": 870
    },
    {
      "line": "line-1",
      "cell": "press-01",
      "machine_state": "RUNNING",
      "state_changes": 14,
      "total_seconds": 2730
    }
  ]
}

Verifying the Integration

1. Install the service and set the parameters with the values of your MariaDB server.

2. Check that the connection is in the Connected state on the service details page in the Admin UI. If the credentials or the database name are wrong, the connection does not reach the connected state.

3. Publish a test message with the machine payload shown in this guide to enterprise/hamburg/assembly/line-1/press-01/machine-state, for example with an MQTT client or the Admin UI.

4. Check that the row arrived in the database, for example with SELECT * FROM machine_states ORDER BY id DESC LIMIT 10. The result of every write is also published to the /res topic of the write endpoint, with value set to true on success.

5. Use the Data Explorer to inspect the aggregated summary on the enterprise/oee/state-summary topic.

Service Commissioning File Example

mariadb-example.yml

4KB

Open

mariadb-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# MariaDB Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for the integration between Connectware and
  MariaDB (Example)

metadata:
  name: MariaDB Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  mariadbHost:
    description: Hostname or IP address of the MariaDB server
    type: string
    default: example.com

  mariadbPort:
    description: Port of the MariaDB server
    type: integer
    default: 3306

  mariadbUsername:
    description: Database user that Connectware uses to connect
    type: string
    default: connectware

  mariadbPassword:
    description: Password of the database user
    type: string

  mariadbDatabase:
    description: Database that contains the machine_states table
    type: string
    default: factory

  pollingInterval:
    description: Interval in milliseconds between polling queries
    type: integer
    default: 60000

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

resources:
  # Connection to the MariaDB database
  mariadbConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'mariadb://${mariadbUsername}:${mariadbPassword}@${mariadbHost}:${mariadbPort}/${mariadbDatabase}'

  # Inserts one row into the machine_states table per state change
  machineStateWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref mariadbConnection
      write:
        query: >-
          INSERT INTO machine_states
          (site, area, line, cell, machine_state, duration_seconds)
          VALUES ($site, $area, $line, $cell, $machine_state, $duration_seconds)

  # Feeds the write endpoint from the ISA-95 topic hierarchy. The named
  # wildcards provide the topic levels in $context.vars, so the transform
  # rule can build one flat object whose keys match the query placeholders.
  machineStateMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-state'
          publish:
            endpoint: !ref machineStateWrite
          rules:
            - transform:
                expression: >-
                  {
                    "site": $context.vars.site,
                    "area": $context.vars.area,
                    "line": $context.vars.line,
                    "cell": $context.vars.cell,
                    "machine_state": machine_state,
                    "duration_seconds": duration_seconds
                  }

  # Aggregates the machine states of the last hour per line, cell, and state
  stateSummaryQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref mariadbConnection
      subscribe:
        query: >-
          SELECT line, cell, machine_state,
          COUNT(*) AS state_changes,
          SUM(duration_seconds) AS total_seconds
          FROM machine_states
          WHERE state_time > NOW() - INTERVAL 1 HOUR
          GROUP BY line, cell, machine_state
          ORDER BY line, cell, machine_state
        # Be careful with low values, they can overload the database
        # (the unit is milliseconds).
        interval: !ref pollingInterval

  stateSummaryMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref stateSummaryQuery
          publish:
            topic: !sub '${topicRoot}/oee/state-summary'