Skip to main content

Create and configure a Snowflake sink connector for Apache Kafka®

The Apache Kafka Connect® Snowflake sink connector moves data from Aiven for Apache Kafka® topics to a Snowflake database. Configure access in Snowflake before creating the connector in Aiven.

Aiven provides two separate connectors:

  • Snowflake Sink: version 3.x, which uses the com.snowflake.kafka.connector.SnowflakeSinkConnector class. Snowflake refers to this version as the classic connector. It's stable and supported, but Snowflake plans to deprecate it.
  • Snowflake v4 Sink: version 4.2.1, which uses the com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector class and the Snowpipe Streaming high-performance architecture.
Note

Snowflake v4 Sink is in early availability.

Aiven doesn't automatically upgrade existing Snowflake Sink connectors to Snowflake v4 Sink. Version 4.x isn't backward compatible with 3.x. To move an existing connector to 4.x, review Migrate from 3.x to 4.x before creating the replacement.

Prerequisites​

  • An Aiven for Apache Kafka service with Apache Kafka Connect enabled, or a dedicated Kafka Connect cluster.
  • Access to a Snowflake account with permissions to create users and roles and grant access to the target database and schema.
  • OpenSSL installed locally to generate a key pair.
  • The connection details listed in the following section.

Connection details​

Collect the following values:

  • SNOWFLAKE_URL: your Snowflake account URL, for example, myorg-myaccount.snowflakecomputing.com. For details, see Snowflake account identifiers.
  • SNOWFLAKE_PRIVATE_KEY: the private key for the user
  • SNOWFLAKE_PRIVATE_KEY_PASSPHRASE: Optional: the private key passphrase. Required only for encrypted private keys.
  • SNOWFLAKE_DATABASE: the target database
  • SNOWFLAKE_SCHEMA: the target schema
  • TOPIC_LIST: a comma-separated list of Kafka topics

For Avro messages, also collect the following values:

  • APACHE_KAFKA_HOST: the Aiven for Apache Kafka service host
  • SCHEMA_REGISTRY_PORT: the Schema Registry port
  • SCHEMA_REGISTRY_USER: the Schema Registry user
  • SCHEMA_REGISTRY_PASSWORD: the Schema Registry password
Tip

To find your account URL in Snowsight, open the account selector, select View account details, and copy Account/Server URL.

For more information about configuration options, see the Snowflake Kafka Connector documentation.

Configure Snowflake​

Set up key-pair authentication, create a Snowflake user and role, and grant access to the target database and schema.

Generate a key pair​

Use OpenSSL to generate a 2048-bit RSA key pair:

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
  • rsa_key.p8 contains the passphrase-protected private key.
  • rsa_key.pub contains the public key.

Use the private key as the SNOWFLAKE_PRIVATE_KEY value in the connector configuration. For the key format the connector accepts, see Snowflake key-pair authentication. You only need the passphrase for an encrypted private key.

Create a user​

  1. Open Worksheets in Snowflake.

  2. Use a role with privileges to create users, such as SECURITYADMIN or ACCOUNTADMIN.

  3. Create a user for the connector:

    CREATE USER aiven;

  4. Copy the contents of rsa_key.pub, excluding the -----BEGIN PUBLIC KEY----- and -----END PUBLIC KEY----- lines. Remove line breaks.

  5. Set the user's public key:

    ALTER USER aiven SET RSA_PUBLIC_KEY='PASTE_PUBLIC_KEY_HERE';

Create a role and assign it to the user​

  1. Create a role:

    CREATE ROLE aiven_snowflake_sink_connector_role;

  2. Grant the role to the connector user:

    GRANT ROLE aiven_snowflake_sink_connector_role TO USER aiven;

  3. Set the user's default role:

    ALTER USER aiven SET DEFAULT_ROLE = aiven_snowflake_sink_connector_role;

Grant privileges on the target database and schema​

Replace TESTDATABASE and TESTSCHEMA with your Snowflake database and schema names. Use a role authorized to grant the privileges.

Version 3.x requires privileges for tables, stages, and pipes:

GRANT USAGE ON DATABASE TESTDATABASE TO ROLE aiven_snowflake_sink_connector_role;
GRANT USAGE ON SCHEMA TESTDATABASE.TESTSCHEMA TO ROLE aiven_snowflake_sink_connector_role;
GRANT CREATE TABLE ON SCHEMA TESTDATABASE.TESTSCHEMA TO ROLE aiven_snowflake_sink_connector_role;
GRANT CREATE STAGE ON SCHEMA TESTDATABASE.TESTSCHEMA TO ROLE aiven_snowflake_sink_connector_role;
GRANT CREATE PIPE ON SCHEMA TESTDATABASE.TESTSCHEMA TO ROLE aiven_snowflake_sink_connector_role;

Create a Snowflake sink connector configuration​

