QuestDB Integration

How to integrate QuestDB with Connectware, including streaming time-series shop floor data over InfluxDB Line Protocol and polling a SAMPLE BY aggregation back into an MQTT topic.

This guide describes how to integrate QuestDB with Connectware. You configure a service commissioning file that streams time-series shop floor data into a QuestDB table over HTTP and polls a SAMPLE BY aggregation back into an MQTT topic. A complete example file is available at the end of this guide.

Objectives

• Establishing a connection between Connectware and the QuestDB HTTP endpoints.

• Streaming shop floor data from an ISA-95-style topic hierarchy into a QuestDB table with InfluxDB Line Protocol (ILP) over HTTP.

• Polling a SAMPLE BY aggregation 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 QuestDB instance that is reachable from Connectware. This can be QuestDB open source or QuestDB Enterprise.

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

QuestDB serves all of its HTTP interfaces on one port, 9000 by default: the REST API with the /exec, /imp, and /exp endpoints, the InfluxDB Line Protocol (ILP) ingestion endpoint /write, the health check /ping, and the Web Console. Connectware communicates with these interfaces through the HTTP/REST connector:

• Write endpoints send a POST request per MQTT message to the /write endpoint. The request body carries the row as ILP text, the ingestion format that QuestDB optimizes for high-frequency inserts.

• Subscribe endpoints send a GET request on a fixed interval to the /exec endpoint, which executes the SQL statement supplied in the query URL parameter and returns the result as JSON. The connector publishes the result to the MQTT broker.

A fresh QuestDB instance accepts HTTP requests without authentication. QuestDB open source supports one user with HTTP Basic Authentication through the http.user and http.password server settings. QuestDB Enterprise adds user management and REST API bearer tokens (see REST API authentication). This guide uses an open instance. For Basic Authentication, add the auth property to the connection. For a bearer token, add an Authorization header through the headers property (see HTTP Connection Properties).

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 Data Table

The integration writes into a single table that stores one row per machine reading. QuestDB creates missing tables automatically on the first ILP line, but creating the table yourself gives you control over the column types and the partitioning. Create it with the /exec endpoint, which executes the SQL statement in the query URL parameter of a GET request:

curl -G http://localhost:9000/exec --data-urlencode "query=
CREATE TABLE machine_data (
  ts TIMESTAMP,
  site SYMBOL,
  area SYMBOL,
  line SYMBOL,
  cell SYMBOL,
  temperature DOUBLE,
  pressure DOUBLE
) TIMESTAMP(ts) PARTITION BY DAY WAL;
"

QuestDB answers with {"ddl":"OK"}. Alternatively, run the statement in the Web Console, which QuestDB serves on the same port.

The SYMBOL columns store the repetitive location values efficiently. TIMESTAMP(ts) marks ts as the designated timestamp, which time-based queries such as SAMPLE BY rely on, and PARTITION BY DAY splits the storage into daily partitions.

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

• questdbScheme: http or https. Use https if TLS is configured for your instance.

• questdbHost and questdbPort: The hostname and port of the QuestDB HTTP endpoints. The default port is 9000.

• pollingInterval: The interval in milliseconds between polling queries.

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

parameters:
  questdbScheme:
    description: Transport scheme, use https if TLS is configured
    type: string
    allowedValues:
      - http
      - https
    default: http

  questdbHost:
    description: Hostname or IP address of the QuestDB instance
    type: string
    default: example.com

  questdbPort:
    description: Port of the QuestDB HTTP endpoints
    type: integer
    default: 9000

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

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

QuestDB Connection

To connect to QuestDB, we set up a Cybus::Connection resource that uses the HTTP/REST connector. All endpoints in this guide share this connection.

The connection also configures the health probing. QuestDB answers GET /ping with an empty response without touching any data, which makes it the ideal probe target. The default probing method of the connector is a HEAD request on /, so we override both properties.

resources:
  questdbConnection:
    type: Cybus::Connection
    properties:
      protocol: Http
      connection:
        scheme: !ref questdbScheme
        host: !ref questdbHost
        port: !ref questdbPort
        probePath: /ping
        probeMethod: GET

Streaming Machine Data into QuestDB

The write endpoint sends one POST request per MQTT message to the /write endpoint. The headers property sets the Content-Type: text/plain header, which is the content type that QuestDB documents for ILP over HTTP.

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 HTTP/REST connector expects the request body in the body property of the message (see Publishing Data to REST Servers). The rule builds one line of ILP text: the table name, the topic levels as tags, and the numeric fields of the machine payload. Tags map to the SYMBOL columns and the fields map to the DOUBLE columns of the table.

machineDataWrite:
  type: Cybus::Endpoint
  properties:
    protocol: Http
    connection: !ref questdbConnection
    write:
      path: /write
      headers:
        Content-Type: text/plain

machineDataMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-data'
        publish:
          endpoint: !ref machineDataWrite
        rules:
          - transform:
              expression: >-
                {
                  "body": "machine_data,site=" & $context.vars.site
                  & ",area=" & $context.vars.area
                  & ",line=" & $context.vars.line
                  & ",cell=" & $context.vars.cell
                  & " temperature=" & $string(temperature)
                  & ",pressure=" & $string(pressure)
                }

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

{
  "temperature": 42.1,
  "pressure": 5.2
}

For this message, the transform rule produces the following request body:

machine_data,site=hamburg,area=assembly,line=line-1,cell=press-01 temperature=42.1,pressure=5.2

