> For the complete documentation index, see [llms.txt](https://docs.cybus.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.cybus.io/2-4-2/guides/system-connectivity/databases-time-series-storage/timescaledb-integration.md). # TimescaleDB Integration 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](https://www.tigerdata.com/docs/self-hosted/latest/install/) 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](/2-4-2/access/admin-ui.md) with sufficient [user permissions](/2-4-2/access/user-management.md). * Basic knowledge of MQTT and the Connectware [services](/2-4-2/data-flows/services.md) concept (for example, [service commissioning files](/2-4-2/data-flows/service-commissioning-files.md), [connections](/2-4-2/data-flows/service-commissioning-files/resources/cybus-connection.md), and [endpoints](/2-4-2/data-flows/service-commissioning-files/resources/cybus-endpoint.md)). ## 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](/2-4-2/connectors/enterprise-connectors/sql.md) 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](/2-4-2/connectors/enterprise-connectors/sql.md#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](https://www.tigerdata.com/docs/api/latest/hyperfunctions/time_bucket) come in, because they run inside the query. The MQTT topics in this guide follow an ISA-95-style equipment hierarchy (`////`). 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: {% code lineNumbers="true" %} ```sql 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')); ``` {% endcode %} 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](https://www.tigerdata.com/docs/reference/timescaledb/hypertables/create_hypertable). 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`. {% code lineNumbers="true" %} ```yaml 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 ``` {% endcode %} ### 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://:@:/`, which the `!sub` substitution assembles from the parameters. Both endpoints in this guide share this connection. {% code lineNumbers="true" %} ```yaml resources: timescaleConnection: type: Cybus::Connection properties: protocol: Sql connection: url: !sub 'postgres://${timescaleUsername}:${timescalePassword}@${timescaleHost}:${timescalePort}/${timescaleDatabase}' ``` {% endcode %} For all available connection properties, including certificate handling and the reconnection strategy, see [SQL Connection Properties](/2-4-2/connectors/enterprise-connectors/sql/sqlconnection.md). ### 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](/2-4-2/data-flows/service-commissioning-files/resources/cybus-mapping.md#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. {% code lineNumbers="true" %} ```yaml 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 } ``` {% endcode %} 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: {% code lineNumbers="true" %} ```json { "timestamp": "2026-07-16T10:40:00.000Z", "sensor": "temperature", "value": 42.1 } ``` {% endcode %} 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](/2-4-2/connectors/enterprise-connectors/sql.md#writing-data)). {% hint style="warning" %} 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. {% endhint %} ### 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. {% code lineNumbers="true" %} ```yaml 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' ``` {% endcode %} 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](/2-4-2/connectors/enterprise-connectors/sql/sqlendpoint.md)). 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 (`[]`). {% code lineNumbers="true" %} ```json { "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 } ] } ``` {% endcode %} If the aggregation query becomes expensive on large hypertables, TimescaleDB [continuous aggregates](https://www.tigerdata.com/docs/use-timescale/latest/continuous-aggregates/about-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](https://www.postgresql.org/docs/current/app-psql.html). 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](/2-4-2/monitoring/data-explorer.md) 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 {% file src="/files/VwZ4IYUfsbd2pbVjmPyZ" %} {% code title="timescaledb-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # 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' ``` {% endcode %} --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.cybus.io/2-4-2/guides/system-connectivity/databases-time-series-storage/timescaledb-integration.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.