> 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/clickhouse-integration.md). # ClickHouse Integration This guide describes how to integrate ClickHouse with Connectware. You configure a service commissioning file that streams shop floor data into a ClickHouse table over the ClickHouse HTTP interface and polls an aggregation query 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 ClickHouse HTTP interface. * Streaming shop floor data from an ISA-95-style topic hierarchy into a production events table. * Polling an aggregation query 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 ClickHouse instance that is reachable from Connectware. This can be a self-hosted ClickHouse server or a ClickHouse Cloud service. * A ClickHouse user 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 ClickHouse Integration ClickHouse exposes an [HTTP interface](https://clickhouse.com/docs/en/interfaces/http) on port `8123` (HTTP) or `8443` (HTTPS, the standard port for ClickHouse Cloud). Every request carries a SQL statement in the `query` URL parameter. For an `INSERT`, the rows travel in the request body. Connectware communicates with this interface through the [HTTP/REST connector](/2-4-2/connectors/enterprise-connectors/http-rest.md): * **Write endpoints** send a POST request per MQTT message. The `query` property of the endpoint holds the `INSERT` statement with the [JSONEachRow](https://clickhouse.com/docs/en/interfaces/formats/JSONEachRow) format clause, and the message body becomes the request body, which is exactly one JSON row. * **Subscribe endpoints** send a GET request on a fixed interval. The `query` property holds a `SELECT` statement, and the connector publishes the JSON result to the MQTT broker. ClickHouse treats GET requests as read-only, so the polling endpoint cannot modify data. ClickHouse supports HTTP Basic Authentication and the `X-ClickHouse-User` and `X-ClickHouse-Key` headers. This guide uses Basic Authentication through the `auth` property of the connection. If you prefer the header variant, set the two headers in the `headers` property instead (see [HTTP Connection Properties](/2-4-2/connectors/enterprise-connectors/http-rest/httpconnection.md)). 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 database. ### Production Events Table The integration writes into a single table that stores one row per shop floor event. Create it in your ClickHouse instance, for example with [clickhouse-client](https://clickhouse.com/docs/en/interfaces/cli) or the ClickHouse Cloud SQL console: {% code lineNumbers="true" %} ```sql CREATE DATABASE IF NOT EXISTS factory; CREATE TABLE factory.production_events ( event_time DateTime64(3) DEFAULT now64(3), site LowCardinality(String), area LowCardinality(String), line LowCardinality(String), cell LowCardinality(String), event_type LowCardinality(String), event_value Float64 ) ENGINE = MergeTree ORDER BY (site, area, line, cell, event_time); ``` {% endcode %} The table uses the [MergeTree](https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/mergetree) engine, the standard choice for high-volume time series data in ClickHouse. The `ORDER BY` clause sorts the data by location and time, which makes queries that filter on the equipment hierarchy fast. `LowCardinality(String)` compresses the location columns, which contain few distinct values. The `event_time` column defaults to the insert time, so the messages do not need to carry a timestamp. If you want to store machine timestamps instead, send an `event_time` key in the row. ### ClickHouse 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. * `clickhouseScheme`: `http` or `https`. Use `https` for ClickHouse Cloud. * `clickhouseHost` and `clickhousePort`: The hostname and port of the ClickHouse HTTP interface. The default port is `8123` for HTTP and `8443` for HTTPS. * `clickhouseUsername` and `clickhousePassword`: The ClickHouse user that Connectware uses to connect. * `clickhouseDatabase`: 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`. {% code lineNumbers="true" %} ```yaml parameters: clickhouseScheme: description: Transport scheme, use https for ClickHouse Cloud type: string allowedValues: - http - https default: http clickhouseHost: description: Hostname or IP address of the ClickHouse HTTP interface type: string default: example.com clickhousePort: description: Port of the ClickHouse HTTP interface (8123 for http, 8443 for https) type: integer default: 8123 clickhouseUsername: description: ClickHouse user that Connectware uses to connect type: string default: default clickhousePassword: description: Password of the ClickHouse user type: string clickhouseDatabase: 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 ``` {% endcode %} ### ClickHouse Connection To connect to ClickHouse, we set up a `Cybus::Connection` resource that uses the HTTP/REST connector with Basic Authentication. All endpoints in this guide share this connection. The connection also configures the health probing. ClickHouse answers `GET /ping` with `Ok.` 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. {% code lineNumbers="true" %} ```yaml resources: clickhouseConnection: type: Cybus::Connection properties: protocol: Http connection: scheme: !ref clickhouseScheme host: !ref clickhouseHost port: !ref clickhousePort auth: username: !ref clickhouseUsername password: !ref clickhousePassword probePath: /ping probeMethod: GET ``` {% endcode %} ### Streaming Production Events into ClickHouse The write endpoint sends one POST request per MQTT message. Its `query` property adds the static URL parameters to every request: * `query`: The `INSERT` statement. The `FORMAT JSONEachRow` clause tells ClickHouse to parse the request body as one JSON object per row, matching keys to column names. Columns that the row omits, such as `event_time`, take their default values. * `database`: The default database for the statement, so the table name stays unqualified. * `async_insert`: Enables [asynchronous inserts](https://clickhouse.com/docs/en/optimize/asynchronous-inserts). Every MQTT message becomes its own `INSERT` request, and many small direct inserts degrade MergeTree performance. With asynchronous inserts, ClickHouse buffers incoming rows server-side and flushes them in batches. The companion setting `wait_for_async_insert` defaults to `1`, so ClickHouse acknowledges a request only after the buffer is flushed and flush errors still reach Connectware. 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 HTTP/REST connector expects the request body in the `body` property of the message (see [Publishing Data to REST Servers](/2-4-2/connectors/enterprise-connectors/http-rest.md#publishing-data-to-rest-servers)). The rule builds this object, combining the topic levels with the `event_type` and `value` fields of the machine payload. Because `body` is a JSON object, the connector serializes it into the request body as a single JSON object, which is exactly one valid JSONEachRow row. {% code lineNumbers="true" %} ```yaml productionEventWrite: type: Cybus::Endpoint properties: protocol: Http connection: !ref clickhouseConnection write: path: / query: database: !ref clickhouseDatabase query: INSERT INTO production_events FORMAT JSONEachRow async_insert: '1' productionEventMapping: type: Cybus::Mapping properties: mappings: - subscribe: topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events' publish: endpoint: !ref productionEventWrite rules: - transform: expression: >- { "body": { "site": $context.vars.site, "area": $context.vars.area, "line": $context.vars.line, "cell": $context.vars.cell, "event_type": event_type, "event_value": value } } ``` {% endcode %} 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: {% code lineNumbers="true" %} ```json { "event_type": "temperature", "value": 42.1 } ``` {% endcode %} {% hint style="warning" %} ClickHouse rejects rows whose values do not match the column types, for example a string in the `event_value` column. 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 match the table schema. {% endhint %} ### 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 aggregates the production events of the last five minutes per line and event type, so downstream consumers such as dashboards or MES systems receive a regularly updated snapshot. The `FORMAT JSON` clause makes ClickHouse return a JSON document with the result rows in its `data` property. The `output_format_json_quote_64bit_integers` parameter stops ClickHouse from quoting 64-bit integers as strings, so the `count()` result arrives as a number. {% code lineNumbers="true" %} ```yaml throughputQuery: type: Cybus::Endpoint properties: protocol: Http connection: !ref clickhouseConnection subscribe: path: / interval: !ref pollingInterval query: database: !ref clickhouseDatabase output_format_json_quote_64bit_integers: '0' query: >- SELECT site, line, event_type, count() AS events, round(avg(event_value), 2) AS avg_value FROM production_events WHERE event_time > now() - INTERVAL 5 MINUTE GROUP BY site, line, event_type ORDER BY site, line, event_type FORMAT JSON throughputMapping: type: Cybus::Mapping properties: mappings: - subscribe: endpoint: !ref throughputQuery publish: topic: !sub '${topicRoot}/production-events/throughput' ``` {% 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 [HTTP Endpoint Properties](/2-4-2/connectors/enterprise-connectors/http-rest/httpendpoint.md)). Every poll publishes a message with a `timestamp` property and a `value` property that contains the ClickHouse response, including the result rows in `value.data` and query statistics: {% code lineNumbers="true" %} ```json { "timestamp": 1768560000000, "value": { "meta": [ { "name": "site", "type": "LowCardinality(String)" }, { "name": "line", "type": "LowCardinality(String)" }, { "name": "event_type", "type": "LowCardinality(String)" }, { "name": "events", "type": "UInt64" }, { "name": "avg_value", "type": "Float64" } ], "data": [ { "site": "hamburg", "line": "line-1", "event_type": "temperature", "events": 42, "avg_value": 42.1 } ], "rows": 1, "statistics": { "elapsed": 0.002, "rows_read": 42, "bytes_read": 1512 } } } ``` {% endcode %} If downstream consumers only need the rows, add a `transform` rule with the expression `value.data` 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 ClickHouse 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`, which does not require authentication, so wrong credentials surface on the endpoints, not on the connection 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 factory.production_events ORDER BY event_time DESC LIMIT 10` in clickhouse-client or the SQL console. The result of every insert is also published to the `/res` topic of the write endpoint. If ClickHouse rejects a request, the message on the `/res` topic contains an `error` property with the HTTP status and the ClickHouse error text. 5. Use the [Data Explorer](/2-4-2/monitoring/data-explorer.md) to inspect the polled aggregation on the `enterprise/production-events/throughput` topic. ## Service Commissioning File Example {% file src="/files/YLgFf9HvGYegiEDzPVX4" %} {% code title="clickhouse-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # ClickHouse Integration (Example) # ---------------------------------------------------------------------------- description: > Service commissioning file for the integration between Connectware and ClickHouse (Example) metadata: name: ClickHouse Integration provider: cybus homepage: https://www.cybus.io version: 1.0.0 parameters: clickhouseScheme: description: Transport scheme, use https for ClickHouse Cloud type: string allowedValues: - http - https default: http clickhouseHost: description: Hostname or IP address of the ClickHouse HTTP interface type: string default: example.com clickhousePort: description: Port of the ClickHouse HTTP interface (8123 for http, 8443 for https) type: integer default: 8123 clickhouseUsername: description: ClickHouse user that Connectware uses to connect type: string default: default clickhousePassword: description: Password of the ClickHouse user type: string clickhouseDatabase: 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 ClickHouse HTTP interface. ClickHouse answers # GET /ping without touching any data, so we probe there. clickhouseConnection: type: Cybus::Connection properties: protocol: Http connection: scheme: !ref clickhouseScheme host: !ref clickhouseHost port: !ref clickhousePort auth: username: !ref clickhouseUsername password: !ref clickhousePassword probePath: /ping probeMethod: GET # Inserts one row into the production_events table per message. # async_insert makes ClickHouse buffer the many small inserts # server-side and flush them in batches. productionEventWrite: type: Cybus::Endpoint properties: protocol: Http connection: !ref clickhouseConnection write: path: / query: database: !ref clickhouseDatabase query: INSERT INTO production_events FORMAT JSONEachRow async_insert: '1' # 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 row. The connector sends the "body" object as the # request body: a single JSON object, which is one JSONEachRow row. productionEventMapping: type: Cybus::Mapping properties: mappings: - subscribe: topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events' publish: endpoint: !ref productionEventWrite rules: - transform: expression: >- { "body": { "site": $context.vars.site, "area": $context.vars.area, "line": $context.vars.line, "cell": $context.vars.cell, "event_type": event_type, "event_value": value } } # Polls a five-minute aggregation per line and event type. Subscribe # endpoints poll with GET requests, which ClickHouse treats as read-only. throughputQuery: type: Cybus::Endpoint properties: protocol: Http connection: !ref clickhouseConnection subscribe: path: / # Be careful with low values, they can overload the database # (the unit is milliseconds). interval: !ref pollingInterval query: database: !ref clickhouseDatabase # Return 64-bit integers, such as the count() result, as JSON # numbers instead of strings output_format_json_quote_64bit_integers: '0' query: >- SELECT site, line, event_type, count() AS events, round(avg(event_value), 2) AS avg_value FROM production_events WHERE event_time > now() - INTERVAL 5 MINUTE GROUP BY site, line, event_type ORDER BY site, line, event_type FORMAT JSON throughputMapping: type: Cybus::Mapping properties: mappings: - subscribe: endpoint: !ref throughputQuery publish: topic: !sub '${topicRoot}/production-events/throughput' ``` {% 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/clickhouse-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.