How to set up a Snowflake integration with HiveMQ

The HiveMQ Snowflake extension allows your HiveMQ broker to stream MQTT messages directly into Snowflake using the Snowpipe Streaming API/SDK.

Below is an overview of how to set up:

Snowflake for MQTT ingestion
HiveMQ for Snowflake integration

📋 Prerequisites

Snowflake
HiveMQ installed
MQTT CLI installed
Snowflake license (.elic)

Instructions

Step 1: Create a Snowflake account

First, we need to sign up for a 30-day trial license to use Snowflake.

Steps:

Go to Snowflake sign-up page:

Fill the sign-up form and once completed, you will receive an email for activation.

Step 2: Create key pair for Snowflake authentication

During HiveMQ startup, the Snowflake extension initiates authentication with Snowflake using a Key Pair. This section will cover how to create the keys.

Steps:

Create the private key:

openssl genrsa -out private_key.pem 2048

Extract the public from the private key:

openssl rsa -in private_key.pem -pubout -out public_key.pem

Note: Keep the keys for safekeeping for the next section in this document.

In this example, we are not using an encrypted key (requires a passphrase) as I was running into the following error:

Caused by: java.security.InvalidKeyException: Wrong algorithm: DESede or TripleDES required

Please see slack post. This may be revisited at a later time.

Step 3: Build the Snowflake database and prepare it for MQTT ingestion

The following steps involve executing SQL queries to build the database, user, roles and MQTT table for ingestion.

Steps:

Log in to the Snowflake console.
Navigate to SQL File:

In the left pane, select Project → Workspaces:

Then click on SQL file:

You should now see the following window:

Note: The play button executes the query in the window or any highlighted queries if multiple are present.

Execute SQL Queries:

Create database:

CREATE DATABASE IOT_DB;

Set session to newly created DB:

USE DATABASE IOT_DB;

Create the messages table, where MQTT messages will be stored:

 CREATE TABLE PUBLIC.IOT_MESSAGES (
  topic          STRING,
  payload        STRING,
  timestamp_iso  TIMESTAMP_TZ
);

Create a Snowflake user with public key (created in Step 2):

CREATE USER snowflake RSA_PUBLIC_KEY = '<copy only the certificate between the headers
<-----BEGIN PUBLIC KEY-----
and 
-----END PUBLIC KEY----- > from the public_key.pem file >';

Create a role for granting insert permissions:

CREATE ROLE MQTT_INGEST_ROLE;
GRANT USAGE ON DATABASE IOT_DB TO ROLE MQTT_INGEST_ROLE;
GRANT USAGE ON SCHEMA IOT_DB.PUBLIC TO ROLE MQTT_INGEST_ROLE;
GRANT INSERT ON TABLE IOT_DB.PUBLIC.IOT_MESSAGES TO ROLE MQTT_INGEST_ROLE;
GRANT ROLE MQTT_INGEST_ROLE TO USER snowflake;

Step 4: Configure HiveMQ to connect to your Snowflake database

With the Snowflake database setup, now let's configure the Snowflake config.xml file in HiveMQ.

Get the Snowflake Account URL:

In the Snowflake console, go to the lower left corner and navigate to Account Icon → Account → View account details:

From View account details, copy the Account/Server URL:

Compose your Snowflake config.xml (including Server URL from above) similar to:

<hivemq-snowflake-extension xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                            xsi:noNamespaceSchemaLocation="config.xsd">
    <snowflakes>
        <snowflake>
            <id>my-snowflake</id>
            <account-url><Server URL></account-url>
            <user>snowflake</user>
            <role>MQTT_INGEST_ROLE</role>
            <private-key>/path/to/private_key.pem</private-key>
            <!--<private-key-passphrase>snowflake</private-key-passphrase> 
            ...not encrypted key-->
        </snowflake>
    </snowflakes>

    <mqtt-to-snowflake-routes>
        <mqtt-to-snowflake-route>
            <id>my-snowflake-route</id>
            <snowflake-id>my-snowflake</snowflake-id>
            <enabled>true</enabled>
            <mqtt-topic-filters>
                <mqtt-topic-filter>#</mqtt-topic-filter>
            </mqtt-topic-filters>
            <processor>
                <snowpipe-streaming>
                    <database>IOT_DB</database>
                    <schema>PUBLIC</schema>
                    <table>IOT_MESSAGES</table>
                    <columns>
                        <column>
                            <name>timestamp_iso</name>
                            <value>timestamp-iso-8601</value>
                        </column>
                        <column>
                            <name>topic</name>
                            <value>mqtt-topic</value>
                        </column>
                        <column>
                            <name>payload</name>
                            <value>mqtt-payload-utf8</value>
                        </column>
                    </columns>
                </snowpipe-streaming>
            </processor>
        </mqtt-to-snowflake-route>
    </mqtt-to-snowflake-routes>
</hivemq-snowflake-extension>

Note: Please see our documentation for additional values recognized for Snowpipe Streaming.

Copy the Snowflake .elic file into <hivemq>/license.
Delete the DISABLED file in <hivemq>/extensions/hivemq-snowflake-extension.
Start HiveMQ service
Publish a message:

mqtt pub -i testclient -t 'factory/machine/sensor' -m 'test message from hivemq'

Verify that the MQTT message has been ingested by Snowflake: