PostgreSQL Integration

How to integrate PostgreSQL with Connectware, including writing shop floor data into a production events table and polling query results back into MQTT topics.

This guide describes how to integrate PostgreSQL with Connectware. You configure a service commissioning file that writes shop floor data into a PostgreSQL table and polls query results back into an MQTT topic. A complete example file is available at the end of this guide.

Objectives

• Establishing a connection between Connectware and PostgreSQL.

• Writing shop floor data from an ISA-95-style topic hierarchy into a production events table.

• Polling recent rows from the database 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 PostgreSQL server that is reachable from Connectware.

• A database role with INSERT and SELECT privileges 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 PostgreSQL Integration

Connectware communicates with PostgreSQL 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 same connector also serves MariaDB databases; only the URL scheme in the connection changes.

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.

Production Events Table

The integration writes into a single table that stores one row per shop floor event. Create it in your target database:

CREATE TABLE production_events (
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  event_time timestamptz NOT NULL DEFAULT now(),
  site text NOT NULL,
  area text NOT NULL,
  line text NOT NULL,
  cell text NOT NULL,
  event_type text NOT NULL,
  event_value double precision NOT NULL
);

The event_time column defaults to the insert time, stored as timestamp with time zone, 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.

PostgreSQL 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.

• postgresHost and postgresPort: The hostname and port of your PostgreSQL server. The default port is 5432.

• postgresUsername and postgresPassword: The database role that Connectware uses to connect.

• postgresDatabase: The database that contains the production_events table.

• pollingInterval: The interval in milliseconds between polling queries.

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

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

  postgresPort:
    description: Port of the PostgreSQL server
    type: integer
    default: 5432

  postgresUsername:
    description: Database role that Connectware uses to connect
    type: string
    default: connectware

  postgresPassword:
    description: Password of the database role
    type: string

  postgresDatabase:
    description: Database that contains the production_events table
    type: string
    default: factory

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

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

PostgreSQL Connection

To connect to the database, we set up a Cybus::Connection resource that uses the SQL connector. The connector expects a single connection URL of the form postgres://<user>:<password>@<host>:<port>/<database>, which the !sub substitution assembles from the parameters. All endpoints in this guide share this connection.

resources:
  postgresConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'postgres://${postgresUsername}:${postgresPassword}@${postgresHost}:${postgresPort}/${postgresDatabase}'

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

Writing Production Events

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 event_type and value fields of the machine payload.

productionEventWrite:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref postgresConnection
    write:
      query: >-
        INSERT INTO production_events
        (site, area, line, cell, event_type, event_value)
        VALUES ($site, $area, $line, $cell, $event_type, $event_value)

productionEventMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events'
        publish:
          endpoint: !ref productionEventWrite
        rules:
          - transform:
              expression: >-
                {
                  "site": $context.vars.site,
                  "area": $context.vars.area,
                  "line": $context.vars.line,
                  "cell": $context.vars.cell,
                  "event_type": event_type,
                  "event_value": value
                }

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

{
  "event_type": "temperature",
  "value": 42.1
}

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 a number for the event_value column.

Polling Recent Production Events

For the opposite direction, a subscribe endpoint executes a query on a fixed interval and publishes the result rows to the MQTT broker. This example reads the production events of the last five minutes, so downstream consumers such as dashboards or MES systems receive a regularly updated snapshot.

recentEventsQuery:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref postgresConnection
    subscribe:
      query: >-
        SELECT id, event_time, site, area, line, cell,
        event_type, event_value
        FROM production_events
        WHERE event_time > now() - interval '5 minutes'
        ORDER BY event_time DESC
        LIMIT 50
      interval: !ref pollingInterval

recentEventsMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref recentEventsQuery
        publish:
          topic: !sub '${topicRoot}/production-events/recent'

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. If the query returns no rows, value is an empty array ([]).

{
  "timestamp": 1768560000000,
  "value": [
    {
      "id": 1,
      "event_time": "2026-01-16T10:40:00.000Z",
      "site": "hamburg",
      "area": "assembly",
      "line": "line-1",
      "cell": "press-01",
      "event_type": "temperature",
      "event_value": 42.1
    }
  ]
}

Verifying the Integration

1. Install the service and set the parameters with the values of your PostgreSQL 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/production-events, for example with an MQTT client or the Admin UI.

4. Check that the row arrived in the database, for example with SELECT * FROM production_events ORDER BY id DESC LIMIT 10 in psql. 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 polled result rows on the enterprise/production-events/recent topic.

Service Commissioning File Example

postgresql-example.yml

4KB

Open

postgresql-example.yml

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

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

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

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

  postgresPort:
    description: Port of the PostgreSQL server
    type: integer
    default: 5432

  postgresUsername:
    description: Database role that Connectware uses to connect
    type: string
    default: connectware

  postgresPassword:
    description: Password of the database role
    type: string

  postgresDatabase:
    description: Database that contains the production_events table
    type: string
    default: factory

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

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

resources:
  # Connection to the PostgreSQL database
  postgresConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'postgres://${postgresUsername}:${postgresPassword}@${postgresHost}:${postgresPort}/${postgresDatabase}'

  # Inserts one row into the production_events table per message
  productionEventWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref postgresConnection
      write:
        query: >-
          INSERT INTO production_events
          (site, area, line, cell, event_type, event_value)
          VALUES ($site, $area, $line, $cell, $event_type, $event_value)

  # 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.
  productionEventMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events'
          publish:
            endpoint: !ref productionEventWrite
          rules:
            - transform:
                expression: >-
                  {
                    "site": $context.vars.site,
                    "area": $context.vars.area,
                    "line": $context.vars.line,
                    "cell": $context.vars.cell,
                    "event_type": event_type,
                    "event_value": value
                  }

  # Polls the production events of the last five minutes
  recentEventsQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref postgresConnection
      subscribe:
        query: >-
          SELECT id, event_time, site, area, line, cell,
          event_type, event_value
          FROM production_events
          WHERE event_time > now() - interval '5 minutes'
          ORDER BY event_time DESC
          LIMIT 50
        # Be careful with low values, they can overload the database
        # (the unit is milliseconds).
        interval: !ref pollingInterval

  recentEventsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref recentEventsQuery
          publish:
            topic: !sub '${topicRoot}/production-events/recent'