Installing and Configuring the Kafka Connector

The Kafka connector is provided as a JAR (Java executable) file.

Snowflake provides two versions of the connector:

  • A version for the Confluent package version of Kafka.

  • A version for the open source software (OSS) Apache Kafka package.

The instructions in this topic specify which steps apply only to either version of the connector.

In this Topic:

Configuring Access Control for Snowflake Objects

Required Privileges

Creating and managing Snowflake objects used by the Kafka connector requires a role with the following minimum privileges:

Object

Privilege

Notes

Database

USAGE

Schema

USAGE . CREATE TABLE . CREATE STAGE . CREATE PIPE

After the schema-level objects have been created, the CREATE object privileges can be revoked.

Table

INSERT . SELECT

Only required when using the Kafka connector to ingest data into an existing table. . If the connector creates a new target table for records from the Kafka topic, the default role for the user specified in the Kafka configuration file becomes the table owner (i.e. has the OWNERSHIP privilege on the table).

Stage

READ . WRITE

Only required when using the Kafka connector to stage data files from Kafka to an existing internal stage (not recommended). . If the connector creates a new stage to temporarily store data files consumed from the Kafka topic, the default role for the user specified in the Kafka configuration file becomes the stage owner (i.e. has the OWNERSHIP privilege on the stage).

Snowflake recommends that you create a separate user (using CREATE USER) and role (using CREATE ROLE) for each Kafka instance so that the access privileges can be individually revoked if needed. The role should be assigned as the default role for the user.

Creating a Role to Use the Kafka Connector

The following script creates a custom role for use by the Kafka connector (e.g. KAFKA_CONNECTOR_ROLE_1). Any role that can grant privileges (e.g. SECURITYADMIN or any role with the MANAGE GRANTS privilege) can grant this custom role to any user to allow the Kafka connector to create the required Snowflake objects and insert data into tables. The script references a specific existing database and schema (kafka_db.kafka_schema) and user (kafka_connector_user_1):

-- Use a role that can create and manage roles and privileges:
USE ROLE securityadmin;

-- Create a Snowflake role with the privileges to work with the connector
CREATE ROLE kafka_connector_role_1;

-- Grant privileges on the database:
GRANT USAGE ON DATABASE kafka_db TO ROLE kafka_connector_role_1;

-- Grant privileges on the schema:
GRANT USAGE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
GRANT CREATE TABLE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
GRANT CREATE STAGE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
GRANT CREATE PIPE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;

-- Only required if the Kafka connector will load data into an existing table:
GRANT SELECT, INSERT ON TABLE existing_table1 TO ROLE kafka_connector_role_1;

-- Only required if the Kafka connector will stage data files in an existing internal stage: (not recommended):
GRANT READ, WRITE ON STAGE existing_stage1 TO ROLE kafka_connector_role_1;

-- Grant the custom role to an existing user:
GRANT ROLE kafka_connector_role_1 TO USER kafka_connector_user_1;

For more information on creating custom roles and role hierarchies, see Configuring Access Control.

Installation Prerequisites

  • The Kafka connector is built for use with the Kafka Connect API 2.0.0. We highly recommended using a Kafka Connect API version between 2.0.0 and 2.3.0 (inclusive). Earlier versions are

    not compatible with the connector, and newer versions have not been tested.

  • If you use Avro format for ingesting data:

  • Configure Kafka with the desired data retention time and/or storage limit.

  • Install and configure the Kafka Connect cluster.

    Each Kafka Connect cluster node should include enough RAM for the Kafka connector. The minimum recommended amount is 5 MB per Kafka partition. This is in addition to the RAM required for any other work that Kafka Connect is doing.

  • We strongly recommend running your Kafka Connect instance in the same cloud provider region as your Snowflake account. This is not strictly required, but typically improves throughput.

Installing the Connector

This section provides instructions for installing and configuring the Kafka connector for Confluent.

Installing the Connector for Confluent

Download the Kafka Connector Files

Download the Kafka connector JAR file from either of the following locations:

Confluent Hub

https://www.confluent.io/hub/

The package includes all dependencies required to use either an encrypted or unencrypted private key for key pair authentication. For more information, see Using Key Pair Authentication (in this topic).

Maven Central Repository

https://mvnrepository.com/artifact/com.snowflake

