> 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-5-0/guides/system-connectivity/databases-time-series-storage/snowflake-integration.md). # Snowflake Integration This guide describes how to integrate Snowflake with Connectware. You configure a service commissioning file that produces shop floor data from MQTT topics to a Kafka topic, and the Snowflake Kafka connector ingests that topic into a Snowflake table with Snowpipe Streaming. A complete example file for the Connectware side is available at the end of this guide. ## Objectives * Streaming shop floor data from an ISA-95-style topic hierarchy to a Kafka topic with the Kafka connector. * Configuring the Snowflake Kafka connector to ingest the topic into a Snowflake table with Snowpipe Streaming. * Querying the ingested rows in Snowflake. ## Prerequisites To follow this guide, you will need the following: * A running instance of Cybus Connectware. * A Kafka cluster that is reachable from Connectware, and a Kafka Connect environment where you can deploy the Snowflake Kafka connector. This can be self-managed Kafka or a managed service such as Confluent Cloud. * A Snowflake account, and a role that can create tables in the target schema. * Access to the [Admin UI](/2-5-0/access/admin-ui.md) with sufficient [user permissions](/2-5-0/access/user-management.md). * Basic knowledge of MQTT and the Connectware [services](/2-5-0/data-flows/services.md) concept (for example, [service commissioning files](/2-5-0/data-flows/service-commissioning-files.md), [connections](/2-5-0/data-flows/service-commissioning-files/resources/cybus-connection.md), and [endpoints](/2-5-0/data-flows/service-commissioning-files/resources/cybus-endpoint.md)). ## Connectware and Snowflake Integration Snowflake does not ingest MQTT natively, and row-by-row `INSERT` statements would keep a warehouse running for a use case that is pure streaming. The recommended path is Kafka, which both Connectware and Snowflake support natively: 1. Connectware subscribes to the shop floor topics and produces every message to a Kafka topic through the [Kafka connector](/2-5-0/connectors/enterprise-connectors/kafka.md). 2. The [Snowflake Kafka connector](https://docs.snowflake.com/en/user-guide/kafka-connector-overview), deployed in your Kafka Connect environment, consumes the topic and streams the records into a Snowflake table. With the [Snowpipe Streaming](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-kafka) ingestion method, the rows are written directly into the table without files, stages, or a running warehouse, and become queryable within seconds. Kafka also decouples the two sides: if Snowflake is briefly unavailable, the data waits in the topic instead of getting lost. 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 travel inside each Kafka record. ## Producing Production Events to Kafka This guide shows the producer setup with SASL SCRAM authentication. For creating the Kafka topic, other security configurations such as mutual TLS, and the details of the record format, see the [Apache Kafka Integration](/2-5-0/guides/system-connectivity/messaging-event-streaming/apache-kafka-integration.md) guide, or the [Confluent Cloud Integration](/2-5-0/guides/system-connectivity/messaging-event-streaming/confluent-cloud-integration.md) guide if your cluster runs on Confluent Cloud. 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. * `brokers`: The broker address of your Kafka cluster. * `saslUsername` and `saslPassword`: The SASL credentials of the Connectware client. * `produceTopic`: The Kafka topic that the Snowflake Kafka connector ingests. Defaults to `factory.production-events`. * `topicRoot`: The root of the MQTT topic hierarchy. Defaults to `enterprise`. {% code lineNumbers="true" %} ```yaml parameters: brokers: description: Broker address of the Kafka cluster type: string default: kafka-1.example.com:9093 saslUsername: description: SASL username of the Connectware client type: string default: connectware saslPassword: description: SASL password of the Connectware client type: string produceTopic: description: Kafka topic that the Snowflake Kafka connector ingests type: string default: factory.production-events topicRoot: description: Root of the MQTT topic hierarchy type: string default: enterprise resources: kafkaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref brokers clientType: producer sasl: mechanism: scram-sha-512 username: !ref saslUsername password: !ref saslPassword ``` {% endcode %} The write endpoint produces to the Kafka topic, and the mapping feeds it from the topic hierarchy. It uses [named wildcards](/2-5-0/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 the record in the format that the Kafka connector expects, a `value` array of records: * `value`: The record content as a string, built with the JSONata `$string` function. It combines the topic levels with the `event_type` and `value` fields of the machine payload into one JSON document, which Snowflake later parses back into a queryable object. * `key`: The equipment path. Kafka keeps all records with the same key on the same partition, which preserves the order of events per cell. {% code lineNumbers="true" %} ```yaml productionEventEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref kafkaProducerConnection write: topic: !ref produceTopic productionEventMapping: type: Cybus::Mapping properties: mappings: - subscribe: topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events' publish: endpoint: !ref productionEventEndpoint rules: - transform: expression: >- { "value": [ { "key": $context.vars.site & "/" & $context.vars.area & "/" & $context.vars.line & "/" & $context.vars.cell, "value": $string({ "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 becomes one record on the Kafka topic. The machine payload only needs the event fields, the topic provides the rest: {% code lineNumbers="true" %} ```json { "event_type": "temperature", "value": 42.1 } ``` {% endcode %} ## Ingesting the Kafka Topic into Snowflake The Snowflake Kafka connector runs in your Kafka Connect environment, not in Connectware. It authenticates to Snowflake with a key pair: generate one, assign the public key to the Snowflake user, and give the connector the private key. The user needs a role with `USAGE` on the target database and schema and `CREATE TABLE` on the schema. See the [Snowflake Kafka connector documentation](https://docs.snowflake.com/en/user-guide/kafka-connector-install) for the full installation and key setup. The following configuration ingests the topic with Snowpipe Streaming, for example submitted to the Kafka Connect REST API: {% code title="snowflake-connector.json" lineNumbers="true" %} ```json { "name": "snowflake-production-events", "config": { "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector", "topics": "factory.production-events", "snowflake.url.name": "https://example-org-account.snowflakecomputing.com:443", "snowflake.user.name": "CONNECTWARE_KAFKA", "snowflake.private.key": "${SNOWFLAKE_PRIVATE_KEY}", "snowflake.role.name": "KAFKA_INGEST", "snowflake.database.name": "FACTORY", "snowflake.schema.name": "SHOP_FLOOR", "snowflake.topic2table.map": "factory.production-events:PRODUCTION_EVENTS", "snowflake.ingestion.method": "SNOWPIPE_STREAMING", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter.schemas.enable": "false", "tasks.max": "1" } } ``` {% endcode %} * `${SNOWFLAKE_PRIVATE_KEY}`: The private key of the key pair as one line without the PEM header and footer. Provide it through the secret handling of your Kafka Connect environment rather than in plain text. * `snowflake.url.name`: Your account identifier followed by `snowflakecomputing.com`. * `snowflake.topic2table.map`: Maps the Kafka topic to the target table. Without this setting, the connector derives the table name from the topic name. The connector creates the table if it does not exist, with two `VARIANT` columns: `RECORD_CONTENT` holds the JSON document that Connectware produced, and `RECORD_METADATA` holds the Kafka metadata, including the record key, topic, partition, offset, and the `CreateTime` timestamp of the record. A view gives downstream users typed columns instead of raw JSON: {% code lineNumbers="true" %} ```sql CREATE VIEW FACTORY.SHOP_FLOOR.PRODUCTION_EVENTS_FLAT AS SELECT TO_TIMESTAMP_LTZ(RECORD_METADATA:CreateTime::NUMBER, 3) AS event_time, RECORD_CONTENT:site::STRING AS site, RECORD_CONTENT:area::STRING AS area, RECORD_CONTENT:line::STRING AS line, RECORD_CONTENT:cell::STRING AS cell, RECORD_CONTENT:event_type::STRING AS event_type, RECORD_CONTENT:event_value::DOUBLE AS event_value FROM FACTORY.SHOP_FLOOR.PRODUCTION_EVENTS; ``` {% endcode %} {% hint style="info" %} The connector can also write the JSON keys into real table columns instead of `RECORD_CONTENT`. Set `snowflake.enable.schematization` to `true` if you prefer typed columns over the view. See the [Snowflake Kafka connector documentation](https://docs.snowflake.com/en/user-guide/kafka-connector-overview) for the trade-offs. {% endhint %} ## Verifying the Integration 1. Install the service and set the parameters with the values of your Kafka cluster. 2. Check that the connection is in the **Connected** state on the service details page in the Admin UI. 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. The result of every produce request is published to the `/res` topic of the endpoint; a failed request carries an `error` property. 4. Check that the record arrived on the Kafka topic, for example with `kafka-console-consumer` or the topic view of your managed Kafka service. 5. Check that the row arrived in Snowflake, for example with `SELECT * FROM FACTORY.SHOP_FLOOR.PRODUCTION_EVENTS_FLAT ORDER BY event_time DESC LIMIT 10` in a Snowsight worksheet. With Snowpipe Streaming, the row appears within seconds. If it does not, check the connector status and logs in Kafka Connect. ## Service Commissioning File Example {% file src="/files/I9M3dqb0ma6DLvifI01v" %} {% code title="snowflake-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # Snowflake Integration (Example) # ---------------------------------------------------------------------------- description: > Service commissioning file for the integration between Connectware and Snowflake through Kafka and the Snowflake Kafka connector (Example) metadata: name: Snowflake Integration provider: cybus homepage: https://www.cybus.io version: 1.0.0 parameters: brokers: description: Broker address of the Kafka cluster type: string default: kafka-1.example.com:9093 saslUsername: description: SASL username of the Connectware client type: string default: connectware saslPassword: description: SASL password of the Connectware client type: string produceTopic: description: Kafka topic that the Snowflake Kafka connector ingests type: string default: factory.production-events topicRoot: description: Root of the MQTT topic hierarchy type: string default: enterprise resources: # Producer connection using SASL SCRAM-SHA-512. # Configuring the sasl property makes the connector connect over TLS. # If your brokers use certificates from an internal CA, add the caCert # property with the Base64-encoded PEM content of the CA certificate. kafkaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref brokers clientType: producer sasl: mechanism: scram-sha-512 username: !ref saslUsername password: !ref saslPassword # Produces the production events to the Kafka topic that the # Snowflake Kafka connector ingests productionEventEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref kafkaProducerConnection write: topic: !ref produceTopic # Streams production events from the MQTT topic hierarchy to Kafka. # The named wildcards provide the topic levels in $context.vars, so # every Kafka record carries its origin in the equipment hierarchy. # The record key keeps all events of one cell on the same partition. productionEventMapping: type: Cybus::Mapping properties: mappings: - subscribe: topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events' publish: endpoint: !ref productionEventEndpoint rules: - transform: expression: >- { "value": [ { "key": $context.vars.site & "/" & $context.vars.area & "/" & $context.vars.line & "/" & $context.vars.cell, "value": $string({ "site": $context.vars.site, "area": $context.vars.area, "line": $context.vars.line, "cell": $context.vars.cell, "event_type": event_type, "event_value": value }) } ] } ``` {% 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-5-0/guides/system-connectivity/databases-time-series-storage/snowflake-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.