TimescaleDB Integration

How to integrate TimescaleDB with Connectware, including writing time-series sensor data into a hypertable and polling time_bucket aggregates back into MQTT topics.

This guide describes how to integrate TimescaleDB with Connectware. You configure a service commissioning file that streams time-series sensor data from the shop floor into a TimescaleDB hypertable and polls aggregated 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 TimescaleDB.

• Writing sensor values from an ISA-95-style topic hierarchy into a hypertable, including the machine timestamps.

• Polling time-bucketed aggregates from the hypertable 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 with the TimescaleDB extension installed, reachable from Connectware. This guide uses the by_range dimension builder, which requires TimescaleDB 2.13 or later.

• 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 TimescaleDB Integration

TimescaleDB extends PostgreSQL with hypertables: tables that are automatically partitioned by time and optimized for high ingest rates and time-based queries. That makes it a natural sink for shop floor sensor data, which arrives as a continuous stream of timestamped values.

Because TimescaleDB is a PostgreSQL extension, Connectware communicates with it through the SQL connector using the postgres:// URL scheme. No TimescaleDB-specific driver is required. 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. This is where TimescaleDB functions such as time_bucket come in, because they run inside the query.

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

Sensor Measurements Hypertable

The integration writes into a single hypertable that stores one row per sensor reading. Create the table in your target database and convert it into a hypertable partitioned by the time column:

CREATE TABLE measurements (
  time timestamptz NOT NULL,
  site text NOT NULL,
  area text NOT NULL,
  line text NOT NULL,
  cell text NOT NULL,
  sensor text NOT NULL,
  value double precision NOT NULL
);

SELECT create_hypertable('measurements', by_range('time'));

TimescaleDB partitions the hypertable into time-based chunks behind the scenes. You keep querying measurements like a regular PostgreSQL table. For all options, such as custom chunk intervals, see the create_hypertable reference.

Unlike a plain event table, the time column has no default. The machines provide their own timestamps, so a reading is stored at the time it was measured, not at the time it arrived in the database.

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

• timescaleHost and timescalePort: The hostname and port of your TimescaleDB server. The default port is 5432.

• timescaleUsername and timescalePassword: The database role that Connectware uses to connect.

• timescaleDatabase: The database that contains the measurements hypertable.

• pollingInterval: The interval in milliseconds between polling queries.

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

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

  timescalePort:
    description: Port of the TimescaleDB server
    type: integer
    default: 5432

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

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

  timescaleDatabase:
    description: Database that contains the measurements hypertable
    type: string
    default: factory

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

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

TimescaleDB 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. Both endpoints in this guide share this connection.

resources:
  timescaleConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'postgres://${timescaleUsername}:${timescalePassword}@${timescaleHost}:${timescalePort}/${timescaleDatabase}'

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

Writing Sensor Measurements

The write endpoint defines the INSERT statement as a query template. Each placeholder, for example $sensor, 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 timestamp, sensor, and value fields of the machine payload.

measurementWrite:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref timescaleConnection
    write:
      query: >-
        INSERT INTO measurements
        (time, site, area, line, cell, sensor, value)
        VALUES ($time, $site, $area, $line, $cell, $sensor, $value)

measurementMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: !sub '${topicRoot}/+site/+area/+line/+cell/measurements'
        publish:
          endpoint: !ref measurementWrite
        rules:
          - transform:
              expression: >-
                {
                  "time": timestamp,
                  "site": $context.vars.site,
                  "area": $context.vars.area,
                  "line": $context.vars.line,
                  "cell": $context.vars.cell,
                  "sensor": sensor,
                  "value": value
                }

Any message published to a matching topic, for example enterprise/hamburg/assembly/line-1/press-01/measurements, now inserts one row into the hypertable. The machine payload carries the timestamp of the reading as an ISO 8601 string, which PostgreSQL casts to the timestamptz column:

{
  "timestamp": "2026-07-16T10:40:00.000Z",
  "sensor": "temperature",
  "value": 42.1
}