The JAR file does not require any additional dependencies to use an unencrypted private key for key pair authentication. To use an encrypted private key, download the Bouncy Castle cryptography library (a JAR file). Snowflake uses Bouncy Castle to decrypt encrypted RSA private keys used to log in:

Download these files to the same local folder as the Kafka connector JAR file.

The source code for the connector is available at https://github.com/snowflakedb/snowflake-kafka-connector.

Install the Kafka Connector

Install the Kafka connector using the instructions provided for installing other connectors:

Installing the Connector for Open Source Apache Kafka

This section provides instructions for installing and configuring the Kafka connector for open source Apache Kafka.

Install Apache Kafka

  1. Download the Kafka package from its official website: https://kafka.apache.org/downloads

  2. In a terminal window, change to the directory where you downloaded the package file.

  3. Execute the following command to decompress the kafka_<scala_version>-<kafka_version>.tgz file:

    tar xzvf kafka_<scala_version>-<kafka_version>.tgz
    

Install the JDK

Install and configure the Java Development Kit (JDK). Snowflake tests with the Standard Edition (SE) of the JDK. The Enterprise Edition (EE) is expected to be compatible but has not been tested.

If you have already completed this step, you can skip this section.

  1. Download the JDK from https://www.oracle.com/technetwork/java/javase/downloads/index.html.

  2. Install or decompress the JDK.

  3. Following the instructions for your operating system, set the environment variable JAVA_HOME to point to the directory containing the JDK.

Download the Kafka Connector JAR Files

  1. Download the Kafka connector JAR file from the Maven Central Repository:

    https://mvnrepository.com/artifact/com.snowflake

  2. The JAR file does not require any additional dependencies to use an unencrypted private key for key pair authentication. To use an encrypted private key, download the Bouncy Castle cryptography library (a JAR file). Snowflake uses Bouncy Castle to decrypt encrypted RSA private keys used to log in:

  3. If your Kafka data is streamed in Apache Avro format, then download the Avro JAR file:

    https://mvnrepository.com/artifact/org.apache.avro/avro

The source code for the connector is available at https://github.com/snowflakedb/snowflake-kafka-connector.

Install the Kafka Connector

Copy the JAR files you downloaded in Download the Kafka Connector JAR Files to the <kafka_dir>/libs folder.

Configuring the Kafka Connector

The connector is configured by creating a file that specifies parameters such as the Snowflake login credentials, topic name(s), Snowflake table name(s), etc.

Important

The Kafka Connect framework broadcasts the configuration settings for the Kafka connector from the master node to worker nodes. The configuration settings include sensitive information (specifically, the Snowflake username and private key). Make sure to secure the communication channel between Kafka Connect nodes. For instructions, see the documentation for your Apache Kafka software.

Each configuration file specifies the topics and corresponding tables for one database and one schema in that database. Note that a connector can ingest messages from any number of topics, but the corresponding tables must all be stored in a single database and schema.

This section provides instructions for both the distributed and standalone modes.

For descriptions of the configuration fields, see Kafka Configuration Properties.

Important

Because the configuration file typically contains security related information, such as the private key, set read/write privileges appropriately on the file to limit access.

In addition, consider storing the configuration file in a secure external location or a key management service. For more information, see Externalizing Secrets (in this topic).

Distributed Mode

Create a Kafka configuration file, e.g. <kafka_dir>/config/connect-distributed.properties. Populate the file with all connector configuration information.

Sample configuration file

{
  "name":"XYZCompanySensorData",
  "Config":{
    "connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "tasks.max":"8",
    "topics":"topic1,topic2",
    "snowflake.topic2table.map": "topic1:table1,topic2:table2",
    "buffer.count.records":"10000",
    "buffer.flush.time":"60",
    "buffer.size.bytes":"5000000",
    "snowflake.url.name":"myaccount.us-west-2.snowflakecomputing.com:443",
    "snowflake.user.name":"jane.smith",
    "snowflake.private.key":"xyz123",
    "snowflake.private.key.passphrase":"jkladu098jfd089adsq4r",
    "snowflake.database.name":"mydb",
    "snowflake.schema.name":"myschema",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "value.converter":"com.snowflake.kafka.connector.records.SnowflakeAvroConverter",
    "value.converter.schema.registry.url":"http://localhost:8081",
    "value.converter.basic.auth.credentials.source":"USER_INFO",
    "value.converter.basic.auth.user.info":"jane.smith:MyStrongPassword"
  }
}

