Snowflake Integration

How to integrate Snowflake with Connectware by producing shop floor data to a Kafka topic and ingesting it into a Snowflake table with the Snowflake Kafka connector and Snowpipe Streaming.

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

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

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. The Snowflake Kafka connector, deployed in your Kafka Connect environment, consumes the topic and streams the records into a Snowflake table.

With the Snowpipe Streaming 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 (<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 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 guide, or the Confluent Cloud Integration 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.

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

The write endpoint produces to the Kafka topic, and the mapping feeds it from the topic hierarchy. It uses named 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.

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

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:

{
  "event_type": "temperature",
  "value": 42.1
}

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

snowflake-connector.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"
  }
}

• ${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:

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;

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 for the trade-offs.

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

snowflake-example.yml

3KB

Open

snowflake-example.yml

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