The line carries no timestamp, so QuestDB sets the designated timestamp to the server time at ingestion. If your machines send their own timestamps, append the epoch value after the fields, separated by a space, and declare its unit with the precision URL parameter, for example by adding precision: ms to a query property of the write endpoint. Without the parameter, QuestDB interprets timestamps as nanoseconds.

QuestDB rejects lines that do not parse, for example when a topic level or a value contains an unescaped space, comma, or equals sign. The rejected request appears as an error property on the /res topic of the endpoint, and the row is lost. The senders of the MQTT messages must ensure that the values produce valid lines.

Polling an Aggregate Back to MQTT

For the opposite direction, a subscribe endpoint polls a SELECT statement on a fixed interval and publishes the result to the MQTT broker. This example uses SAMPLE BY, the QuestDB SQL extension for time-based aggregation, to average the temperatures of the last five minutes per site and line in one-minute buckets. Downstream consumers such as dashboards or MES systems receive a regularly updated snapshot.

SAMPLE BY groups the rows by their designated timestamp. The non-aggregated site and line columns act as additional grouping keys, so the result contains one row per minute, site, and line.

aggregateQuery:
  type: Cybus::Endpoint
  properties:
    protocol: Http
    connection: !ref questdbConnection
    subscribe:
      path: /exec
      interval: !ref pollingInterval
      query:
        query: >-
          SELECT ts, site, line, avg(temperature) AS avg_temperature,
          count() AS samples
          FROM machine_data
          WHERE ts > dateadd('m', -5, now())
          SAMPLE BY 1m

aggregateMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref aggregateQuery
        publish:
          topic: !sub '${topicRoot}/machine-data/temperature-per-line'

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 HTTP Endpoint Properties).

Every poll publishes a message with a timestamp property and a value property that contains the QuestDB response, including the column schema in value.columns and the result rows as arrays in value.dataset:

{
  "timestamp": 1768560000000,
  "value": {
    "query": "SELECT ts, site, line, avg(temperature) AS avg_temperature, count() AS samples FROM machine_data WHERE ts > dateadd('m', -5, now()) SAMPLE BY 1m",
    "columns": [
      { "name": "ts", "type": "TIMESTAMP" },
      { "name": "site", "type": "SYMBOL" },
      { "name": "line", "type": "SYMBOL" },
      { "name": "avg_temperature", "type": "DOUBLE" },
      { "name": "samples", "type": "LONG" }
    ],
    "dataset": [["2026-07-16T09:05:00.000000Z", "hamburg", "line-1", 42.1, 12]],
    "count": 1
  }
}

If downstream consumers only need the rows, add a transform rule with the expression value.dataset to the mapping, which publishes the plain array instead.

Verifying the Integration

1. Install the service and set the parameters with the values of your QuestDB instance.

2. Check that the connection is in the Connected state on the service details page in the Admin UI. The probing calls GET /ping.

3. Publish a test message with the machine payload shown in this guide to enterprise/hamburg/assembly/line-1/press-01/machine-data, 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_data ORDER BY ts DESC LIMIT 10; in the Web Console. The result of every insert is also published to the /res topic of the write endpoint. If QuestDB rejects a request, the message on the /res topic contains an error property with the HTTP status and the QuestDB error text.

5. Use the Data Explorer to inspect the polled aggregation on the enterprise/machine-data/temperature-per-line topic.

Service Commissioning File Example

questdb-example.yml

4KB

Open

questdb-example.yml

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

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

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

parameters:
  questdbScheme:
    description: Transport scheme, use https if TLS is configured
    type: string
    allowedValues:
      - http
      - https
    default: http

  questdbHost:
    description: Hostname or IP address of the QuestDB instance
    type: string
    default: example.com

  questdbPort:
    description: Port of the QuestDB HTTP endpoints
    type: integer
    default: 9000

  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 QuestDB HTTP endpoints. QuestDB answers GET /ping
  # without touching any data, so we probe there.
  questdbConnection:
    type: Cybus::Connection
    properties:
      protocol: Http
      connection:
        scheme: !ref questdbScheme
        host: !ref questdbHost
        port: !ref questdbPort
        probePath: /ping
        probeMethod: GET

  # Sends one InfluxDB Line Protocol line per message to /write. Lines
  # without a timestamp receive the server time at ingestion.
  machineDataWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Http
      connection: !ref questdbConnection
      write:
        path: /write
        headers:
          Content-Type: text/plain

  # 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 the line: tags from the topic levels, fields from the
  # machine payload.
  machineDataMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-data'
          publish:
            endpoint: !ref machineDataWrite
          rules:
            - transform:
                expression: >-
                  {
                    "body": "machine_data,site=" & $context.vars.site
                    & ",area=" & $context.vars.area
                    & ",line=" & $context.vars.line
                    & ",cell=" & $context.vars.cell
                    & " temperature=" & $string(temperature)
                    & ",pressure=" & $string(pressure)
                  }

  # Polls a five-minute SAMPLE BY aggregation per site and line. Subscribe
  # endpoints poll with GET requests, and /exec executes the SQL statement
  # in the query URL parameter.
  aggregateQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Http
      connection: !ref questdbConnection
      subscribe:
        path: /exec
        # Be careful with low values, they can overload the database
        # (the unit is milliseconds).
        interval: !ref pollingInterval
        query:
          query: >-
            SELECT ts, site, line, avg(temperature) AS avg_temperature,
            count() AS samples
            FROM machine_data
            WHERE ts > dateadd('m', -5, now())
            SAMPLE BY 1m

  aggregateMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref aggregateQuery
          publish:
            topic: !sub '${topicRoot}/machine-data/temperature-per-line'