Standalone Mode

Create a configuration file, e.g. <kafka_dir>/config/connect-standalone.properties. Populate the file with all connector configuration information.

Sample configuration file

connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
tasks.max=8
topics=topic1,topic2
snowflake.topic2table.map= topic1:table1,topic2:table2
buffer.count.records=10000
buffer.flush.time=60
buffer.size.bytes=5000000
snowflake.url.name=myaccount.us-west-2.snowflakecomputing.com:443
snowflake.user.name=jane.smith
snowflake.private.key=xyz123
snowflake.private.key.passphrase=jkladu098jfd089adsq4r
snowflake.database.name=mydb
snowflake.schema.name=myschema
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=com.snowflake.kafka.connector.records.SnowflakeAvroConverter
value.converter.schema.registry.url=http://localhost:8081
value.converter.basic.auth.credentials.source=USER_INFO
value.converter.basic.auth.user.info=jane.smith:MyStrongPassword

Kafka Configuration Properties

The following properties can be set in the Kafka configuration file:

Required Properties

name

Application name. This must be unique across all Kafka connectors used by the customer. This name name must be a valid Snowflake unquoted identifier. For information about valid identifiers, see Identifier Requirements.

connector.class

com.snowflake.kafka.connector.SnowflakeSinkConnector

topics

Comma-separated list of topics. By default, Snowflake assumes that the table name is the same as the topic name. If the table name is not the same as the topic name, then use the optional topic2table.map parameter (below) to specify the mapping from topic name to table name. This table name must be a valid Snowflake unquoted identifier. For information about valid table names, see Identifier Requirements.

Note

Either topics or topics.regex is required; not both.

topics.regex

This is a regular expression (“regex”) that specifies the topics that contain the messages to load into Snowflake tables. The connector loads data from any topic name that matches the regex. The regex must follow the rules for Java regular expressions (i.e. be compatible with java.util.regex.Pattern). The configuration file should contain either topics or topics.regex, not both.

snowflake.url.name

The URL for accessing your Snowflake account, in the form of https://<account_name>.<region_id>.snowflakecomputing.com:443. Note that the https:// and port number are optional. The region ID is not used if your account is in the AWS US West region and your are not using AWS PrivateLink. For more information about Snowflake account names and region names, see Supported Regions.

snowflake.user.name

User login name for the Snowflake account.

snowflake.private.key

The private key to authenticate the user. Include only the key, not the header or footer. If the key is split across multiple lines, remove the line breaks. You can provide either an unencrypted key, or you can provide an encrypted key and provide the snowflake.private.key.passphrase parameter to enable Snowflake to decrypt the key. Use this parameter if and only if the snowflake.private.key parameter value is encrypted. This decrypts private keys that were encrypted according to the instructions in Using Key Pair Authentication (in this topic).

Note

Also see snowflake.private.key.passphrase in Optional Properties (in this topic).

snowflake.database.name

The name of the database that contains the table to insert rows into.

snowflake.schema.name

The name of the schema that contains the table to insert rows into.

key.converter

This is the Kafka record’s key converter (e.g. "org.apache.kafka.connect.storage.StringConverter"). This is not used by the Kafka connector, but is required by the Kafka Connect Platform.

value.converter

If the records are formatted in JSON, this should be "com.snowflake.kafka.connector.records.SnowflakeJsonConverter".

If the records are formatted in Avro and use Kafka’s Schema Registry Service, this should be "com.snowflake.kafka.connector.records.SnowflakeAvroConverter".

If the records are formatted in Avro and contain the schema (and therefore do not need Kafka’s Schema Registry Service), this should be "com.snowflake.kafka.connector.records.SnowflakeAvroConverterWithoutSchemaRegistry".

Optional Properties

snowflake.private.key.passphrase

If the value of this parameter is not empty, the Kafka uses this phrase to try to decrypt the private key.

tasks.max

Number of tasks, usually the same as the number of CPU cores across the worker nodes in the Kafka Connect cluster. This number can be set lower or higher; however, Snowflake does not recommend setting it higher.

snowflake.topic2table.map

