Snowflake

The akenza Snowflake connector enables direct ingestion of device telemetry into the Snowflake Cloud Data Platform for large-scale analytics and cloud-native data warehousing.

The connector uses the Snowflake Snowpipe Streaming API to continuously ingest device data with low latency and high throughput (check its high-performance architecture).

This allows sensor data processed by akenza to be written directly into Snowflake tables without intermediate storage layers.

Setup

Follow the steps below to configure Snowflake for use with the akenza Snowflake connector.

Prerequisites: Before configuring the Snowflake connector, make sure you have access to a Snowflake account. If you do not already have a user, you will first need to create one.

Snowflake users can be created by an account administrator. The process is straightforward and can be done throught the Snowsight UI. Please follow the instructions in the Snowflake documentation. This guide explains how to create and manage users in Snowflake using Snowsight.

Create the database and table

Create a database and a table that will store the incoming device data.

The table schema must match the structure of the custom payload defined in the akenza Snowflake output connector configuration. Each field defined in the custom payload must correspond to a column in the target Snowflake table, and the data types must be compatible.

CREATE DATABASE <database_name>;

CREATE SCHEMA <schema_name>;

CREATE TABLE <database_name>.<schema_name>.<table_name> (
    <col_name> <col_type>,
    <col_name> <col_type>,
    ...
);

Example

CREATE DATABASE TEST_DATABASE;

CREATE SCHEMA TEST_SCHEMA;

CREATE TABLE TEST_DATABASE.TEST_SCHEMA.TEST_TABLE (
    DEVICE_ID VARCHAR,
    TEMPERATURE FLOAT,
    TIMESTAMP TIMESTAMP_TZ
);

Example custom payload

{
    "DEVICE_ID": {{device.id}},
    "TEMPERATURE": {{data.temperature}},
    "TIMESTAMP": {{timestamp}}
}

The names in the payload must match the column names used in the Snowflake table, as these fields are mapped directly when the data is ingested through Snowpipe Streaming.

Create the Snowpipe Streaming pipe

Create a pipe that reads data from the Snowpipe Streaming data source.