Create a JSON file such as snowflake_sink.json. Choose the example for the connector you're creating, then replace the placeholders with your connection details.

The following examples read Avro messages using Aiven's Karapace Schema Registry.

{
  "name": "my-test-snowflake",
  "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
  "topics": "TOPIC_LIST",
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT",
  "key.converter.basic.auth.credentials.source": "USER_INFO",
  "key.converter.schema.registry.basic.auth.user.info": "SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT",
  "value.converter.basic.auth.credentials.source": "USER_INFO",
  "value.converter.schema.registry.basic.auth.user.info": "SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD",
  "snowflake.url.name": "SNOWFLAKE_URL",
  "snowflake.user.name": "aiven",
  "snowflake.private.key": "SNOWFLAKE_PRIVATE_KEY",
  "snowflake.private.key.passphrase": "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE",
  "snowflake.database.name": "SNOWFLAKE_DATABASE",
  "snowflake.schema.name": "SNOWFLAKE_SCHEMA"
}

Configuration parameters​

Connector and converter settings:

  • name: Unique connector name
  • connector.class: Connector implementation for the selected version
  • topics: Comma-separated topics to load into Snowflake
  • snowflake.topic2table.map: Maps Kafka topics to Snowflake tables. Specify comma-separated topic:table pairs, for example, book:BOOK,product:PRODUCT. If omitted, the connector derives table names from Kafka topic names.
  • key.converter and value.converter: Converters for Kafka message keys and values. The example uses io.confluent.connect.avro.AvroConverter.

Schema Registry settings for Avro:

  • key.converter.schema.registry.url and value.converter.schema.registry.url: Schema Registry URL in the form https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT
  • key.converter.basic.auth.credentials.source and value.converter.basic.auth.credentials.source: Credentials source for basic authentication, set to USER_INFO
  • key.converter.schema.registry.basic.auth.user.info and value.converter.schema.registry.basic.auth.user.info: Schema Registry credentials in the form SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD

Snowflake connection settings:

  • snowflake.url.name: Snowflake account URL
  • snowflake.user.name: Snowflake user for ingestion
  • snowflake.private.key: Private key used to authenticate the user
  • snowflake.private.key.passphrase: Optional: Passphrase protecting the private key. Required only for encrypted private keys.
  • snowflake.database.name: Target database
  • snowflake.schema.name: Target schema

Additional 4.x settings shown in the example:

  • snowflake.role.name: Role the connector uses to write data. The example uses the role you created in Configure Snowflake.
  • snowflake.validation: Validation mode. Use client_side to validate against schema types before ingestion. The 4.x default is server_side.
  • snowflake.streaming.validate.compatibility.with.classic: Migration-related compatibility checks. Set to false for new deployments, but not automatically for migrations.

For all 4.x settings, see Snowflake connector configuration.

Create a Snowflake sink connector​

  1. In the Aiven Console, click your Aiven for Apache Kafka or Aiven for Apache Kafka Connect service.

  2. Click Manage stream > Connectors.

  3. Click Create connector if Apache Kafka Connect is already enabled on the service. If not, click Enable connector on this service.

    To enable connectors:

    1. In the sidebar, click Service settings.
    2. In the Service management section, click Actions > Enable Kafka Connect.

  4. In the list of sink connectors, find Snowflake Sink for 3.x or Snowflake v4 Sink for 4.x, then click Get started.

  5. On the connector page, go to the Common tab.

  6. Locate Connector configuration and click Edit.

  7. Paste the configuration for the selected version from your snowflake_sink.json file. Replace all placeholders with actual values.

  8. Click Apply.

  9. Click Create connector.

  10. Verify the connector status on the Manage stream > Connectors page.

  11. Confirm that data from the Apache Kafka topics appears in the target Snowflake database.

Migrate from 3.x to 4.x​

Snowflake connector 4.x isn't backward compatible with 3.x, and Aiven doesn't upgrade existing 3.x connectors automatically. To move to 4.x, create a 4.x connector as described in this section.

Before you migrate:

  1. Review the Snowflake migration guide for breaking changes, configuration requirements, and migration steps.
  2. Check the snowflake.ingestion.method setting of your existing 3.x connector. It is either SNOWPIPE, which is the default, or SNOWPIPE_STREAMING. The offset migration procedure depends on this setting.
  3. Prepare a new configuration that uses the com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector class. Set snowflake.streaming.classic.offset.migration to skip unless your 3.x connector uses SNOWPIPE_STREAMING. Migrating from SNOWPIPE_STREAMING can require special attention depending on your use case. See migrating from Snowpipe Streaming mode.
  4. Test the migration procedure on a non-production service before you migrate production workloads.