This optional parameter allows a user to specify which topics should be mapped to which tables. Each topic and its table name should be separated by a colon (see example below). This table name must be a valid Snowflake unquoted identifier. For information about valid table names, see Identifier Requirements.

buffer.count.records

Number of records buffered in memory per Kafka partition before ingesting to Snowflake. The default value is 10000 records.

buffer.flush.time

Number of seconds between buffer flushes, where the flush is from the Kafka’s memory cache to the internal stage. The default value is 30 seconds.

buffer.size.bytes

Cumulative size in bytes of records buffered in memory per Kafka partition before ingesting to Snowflake. The default value for this is 5 MB.

value.converter.schema.registry.url

If the format is Avro and you are using a Schema Registry Service, this should be the URL of the Schema Registry Service. Otherwise this field should be empty.

jvm.proxy.host

To enable the Snowflake Kafka Connector to access Snowflake through a proxy server, set this parameter to specify the host of that proxy server.

jvm.proxy.port

To enable the Snowflake Kafka Connector to access Snowflake through a proxy server, set this parameter to specify the port of that proxy server.

value.converter.basic.auth.credentials.source

If you are using the Avro data format and require secure access to the Kafka schema registry, set this parameter to the string “USER_INFO”, and set the value.converter.basic.auth.user.info parameter described below. Otherwise, omit this parameter.

value.converter.basic.auth.user.info

If you are using the Avro data format and require secure access to the Kafka schema registry, set this parameter to the string “<user_ID>:<password>”, and set the value.converter.basic.auth.credentials.source parameter described above. Otherwise, omit this parameter.

Using Key Pair Authentication

The Kafka connector relies on key pair authentication rather than the typical username/password authentication. This authentication method requires a 2048-bit (minimum) RSA key pair. Generate the public-private key pair using OpenSSL. The public key is assigned to the Snowflake user defined in the configuration file.

To configure the public/private key pair:

  1. From the command line in a terminal window, generate a private key.

    You can generate either an encrypted version or unencrypted version of the private key.

    Note

    The Kafka connector supports encryption algorithms that are validated to meet the Federal Information Processing Standard (140-2) (FIPS 140-2) requirements. For information about FIPS 140-2, see https://csrc.nist.gov/publications/detail/fips/140/2/final

    To generate an unencrypted version, use the following command:

    $ openssl genrsa -out rsa_key.pem 2048
    

    To generate an encrypted version, use the following command:

    $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 <algorithm> -inform PEM -out rsa_key.p8
    

    Where <algorithm> is a FIPS 140-2 compliant encryption algorithm.

    For example, to specify AES 256 as the encryption algorithm:

    $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 aes256 -inform PEM -out rsa_key.p8
    

    If you generate an encrypted version of the private key, record the passphrase. Later, you will specify the passphrase in the snowflake.private.key.passphrase property in the Kafka configuration file.

    Sample PEM private key

    -----BEGIN ENCRYPTED PRIVATE KEY-----
    MIIE6TAbBgkqhkiG9w0BBQMwDgQILYPyCppzOwECAggABIIEyLiGSpeeGSe3xHP1
    wHLjfCYycUPennlX2bd8yX8xOxGSGfvB+99+PmSlex0FmY9ov1J8H1H9Y3lMWXbL
    ...
    -----END ENCRYPTED PRIVATE KEY-----
    
  2. From the command line, generate the public key by referencing the private key:

    Assuming the private key is encrypted and contained in the file named rsa_key.p8, use the following command:

    $ openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
    

    Sample PEM public key

    -----BEGIN PUBLIC KEY-----
    MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy+Fw2qv4Roud3l6tjPH4
    zxybHjmZ5rhtCz9jppCV8UTWvEXxa88IGRIHbJ/PwKW/mR8LXdfI7l/9vCMXX4mk
    ...
    -----END PUBLIC KEY-----
    
  3. Copy the public and private key files to a local directory for storage. Record the path to the files. Note that the private key is stored using the PKCS#8 (Public Key Cryptography Standards) format and is encrypted using the passphrase you specified in the previous step; however, the file should still be protected from unauthorized access using the file permission mechanism provided by your operating system. It is your responsibility to secure the file when it is not being used.

  4. Log into Snowflake. Assign the public key to the Snowflake user using ALTER USER. For example:

    ALTER USER jsmith SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';
    

    Note

    • Only security administrators (i.e. users with the SECURITYADMIN role) or higher can alter a user.

    • Exclude the public key header and footer in the SQL statement.

    Verify the user’s public key fingerprint using DESCRIBE USER:

    DESC USER jsmith;
    +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
    | property                      | value                                               | default | description                                                                   |
    |-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------|
    | NAME                          | JSMITH                                              | null    | Name                                                                          |
    ...
    ...
    | RSA_PUBLIC_KEY_FP             | SHA256:nvnONUsfiuycCLMXIEWG4eTp4FjhVUZQUQbNpbSHXiA= | null    | Fingerprint of user's RSA public key.                                         |
    | RSA_PUBLIC_KEY_2_FP           | null                                                | null    | Fingerprint of user's second RSA public key.                                  |
    +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
    

    Note

    The RSA_PUBLIC_KEY_2_FP property is described in Key Rotation (in this topic).

  5. Copy and paste the entire private key into the snowflake.private.key field in the configuration file. Save the file.

