ClickHouse Integration

How to integrate ClickHouse with Connectware, including streaming shop floor data into a MergeTree table over the ClickHouse HTTP interface and polling aggregated results back into an MQTT topic.

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 with sufficient user permissions.

• Basic knowledge of MQTT and the Connectware services concept (for example, service commissioning files, connections, and endpoints).

Connectware and ClickHouse Integration

ClickHouse exposes an HTTP interface 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:

• Write endpoints send a POST request per MQTT message. The query property of the endpoint holds the INSERT statement with the 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).

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 ClickHouse instance, for example with clickhouse-client or the ClickHouse Cloud SQL console:

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);

The table uses the 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.

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

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.

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

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

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
                  }
                }

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
}

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.

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.

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'

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 ClickHouse response, including the result rows in value.data and query statistics:

{
  "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 }
  }
}

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 to inspect the polled aggregation on the enterprise/production-events/throughput topic.

Service Commissioning File Example

clickhouse-example.yml

5KB

Open

clickhouse-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# 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'