Features

1. Idempotent writes: The default inser.mode is INSERT. If it is configured as UPSERT, the connector will use ascending semantics instead of simple insertion statements. Uppsert's semantics relates to atomic adding a new row or updating the existing row if there is a main violation of the key restriction, providing idempotence.
   
2. Schemes: The connector supports the Avro input format for Key and Value. Scheme logging must be able to use a format based on schema logging.
   
3. Self-creation of table and column: auto.create and self-evolution are compatible. Missing tables or columns can be created automatically. Table names are created based on the name of the topics.
4. Raw data: Connector supports sinking raw data into Snowflake when inser.mode is INSERT and pk.mode is none.
   

Prerequisites

1. Confluent platform with support for Confluent custom connector.
2. Confluent Schema Registry running.
3. Access to a snowflake data store.
4. Valid snowflake credentials.
5. Onibex connector installed on Confluent Platform.
   

---

Quick Start Guide

1. Installation:

Contact Onibex for more details on installing this component.

2. Select the connector to configure it.

3. Choose the theme you want to insert as a table in Snowflake.

4. Specify the class of connector to manage integration with Snowflake:

1. com.onibex.connect.datalake.jdbc.OnibexSnowflakeSinkConnector

5. Connector configuration details:

• First name: The name of the connector.

• Task Max: Maximum number of tasks performed by the connector.

• Key converter class: Key converter format.
  

• Value converter class: Value converter format.
  • Key (Avro): io.confluent.connect.avro.AvroConverter
  • Value (Avro): io.confluent.connect.avro.AvroConverter
  • Key (JSON): io.confluent.connect.json.JsonSchemaConverter
  • Value (JSON): io.confluent.connect.json.JsonSchemaConverter
  • Key (String): apache.kafka.connect.storage.StringConverter
  • Value (cooling): apache.kafka.connect.storage.StringConverter

• Transformation:Transformations applied to data.
  • ExtractTimestamp
  • InsertTimezone
    

6. JDBC configuration:

• Jdbc url: The URL to connect to the Snowflake database, specifying the database, warehouse and outline.
  Example:
  

8. Table and schema settings:

• Table name format: Specify the format of appointment of tables.
• Primary key mode:Define how the main key is extracted. Options: record_key, none, ,record_value.
• Primary key fields: Specify the main key to the topic.
• Create a table: Create tables automatically if they do not exist. Options: True or False.
• Automatic evolution: Automatically update table outlines if changes occur. Options: True or False.
  

9. Configuring the schema record:

• Key converter schema log URL: Confluent schema log URLs.
  Example:

1. http://<ip_schema_registry>:<port>

• Value converter scheme record URL: URL of the Confluent Schemes Register for Values.
  Example:

1. http://<ip_schema_registry>:<port>

1. Transformations:
   

1. ExtractTimestamp.type:Add a timestamp to the records.
   Example: org.apache.kafka.connect.transforms.InsertField$value
2. transformes.ExtractTimestamp.timestamp.field:Field where the timestamp of the event will be inserted.
   Example:timestamp
3. transforms.InsertTimezone.type:Adds a time zone field to the records.
   Example: org.apache.kafka.connect.transforms.InsertField$value
4. transforms.InsertTimezone.static.field:Static field for time zone.
   Example: timezone
5. transforms.InsertTimezone.static.value:Time zone field value.
   Example:America/Mexico_City

---

Limitations

1. Review the limitations and capabilities of the Snowflake JDBC controller.

1. The snowflake user must be without MFA authentication.

1. Column self-creation does not support Generated all communications expression. The default value for overriding columns is Null.
   

---

Data supported for the connector

Length	DataType
Int8	Smallint
Int16	Smallint
Int32	Integrator
Int64	Bigint
Float32	Float
Float64	Double
Boolean	Boolean
Narrow	Primary key ->(English).64) Teld -> varchar
Bytes	Pornario
Null	Null

---

Configuration properties

Common

Name	Description	Values
name	Name of connector	<name_connector>
connector.class	Specifies the kind of connector that integration will handle integration with Snowflake	com.onibex.connect.datalake.jdbc.OnibexSnowflakeSinkConnector
tasks.max	Define the maximum number of tasks that the connector will run	Positive integer value > 1
topics	List of topics to be consumed by this connector	<Topic_Name>

Connection

Name	Description	Values
connection.url	JDBC URL to connect to the Snowflake database, specifying the database, warehouse and outline	dbc:snowflake://<Account/Server URL>:443/?db=<snowflake_database>&warehouse<snowflake_wharehouse>=&schema=<snowflake_schema>
connection.user	Snowflake user	<snowflake_user>
connection.password	Snowflake Password	<Password>
connection.privateKey	Private key used for snowflake authentication	<PrivateKey>
connection.privateKeyPassphrase	Password phrase for private key used in snowflake authentication.	<PrivateKeyPassphrase>

Transaction

Support for developments in the scheme

Configuration example with schema registry and password  (Confluent Cloud)

 {

"auto.create": "true",

  "auto.evolve": "true",

  "confluent.custom.schema.registry.auto": "true",

  "connection.maximumPoolSize": "1",

  "connection.url":  "jdbc:snowflake://<Server_Hostname>:443/?db=<Database_Name>&warehouse=<COMPUTE_WH>&schema=<Schema_Name>",

  "connection.user": "<snowflake_user>",

 "connection.password": "<snowflake_password>",

  "delete.enabled": "true",

  "insert.mode": "INSERT",

  "key.converter": "io.confluent.connect.avro.AvroConverter",

  "pk.mode": "record_key",

  "table.name.format": "${topic}",

  "topics": "<topic_name>",

  "value.converter": "io.confluent.connect.avro.AvroConverter"

}

Configuration example with schema registry and private key  (Confluent Cloud)

 {

"auto.create": "true",

  "auto.evolve": "true",

  "confluent.custom.schema.registry.auto": "true",

  "connection.authentication": "SNOWFLAKE_JWT",

  "connection.maximumPoolSize": "1",

  "connection.privateKey": "<private_key>",

  "connection.url":  "jdbc:snowflake://<Server_Hostname>:443/?db=<Database_Name>&warehouse=<COMPUTE_WH>&schema=<Schema_Name>",

  "connection.user": "<snowflake_user>",

  "delete.enabled": "true",

  "insert.mode": "INSERT",

  "key.converter": "io.confluent.connect.avro.AvroConverter",

  "pk.mode": "record_key",

  "table.name.format": "${topic}",

  "topics": "<topic_name>",

  "value.converter": "io.confluent.connect.avro.AvroConverter"

}

Configuration example with schema registry, private key and passphrase (Confluent Cloud)

Configuration guidelines

1. Configure the connector to map and transform these fields as specified to match the snowflake table structure.
2. Make sure that the snowflake table has the same column names and data types as outlined above to prevent schema mismatches or type errors.
3. Verify the data structuring on the Confluent platform so that each topic message is properly divided into 'heading' , 'key', ,seal of time', and 'value' to ingestion in snowflake.
   

Examples

Example of configuration with string values and password  without schema registry

Example of configuration with string values and private key without schema registry

Example of configuration with string values, private key and password phrase without schema registry