If your machines publish Unix timestamps in milliseconds instead, convert them in the query template with to_timestamp($time / 1000.0).

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 value column and a valid timestamp for the time column.

Polling Aggregated Measurements

Publishing every raw reading back to MQTT rarely makes sense at shop floor data rates. Instead, a subscribe endpoint uses the TimescaleDB time_bucket function to downsample the data in the database: the query groups the readings of the last 15 minutes into one-minute buckets per sensor and returns the average, minimum, and maximum for each bucket. Downstream consumers such as dashboards receive a compact, regularly updated summary.

aggregatedMeasurementsQuery:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref timescaleConnection
    subscribe:
      query: >-
        SELECT time_bucket('1 minute', time) AS bucket,
        site, area, line, cell, sensor,
        avg(value) AS avg_value,
        min(value) AS min_value,
        max(value) AS max_value
        FROM measurements
        WHERE time > now() - interval '15 minutes'
        GROUP BY bucket, site, area, line, cell, sensor
        ORDER BY bucket DESC
      interval: !ref pollingInterval

aggregatedMeasurementsMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref aggregatedMeasurementsQuery
        publish:
          topic: !sub '${topicRoot}/measurements/aggregated'

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": 1784198460000,
  "value": [
    {
      "bucket": "2026-07-16T10:40:00.000Z",
      "site": "hamburg",
      "area": "assembly",
      "line": "line-1",
      "cell": "press-01",
      "sensor": "temperature",
      "avg_value": 42.0,
      "min_value": 41.5,
      "max_value": 42.6
    }
  ]
}

If the aggregation query becomes expensive on large hypertables, TimescaleDB continuous aggregates precompute the same time_bucket results incrementally in the background, so the polling endpoint can select from the materialized view instead.

Verifying the Integration

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

4. Check that the row arrived in the hypertable, for example with SELECT * FROM measurements ORDER BY time 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 aggregated buckets on the enterprise/measurements/aggregated topic. The first bucket appears after the polling interval has elapsed.

Service Commissioning File Example

timescaledb-example.yml

4KB

Open

timescaledb-example.yml

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

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

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

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

  timescalePort:
    description: Port of the TimescaleDB server
    type: integer
    default: 5432

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

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

  timescaleDatabase:
    description: Database that contains the measurements hypertable
    type: string
    default: factory

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

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

resources:
  # Connection to the TimescaleDB database. TimescaleDB is a PostgreSQL
  # extension, so the connection uses the postgres URL scheme.
  timescaleConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'postgres://${timescaleUsername}:${timescalePassword}@${timescaleHost}:${timescalePort}/${timescaleDatabase}'

  # Inserts one row into the measurements hypertable per message,
  # including the machine timestamp
  measurementWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref timescaleConnection
      write:
        query: >-
          INSERT INTO measurements
          (time, site, area, line, cell, sensor, value)
          VALUES ($time, $site, $area, $line, $cell, $sensor, $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.
  measurementMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/measurements'
          publish:
            endpoint: !ref measurementWrite
          rules:
            - transform:
                expression: >-
                  {
                    "time": timestamp,
                    "site": $context.vars.site,
                    "area": $context.vars.area,
                    "line": $context.vars.line,
                    "cell": $context.vars.cell,
                    "sensor": sensor,
                    "value": value
                  }

  # Polls one-minute aggregates of the last 15 minutes using time_bucket
  aggregatedMeasurementsQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref timescaleConnection
      subscribe:
        query: >-
          SELECT time_bucket('1 minute', time) AS bucket,
          site, area, line, cell, sensor,
          avg(value) AS avg_value,
          min(value) AS min_value,
          max(value) AS max_value
          FROM measurements
          WHERE time > now() - interval '15 minutes'
          GROUP BY bucket, site, area, line, cell, sensor
          ORDER BY bucket DESC
        # Be careful with low values, they can overload the database
        # (the unit is milliseconds).
        interval: !ref pollingInterval

  aggregatedMeasurementsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref aggregatedMeasurementsQuery
          publish:
            topic: !sub '${topicRoot}/measurements/aggregated'