To preserve offsets, Snowflake's migration procedure requires the same connector name for 3.x and 4.x. The 4.x connector then inherits the Kafka consumer group offsets and resumes where 3.x stopped. Aiven doesn't allow two connectors with the same name, so delete the 3.x connector before you create the 4.x connector. Follow Snowflake's instructions for your ingestion method, such as migrating from Snowpipe mode.

To migrate:

  1. Shut down the 3.x connector gracefully, then delete it. A forceful shutdown might prevent the offsets from advancing and cause duplicate records.
  2. For Snowpipe mode, wait until Snowflake finishes loading all staged data. Otherwise, data might arrive out of order.
  3. Deploy the 4.x connector with the same name on the same Aiven for Apache Kafka service as the 3.x connector. That service holds the consumer group offsets, so a connector on a different service doesn't resume from the 3.x offsets.
Important

Ingestion stops from the time you delete the 3.x connector until the 4.x connector starts. Start the 4.x connector before the consumer group offsets expire. The offsets.retention.minutes default is 7 days, and 3 days for free and developer plans.

Downgrade from 4.x to 3.x​

To go back to 3.x:

  1. Stop the 4.x connector.
  2. Create the 3.x connector again with the same name and configuration.
  3. Use the RECORD_METADATA column to find and remove duplicate records.

Expect duplicates after a downgrade, because 3.x and 4.x track offsets differently. For the full procedure and a query to remove duplicates, see Downgrading from v4 to v3.

Examples​

The following examples show how to configure the Snowflake sink connector for Avro and JSON messages. The Avro example covers both versions. The JSON example covers 3.x only, because 4.x uses a different class and can create a different Snowflake table structure by default. Review the 4.x configuration before you adapt the JSON example for 4.x.

Create a Snowflake sink connector for an Avro topic​

This example uses:

  • Connector name: my_snowflake_sink
  • Kafka topic: test
  • Snowflake database: testdb
  • Snowflake schema: testschema
  • Snowflake URL: myorg-myaccount.snowflakecomputing.com
  • Snowflake user: aiven

Replace the Snowflake URL, private key, passphrase, and Schema Registry placeholders with your own values.

{
  "name": "my_snowflake_sink",
  "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT",
  "key.converter.basic.auth.credentials.source": "USER_INFO",
  "key.converter.schema.registry.basic.auth.user.info": "SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "https://APACHE_KAFKA_HOST:SCHEMA_REGISTRY_PORT",
  "value.converter.basic.auth.credentials.source": "USER_INFO",
  "value.converter.schema.registry.basic.auth.user.info": "SCHEMA_REGISTRY_USER:SCHEMA_REGISTRY_PASSWORD",
  "topics": "test",
  "snowflake.url.name": "myorg-myaccount.snowflakecomputing.com",
  "snowflake.user.name": "aiven",
  "snowflake.private.key": "SNOWFLAKE_PRIVATE_KEY",
  "snowflake.private.key.passphrase": "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE",
  "snowflake.database.name": "testdb",
  "snowflake.schema.name": "testschema"
}

Create a Snowflake sink connector 3.x for a topic with a JSON schema​

This example reads JSON messages from the iot_measurements topic. Each message includes an inline schema.

Sample message:

{
  "schema": {
    "type": "struct",
    "fields": [
      { "type": "int64", "optional": false, "field": "iot_id" },
      { "type": "string", "optional": false, "field": "metric" },
      { "type": "int32", "optional": false, "field": "measurement" }
    ]
  },
  "payload": { "iot_id": 1, "metric": "Temperature", "measurement": 14 }
}

Another sample message:

{
  "schema": {
    "type": "struct",
    "fields": [
      { "type": "int64", "optional": false, "field": "iot_id" },
      { "type": "string", "optional": false, "field": "metric" },
      { "type": "int32", "optional": false, "field": "measurement" }
    ]
  },
  "payload": { "iot_id": 2, "metric": "Humidity", "measurement": 60 }
}
Note

Each JSON record includes a full schema, which adds overhead. To avoid repeating the schema, use Avro with Aiven's Karapace Schema Registry.

Create a connector configuration and replace the placeholders with your values:

{
  "name": "my-test-snowflake-1",
  "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "true",
  "topics": "iot_measurements",
  "snowflake.url.name": "SNOWFLAKE_URL",
  "snowflake.user.name": "aiven",
  "snowflake.private.key": "SNOWFLAKE_PRIVATE_KEY",
  "snowflake.private.key.passphrase": "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE",
  "snowflake.database.name": "SNOWFLAKE_DATABASE",
  "snowflake.schema.name": "SNOWFLAKE_SCHEMA"
}
  • topics: Kafka topic the connector reads from and writes to Snowflake.
  • value.converter: Converter for message values. The example uses the Kafka Connect JSON converter.
  • value.converter.schemas.enable: Set to true when each message contains a schema and a payload field. The example sets this explicitly for clarity.

The configuration has no key converter because the example messages have no keys.