Externalizing Secrets

Snowflake strongly recommends externalizing secrets such as the private key and storing them in an encrypted form or in a key management service such as AWS Key Management Service (KMS), Microsoft Azure Key Vault, or HashiCorp Vault. This can be accomplished by using a ConfigProvider implementation on your Kafka Connect cluster.

For more information, see the Confluent description of this service: https://docs.confluent.io/current/connect/security.html#externalizing-secrets.

Key Rotation

Snowflake supports multiple active keys to allow for uninterrupted rotation. Rotate and replace your public and private keys based on the expiration schedule you follow internally.

Currently, you can use the RSA_PUBLIC_KEY and RSA_PUBLIC_KEY_2 parameters for ALTER USER to associate up to 2 public keys with a single user.

To rotate your keys:

  1. Complete the steps in Using Key Pair Authentication to:

    • Generate a new private and public key set.

    • Assign the public key to the user. Set the public key value to either RSA_PUBLIC_KEY or RSA_PUBLIC_KEY_2 (whichever key value is not currently in use). For example:

      alter user jsmith set rsa_public_key_2='JERUEHtcve...';
      
  2. Update the code to connect to Snowflake. Specify the new private key.

    Snowflake verifies the correct active public key for authentication based on the private key submitted with your connection information.

  3. Remove the old public key from the user profile. For example:

    alter user jsmith unset rsa_public_key;
    

Starting Kafka

Start Kafka using the instructions provided in the third-party Confluent or Apache Kafka documentation.

Starting the Kafka Connector

You can start the Kafka connector in either distributed mode or standalone mode. Instructions for each are shown below:

Distributed Mode

In a terminal window, execute the following command:

curl -X POST -H "Content-Type: application/json" --data @<config_file>.json http://localhost:8083/connectors

Standalone Mode

In a terminal window, execute the following command:

<kafka_dir>/bin/connect-standalone.sh <kafka_dir>/<path>/connect-standalone.properties <kafka_dir>/config/connect-standalone.properties

Where:

Testing and Using the Kafka Connector

We recommend testing the Kafka connector with a small amount of data before using the connector in a production system. The process for testing is the same as the process for using the connector normally:

  1. Verify that Kafka and Kafka Connect are running.

  2. Verify that you have created the appropriate Kafka topic.

  3. Create (or use an existing) message publisher. Make sure that the messages published to the topic have the right format (JSON or Avro).

  4. Create a configuration file that specifies the topic to subscribe to, and the Snowflake table to write to. Also see the next step.

  5. (Optional) Create a table into which to write data. This step is optional; if you do not create the table, the Kafka connector creates the table for you. If you do not plan to use the connector to add data to an existing, non-empty table, then we recommend that you let the connector create the table for you to minimize the possibility of a schema mismatch.

  6. Grant the minimum privileges required on the Snowflake objects (database, schema, target table, etc.) to the role that will be used to ingest data.

  7. Publish a sample set of data to the configured Kafka topic.

  8. Wait a few minutes for data to propagate through the system, and then check the Snowflake table to verify that the records were inserted.

Tip

Consider verifying your network connection to Snowflake using SnowCD before loading data to Snowflake in your test and production environments.