The Snowflake sink connector pulls data from Pulsar topics and persists data to Snowflake. For more information about connectors, see Connector Overview.

This document introduces how to get started with creating a Snowflake sink connector and get it up and running.

Quick start

Prerequisites

The prerequisites for connecting a Snowflake sink connector to external systems include:

1. Prepare a snowflake account

2. Get the account URL from the Admin - Accounts page and click the link. It should be the format like https://<account_identifier>.snowflakecomputing.com.

3. Generate the public key and private key for the authentication. For more details, please check this guide

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

It will generate rsa_key.p8 (the private key) and rsa_key.pub (the public key) locally.

1. Log in and configure the public key.

See Installing SnowSQL to install the SnowSQL.

snowsql -a ${account_identifier} -u ${account_name}

The -a is followed by an account identifier, which is a substring of the account URL before. The -u is followed by your account name. After logging in, set the public key passphrase:

ALTER USER ${account_name} SET RSA_PUBLIC_KEY='MIIBIjA...';

You can get the public key passphrase (MIIBIjA…) by running the following command:

grep -v "\-\-\-" rsa_key.pub | tr -d '\n'

1. Create Snowflake objects and grant permission

Before creating the connector, you need to grant the permissions in Snowflake. Write the following content into a file, e.g. name it with grant.sql. The script creates a user snservice that will be used in the sink config later and associate it with a role snrole that is only used in Snowflake internally. Then it grants the necessary permissions.

CREATE DATABASE st_tuts;
CREATE SCHEMA st_tuts.demo;
CREATE ROLE snrole;
CREATE USER snservice;
GRANT ROLE snrole TO USER snservice;
ALTER USER snservice SET DEFAULT_ROLE = snrole;

GRANT USAGE ON DATABASE st_tuts TO ROLE snrole;
GRANT USAGE ON SCHEMA st_tuts.demo TO ROLE snrole;
GRANT USAGE ON WAREHOUSE compute_wh TO ROLE snrole;
GRANT CREATE TABLE ON SCHEMA st_tuts.demo TO ROLE snrole;
GRANT CREATE STAGE ON SCHEMA st_tuts.demo TO ROLE snrole;
GRANT CREATE PIPE ON SCHEMA st_tuts.demo TO ROLE snrole;
ALTER USER snservice SET RSA_PUBLIC_KEY='MIIBI...';

Then run the following command to execute the SQL script above.

snowsql -a ${account_identifier} -u ${account_name} -f grant.sql

2. Create a connector

The following command shows how to use pulsarctl to create a builtin connector. If you want to create a non-builtin connector, you need to replace --sink-type snowflake with --archive /path/to/pulsar-io-snowflake.nar. You can find the button to download the nar package at the beginning of the document.

pulsarctl sinks create \
  --sink-type snowflake \
  --name snowflake \
  --tenant public \
  --namespace default \
  --inputs "Your topic name" \
  --parallelism 1 \
  --sink-config \
  '{
    "user": "SNSERVICE",
    "host": "https://<account_identifier>.snowflakecomputing.com",
    "schema": "demo",
    "database": "st_tuts",
    "privateKey": "...",
    "warehouse": "compute_wh",
    "topic2table": "input-snowflake:emp_basic"
  }'

The --sink-config is the minimum necessary configuration for starting this connector, and it is a JSON string. You need to substitute the relevant parameters with your own.

You can get the private key passphrase (MIIBIjA…) by running the following command:

grep -v '\-\-\-' rsa_key.p8 | tr -d '\n'

If you want to configure more parameters, see Configuration Properties for reference.

3. Send messages to he topic

   PulsarClient client = PulsarClient.builder()
            .serviceUrl("{{Your Pulsar URL}}")
            .build();

   Producer<String> producer = client.newProducer(Schema.STRING)
     .topic("{{Your topic name}}")
     .create();

   String message = "hello world";
   MessageId msgID = producer.send(message);
   System.out.println("Publish " + message + " and message ID " + msgID);

   producer.flush();
   producer.close();
   client.close();

You can also send the message using the command line:

$ bin/pulsar-client produce pulsar-topic-name --messages "hello world"

4. Check the data on Snowflake table

First, you need to execute the following SQL command to grant the role SNROLE to the user you logged in.

GRANT ROLE SNROLE TO USER ${account_name};

Then, switch the role to SNROLE, under Data - Database - ST_TUTS - DEMO - Tables you will find table PERSISTENT___PUBLIC_DEFAULT_INPUT_SNOWFLAKE_1118738946 is created and records the messages produced above.

Configuration Properties

This table outlines the properties and the descriptions.

Advanced features

This section describes the advanced features of the Snowflake sink connector. For details about how to configure these features, see how to configure.

Delivery guarantees

The Pulsar IO connector framework provides three delivery guarantees: at-most-once, at-least-once, and effectively-once.

Currently, the Snowflake sink connector only supports the at-least-once delivery guarantee semantic.

Table name mapping

The Snowflake sink connector supports automatically creating a table when the table does not exist. You can configure the following options:

autoCreateTable=true

The Snowflake sink connector allows you to specify the mapping relationship between topics and tables. Each topic and its mapped table name should be separated by a colon. The parameter topic2table see example below:

topic2table=topic1:table1,topic2:table2

Metadata Fields

There are two fields in the table: metadata and content. Metadata is ancillary information in content, such as topic, messageId, publishTime, and so on. By default, the following metadata fields of Pulsar will be created as the metadata:

metadataField=__message_id__,__partition__,__topic__,__event_time__

Data format types

The snowflake connector supports converting some Pulsar schemas, as listed in the following table.

Batch progress

To increase write throughput, you can configure the buffer size and latency for the Snowflake sink connector.

bufferCountRecords = 10_000
bufferSizeBytes = 5_000_000
bufferFlushTimeInSeconds = 120