CREATE OR REPLACE PIPE <database_name>.<schema_name>.<pipe_name>
AS COPY INTO <database_name>.<schema_name>.<table_name>
FROM (
    SELECT
        $1:<col_name>,
        $1:<col_name,
        ...
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Example

CREATE OR REPLACE PIPE TEST_DATABASE.TEST_SCHEMA.TEST_PIPE
AS COPY INTO TEST_DATABASE.TEST_SCHEMA.TEST_TABLE
FROM (
    SELECT
        $1:DEVICE_ID,
        $1:TEMPERATURE,
        $1:TIMESTAMP
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

This pipe defines how incoming streamed data is mapped to the columns of the target table.

Each field referenced with $1:<field_name> corresponds to a field in the JSON payload sent to Snowflake via Snowpipe Streaming. The field names must match the fields used in the custom payload described in the previous section.

Create a role with required privileges for Snowpipe Streaming ingestion

To follow the principle of least privilege, it is recommended to create a dedicated role that only allows the connector to ingest data through the Snowpipe Streaming pipe. This role will be assigned to a service user used by the connector.

Create the role

Create a dedicated role that will be used for Snowpipe Streaming ingestion. This role will have only the privileges necessary to ingest data

CREATE ROLE <role_name>;

Example

CREATE ROLE TEST_ROLE;

Grant required privileges

Grant the minimal privileges required for the role to access the database, schema, and streaming pipe.

Note: You have two options to run the commands:

1. Run each command separately – place your cursor on a command and execute it. This works fine for all four commands.

2. Run all commands at once – do not just copy and paste all commands and hit run. If you paste all four and execute without selecting them, only the command where your cursor is will run (usually the last one), which can cause errors.

GRANT USAGE ON DATABASE <database_name> TO ROLE <role_name>;
GRANT USAGE ON SCHEMA <database_name>.<schema_name> TO ROLE <role_name>;
GRANT OPERATE ON PIPE <database_name>.<schema_name>.<pipe_name> TO ROLE <role_name>;
GRANT INSERT, SELECT ON TABLE <database_name>.<schema_name>.<table_name> TO ROLE <role_name>;

Example

GRANT USAGE ON DATABASE TEST_DATABASE TO ROLE TEST_ROLE;
GRANT USAGE ON SCHEMA TEST_DATABASE.TEST_SCHEMA TO ROLE TEST_ROLE;
GRANT OPERATE ON PIPE TEST_DATABASE.TEST_SCHEMA.TEST_PIPE TO ROLE TEST_ROLE;
GRANT INSERT, SELECT ON TABLE TEST_DATABASE.TEST_SCHEMA.TEST_TABLE TO ROLE TEST_ROLE;

Create a Snowpipe Streaming user

Create a user that will be used by the connector to authenticate with Snowflake.

CREATE USER <user_name>
DEFAULT_ROLE = <role_name>
DEFAULT_NAMESPACE = <database_name>.<schema_name>
DEFAULT_WAREHOUSE = <warehouse_name>
RSA_PUBLIC_KEY = 'MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAvarY+sda3OZsHZGAdoc9mqiClLbs8rVs0MH/lbPYqg8e45hJ1m02vji12DOhEm/mEUzR2r/42I61osBrgBXHMgNsTHW5Mb1A3GXYrEikZv6HBrvx/6ZoVLn6QuKrdZ4dH0FUKO+6sWCu1yQ74i6ixHLqdKJbBr2Bk3ehB5YlqHb7G4TU9XMwzlzYkLMjEFP5OjLNxCMFthvIuugQ39ULcqZk7OUCbZigQdQhihF+TEbG1aDU6MxLcaQx6rB/Fq2a6iw+E9SkFxTFqQbtCi8hT1dy1pZCCKJFjQEQgMhvuLIgywYSD344RSttci/Bo8CZUjtn8prhkSEIFZaUboLN+rfhbzLzTPTfnqqP4umi5mtZRTqhIieGFmuzkOkCrhWSOuvL2l/SjEcVEHdkpIqb2kquK2EdyDZnEX0n1iaGOTe4Mtr9+wvBrMf8OaeGKro2oX4zIugDNMBzAx2duW8mr3cgyGCVX8M/BG3fdvB1cUfspwmVZNZ0IjZrPU0+69wZxgTcOze1ndXmMDTyzsI4EYeNvD02AeaVmDGY5xe57BtOVz0MQRNoCP0GCCSzjbpUXMpaMgZ/2O/UqmWFd5fMqTFtp57h/UxAHV2xln8rh2Z5JXOThK6qzuA4k8ekRfc+9seCIrPrbTAch6RguXu4sv9eHK2gLmMcOp8j6scwMdsCAwEAAQ==';

The akenza Snowflake connector authenticates using RSA key pair authentication.

Akenza public key:

MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAvarY+sda3OZsHZGAdoc9mqiClLbs8rVs0MH/lbPYqg8e45hJ1m02vji12DOhEm/mEUzR2r/42I61osBrgBXHMgNsTHW5Mb1A3GXYrEikZv6HBrvx/6ZoVLn6QuKrdZ4dH0FUKO+6sWCu1yQ74i6ixHLqdKJbBr2Bk3ehB5YlqHb7G4TU9XMwzlzYkLMjEFP5OjLNxCMFthvIuugQ39ULcqZk7OUCbZigQdQhihF+TEbG1aDU6MxLcaQx6rB/Fq2a6iw+E9SkFxTFqQbtCi8hT1dy1pZCCKJFjQEQgMhvuLIgywYSD344RSttci/Bo8CZUjtn8prhkSEIFZaUboLN+rfhbzLzTPTfnqqP4umi5mtZRTqhIieGFmuzkOkCrhWSOuvL2l/SjEcVEHdkpIqb2kquK2EdyDZnEX0n1iaGOTe4Mtr9+wvBrMf8OaeGKro2oX4zIugDNMBzAx2duW8mr3cgyGCVX8M/BG3fdvB1cUfspwmVZNZ0IjZrPU0+69wZxgTcOze1ndXmMDTyzsI4EYeNvD02AeaVmDGY5xe57BtOVz0MQRNoCP0GCCSzjbpUXMpaMgZ/2O/UqmWFd5fMqTFtp57h/UxAHV2xln8rh2Z5JXOThK6qzuA4k8ekRfc+9seCIrPrbTAch6RguXu4sv9eHK2gLmMcOp8j6scwMdsCAwEAAQ==

Example

CREATE USER TEST_USER
DEFAULT_ROLE = TEST_ROLE
DEFAULT_NAMESPACE = TEST_DATABASE.TEST_SCHEMA
DEFAULT_WAREHOUSE = COMPUTE_WH
RSA_PUBLIC_KEY ='MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAvarY+sda3OZsHZGAdoc9mqiClLbs8rVs0MH/lbPYqg8e45hJ1m02vji12DOhEm/mEUzR2r/42I61osBrgBXHMgNsTHW5Mb1A3GXYrEikZv6HBrvx/6ZoVLn6QuKrdZ4dH0FUKO+6sWCu1yQ74i6ixHLqdKJbBr2Bk3ehB5YlqHb7G4TU9XMwzlzYkLMjEFP5OjLNxCMFthvIuugQ39ULcqZk7OUCbZigQdQhihF+TEbG1aDU6MxLcaQx6rB/Fq2a6iw+E9SkFxTFqQbtCi8hT1dy1pZCCKJFjQEQgMhvuLIgywYSD344RSttci/Bo8CZUjtn8prhkSEIFZaUboLN+rfhbzLzTPTfnqqP4umi5mtZRTqhIieGFmuzkOkCrhWSOuvL2l/SjEcVEHdkpIqb2kquK2EdyDZnEX0n1iaGOTe4Mtr9+wvBrMf8OaeGKro2oX4zIugDNMBzAx2duW8mr3cgyGCVX8M/BG3fdvB1cUfspwmVZNZ0IjZrPU0+69wZxgTcOze1ndXmMDTyzsI4EYeNvD02AeaVmDGY5xe57BtOVz0MQRNoCP0GCCSzjbpUXMpaMgZ/2O/UqmWFd5fMqTFtp57h/UxAHV2xln8rh2Z5JXOThK6qzuA4k8ekRfc+9seCIrPrbTAch6RguXu4sv9eHK2gLmMcOp8j6scwMdsCAwEAAQ==';

Grant the required role

Grant the necessary role to the user.

GRANT ROLE <role_name> TO USER <user_name>;

Example

GRANT ROLE TEST_ROLE TO USER TEST_USER;

Verify the configuration (optional step)

You can run the following commands to verify that the user, pipe, privileges, and data ingestion are configured correctly.

-- Check the user configuration
DESC USER <user_name>;

-- List pipes in the schema
SHOW PIPES IN SCHEMA <database_name>.<schema_name>;

-- Check the privileges granted to the user
SHOW GRANTS TO USER <user_name>;

-- Inspect the data in the target table
SELECT * FROM <database_name>.<schema_name>.<table_name>;

-- Check the number of ingested rows
SELECT COUNT(*) FROM <database_name>.<schema_name>.<table_name>;

-- Optionally, you can sort the query results when inspecting the table data:
SELECT * FROM <database_name>.<schema_name>.<table_name>
ORDER BY <col_name> DESC;

or

SELECT * FROM <database_name>.<schema_name>.<table_name>
ORDER BY <col_name> ASC;

Finding Your Snowflake Account Identifier and URL

To configure a Snowflake output connector, you will need your account identifier and URL.

You can find both directly in the Snowflake web interface.

1. Log in to your Snowflake account.

2. Look at the account settings (bottom-left corner or as shown in the screenshot below). Navigate to View account details. You will see: - Account identifier - a string like ZQPUGTG-YF50693 - Account/Server URL - a hostname like ZQPUGTG-YF50693.snowflakecomputing.com

3. Use these values when creating a new Snowflake output connector: - Account identifier: <your-account-identifier>, like ZQPUGTG-YF50693 - URL: <your-account-url>, like https://ZQPUGTG-YF50693.snowflakecomputing.com (prepend https:// to the hostname)

Troubleshooting

If ingestion does not work as expected, check the following link.

• Troubleshooting Snowpipes