Snowflake High Performance connector for Kafka: Install and configure¶

Installing the connector for Confluent¶

Installation du connecteur pour Apache Kafka open source¶

Cette section fournit des instructions pour l’installation et la configuration du connecteur Kafka pour Apache Kafka Open Source.

Mode distribué¶

Créez le fichier de configuration Kafka, par exemple <path>/<config_file>.json. Remplissez le fichier avec toutes les informations de configuration du connecteur. Le fichier doit être au format JSON.

Sample configuration file

{
  "name":"XYZCompanySensorData",
  "config":{
      "connector.class": "com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector",
      "tasks.max": "1",
      "snowflake.topic2table.map": "topic1:table_1,topic2:table_2",
      "snowflake.url.name": "myorganization-myaccount.snowflakecomputing.com:443",
      "snowflake.warehouse.name": "WH",
      "snowflake.private.key": "-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n",
      "snowflake.schema.name": "MY_SCHEMA",
      "snowflake.database.name": "MY_DATABASE",
      "snowflake.role.name": "MY_ROLE",
      "snowflake.user.name": "MY_USER",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "errors.log.enable": "true",
      "topics": "topic1,topic2",
      "value.converter.schemas.enable": "false",
      "errors.tolerance": "none"
      }
}

Mode autonome¶

Créez un fichier de configuration, par exemple <kafka_dir>/config/SF_connect.properties. Remplissez le fichier avec toutes les informations de configuration du connecteur.

Sample configuration file

connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=1
snowflake.topic2table.map=topic1:table_1,topic2:table_2
snowflake.url.name=myorganization-myaccount.snowflakecomputing.com:443
snowflake.warehouse.name=WH
snowflake.private.key=-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n
snowflake.schema.name=MY_SCHEMA
snowflake.database.name=MY_DATABASE
snowflake.role.name=MY_ROLE
snowflake.user.name=MY_USER
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
errors.log.enable=true
topics=topic1,topic2
name=XYZCompanySensorData
value.converter.schemas.enable=false
errors.tolerance=none

Propriétés requises¶

name

Nom de l’application. Cela doit être unique pour tous les connecteurs Kafka utilisés par le client. Ce nom doit être un identificateur non délimité Snowflake et valide. Pour plus d’informations sur les identificateurs valides, voir Exigences relatives à l’identificateur.

connector.class

com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector

topics

Liste de sujets séparés par des virgules. Par défaut, Snowflake suppose que le nom de la table est identique à celui du sujet. Si le nom de la table est différent du nom du sujet, utilisez le paramètre facultatif topic2table.map (ci-dessous) pour spécifier le mappage entre le nom du sujet et le nom de la table. Ce nom de table doit être un identificateur non spécifié Snowflake et valide. Pour plus d’informations sur les noms de table valides, voir Exigences relatives à l’identificateur.

Note

:emph:`` topics ou topics.regex est requis ; pas les deux.

topics.regex

Il s’agit d’une expression régulière (« regex ») qui spécifie les sujets contenant les messages à charger dans les tables Snowflake. Le connecteur charge les données de tout nom de sujet correspondant à l’expression régulière. L’expression régulière doit respecter les règles applicables aux expressions régulières Java (c.-à-d. être compatible avec java.util.regex.Pattern). Le fichier de configuration doit contenir topics ou topics.regex, pas les deux.

snowflake.url.name

L’URL pour accéder à votre compte Snowflake. Cette URL doit inclure votre identificateur de compte. Notez que le protocole (https://) et le numéro de port sont facultatifs.

snowflake.user.name

Nom de connexion de l’utilisateur pour le compte Snowflake.

snowflake.role.name

Nom du rôle que le connecteur utilisera pour insérer des données dans la table.

snowflake.private.key

Clé privée pour authentifier l’utilisateur. Incluez uniquement la clé, pas l’en-tête ni le pied de page. Si la clé est divisée sur plusieurs lignes, supprimez les sauts de ligne. Vous pouvez fournir une clé non chiffrée ou une clé chiffrée et fournir le paramètre snowflake.private.key.passphrase pour permettre à Snowflake de déchiffrer la clé. Utilisez ce paramètre si et seulement si la valeur du paramètre snowflake.private.key est chiffrée. Celui-ci déchiffre les clés privées qui ont été chiffrées conformément aux instructions fournies dans la section Authentification par paire de clés et rotation de paires de clés.

Note

Consultez également snowflake.private.key.passphrase dans les Propriétés facultatives.

snowflake.database.name

Nom de la base de données contenant la table dans laquelle insérer des lignes.

snowflake.schema.name

Nom du schéma contenant la table dans laquelle insérer des lignes.

header.converter

Obligatoire uniquement si les enregistrements sont formatés en Avro et incluent un en-tête. La valeur par défaut est "org.apache.kafka.connect.storage.StringConverter".

key.converter

Il s’agit du convertisseur de clé de l’enregistrement Kafka, (par exemple "org.apache.kafka.connect.storage.StringConverter"). Ce connecteur n’est pas utilisé par le connecteur Kafka, mais il est requis par la plate-forme Kafka Connect.

Voir Limitations du connecteur Kafka pour les limitations actuelles.

value.converter

Le connecteur prend en charge les convertisseurs communautaires Kafka standard. Choisissez le convertisseur approprié en fonction de votre format de données :

• Pour les enregistrements JSON : "org.apache.kafka.connect.json.JsonConverter"

• Pour les enregistrements Avro avec le registre de schéma : "io.confluent.connect.avro.AvroConverter"

Voir Limitations du connecteur Kafka pour les limitations actuelles.

Propriétés facultatives¶

snowflake.private.key.passphrase

Si la valeur de ce paramètre n’est pas vide, le connecteur utilise cette expression pour tenter de déchiffrer la clé privée.

tasks.max

Nombre de tâches, généralement égal au nombre de cœurs CPU sur les nœuds de travail du cluster Kafka Connect. Pour obtenir les meilleures performances, Snowflake recommande de définir un nombre de tâches égal au nombre total de partitions Kafka, sans dépasser le nombre de cœurs de CPU. Un nombre élevé de tâches peut entraîner une consommation accrue de mémoire et des rééquilibrages fréquents.

snowflake.topic2table.map

Ce paramètre facultatif permet à un utilisateur de spécifier quelles rubriques doivent être mappées à quelles tables. Chaque sujet et son nom de table doivent être séparés par le signe deux-points (voir exemple ci-dessous). Ce nom de table doit être un identificateur non spécifié Snowflake et valide. Pour plus d’informations sur les noms de table valides, voir Exigences relatives à l’identificateur. La configuration de la rubrique sujet permet l’utilisation d’expressions régulières pour définir des rubriques, tout comme l’utilisation de topics.regex. Les expressions régulières ne peuvent pas être ambiguës. Toute rubrique correspondante doit correspondre à une seule table cible.

Exemple :

topics="topic1,topic2,topic5,topic6"
snowflake.topic2table.map="topic1:low_range,topic2:low_range,topic5:high_range,topic6:high_range"

Copy

peut s’écrire ainsi :

topics.regex="topic[0-9]"
snowflake.topic2table.map="topic[0-4]:low_range,topic[5-9]:high_range"

Copy

value.converter.schema.registry.url

Si le format est Avro et que vous utilisez un service de registre de schéma, il doit s’agir de l’URL du service de registre de schéma. Sinon, ce champ devrait être vide.

value.converter.break.on.schema.registry.error

Si vous chargez des données Avro à partir du service de registre de schéma, cette propriété détermine si le connecteur Kafka doit cesser de consommer des enregistrements s’il rencontre une erreur lors de la récupération de l’ID de schéma. La valeur par défaut est false. Définissez la valeur sur true pour activer ce comportement.

jvm.proxy.host

Pour permettre au connecteur Snowflake Kafka d’accéder à Snowflake via un serveur proxy, définissez ce paramètre pour spécifier l’hôte de ce serveur proxy.

jvm.proxy.port

Pour permettre au connecteur Snowflake Kafka d’accéder à Snowflake via un serveur proxy, définissez ce paramètre pour spécifier le port de ce serveur proxy.

snowflake.streaming.max.client.lag

Specifies how often the connector flushes the data to Snowflake, in seconds.

Valeurs:

• Minimum : 1 seconde

• Maximum : 600 secondes

Par défaut:

:code:`1`seconde

jvm.proxy.username

Nom d’utilisateur qui s’authentifie auprès du serveur proxy.

jvm.proxy.password

Mot de passe du nom d’utilisateur qui s’authentifie auprès du serveur proxy.

snowflake.jdbc.map

Exemple : "snowflake.jdbc.map": "networkTimeout:20,tracing:WARNING"

Les propriétés JDBC supplémentaires (voir Référence Paramètre de connexion pilote JDBC) ne sont pas validées. Ces propriétés supplémentaires ne sont pas validées et ne doivent pas remplacer ni être utilisées à la place des propriétés obligatoires telles que : jvm.proxy.xxx, snowflake.user.name, snowflake.private.key, snowflake.schema.name etc.

Spécification de l’une des combinaisons suivantes :

• Propriété tracing avec variable d’environnement JDBC_TRACE

• Propriété database avec snowflake.database.name

Cela entraînera un comportement ambigu et le comportement sera déterminé par le pilote JDBC.

value.converter.basic.auth.credentials.source

Si vous utilisez le format de données Avro et avez besoin d’un accès sécurisé au registre de schémas Kafka, définissez ce paramètre sur la chaîne « USER_INFO », puis définissez le paramètre value.converter.basic.auth.user.info décrit ci-dessous. Sinon, omettez ce paramètre.

value.converter.basic.auth.user.info

Si vous utilisez le format de données Avro et avez besoin d’un accès sécurisé au registre de schémas Kafka, définissez ce paramètre sur la chaîne « <ID_utilisateur>:<motdepasse> », puis définissez le paramètre value.converter.basic.auth.credentials.source décrit ci-dessus. Sinon, omettez ce paramètre.

snowflake.metadata.createtime

Si la valeur est définie sur FALSE, la valeur de la propriété CreateTime est omise des métadonnées dans la colonne RECORD_METADATA. La valeur par défaut est TRUE.

snowflake.metadata.topic

Si la valeur est définie sur FALSE, la valeur de la propriété topic est omise des métadonnées dans la colonne RECORD_METADATA. La valeur par défaut est TRUE.

snowflake.metadata.offset.and.partition

Si la valeur est définie sur FALSE, les valeurs de propriété Offset et Partition sont omises des métadonnées de la colonne RECORD_METADATA. La valeur par défaut est TRUE.

snowflake.metadata.all

Si la valeur est définie sur FALSE, les métadonnées de la colonne RECORD_METADATA sont complètement vides. La valeur par défaut est TRUE.

transforms

Spécifier pour ignorer les enregistrements tombstone rencontrés par le connecteur Kafka et ne pas les charger dans la table cible. Un enregistrement de type tombstone est défini comme un enregistrement dont le champ de valeur est entièrement nul.

Définir la valeur de la propriété sur "tombstoneHandlerExample".

Note

Utilisez cette propriété uniquement avec les convertisseurs de communauté Kafka (c’est-à-dire la valeur de la propriété value.converter) (par exemple org.apache.kafka.connect.json.JsonConverter ou org.apache.kafka.connect.json.AvroConverter). Pour gérer le traitement des enregistrements tombstone avec les convertisseurs Snowflake, utilisez plutôt la propriété behavior.on.null.values.

transforms.tombstoneHandlerExample.type

Requis lors de la définition de la propriété transforms.

Définissez la valeur de la propriété sur "io.confluent.connect.transforms.TombstoneHandler"

behavior.on.null.values

Spécifiez comment le connecteur Kafka doit traiter les enregistrements tombstone. Un enregistrement de type tombstone est défini comme un enregistrement dont le champ de valeur est entièrement nul. Pour Snowpipe, cette propriété est prise en charge par le connecteur Kafka version 1.5.5 et ultérieure. Pour Snowpipe Streaming, cette propriété est prise en charge par le connecteur Kafka à partir de la version 2.1.0.

Cette propriété prend en charge les valeurs suivantes :

DEFAULT

Lorsque le connecteur Kafka rencontre un enregistrement tombstone, il insère une chaîne JSON vide dans la colonne de contenu.

IGNORE

Le connecteur Kafka ignore les enregistrements tombstone et n’insère pas de lignes pour ces enregistrements.

La valeur par défaut est DEFAULT.

Note

L’ingestion des enregistrements tombstone varie selon les méthodes d’ingestion :

• Pour Snowpipe, le connecteur Kafka utilise uniquement les convertisseurs Snowflake. Pour gérer le traitement des enregistrements tombstone avec les convertisseurs communautaires Kafka, utilisez plutôt les propriétés transform et transforms.tombstoneHandlerExample.type.

• Pour Snowpipe Streaming, le connecteur Kafka utilise uniquement des convertisseurs communautaires.

Les enregistrements envoyés aux courtiers Kafka ne doivent pas être NULL, car ces enregistrements seront abandonnés par le connecteur Kafka, ce qui entraînera des décalages manquants. Les décalages manquants interrompront le connecteur Kafka dans des cas d’utilisation spécifiques. Il est recommandé d’utiliser les enregistrements tombstone plutôt que les enregistrements NULL.

Utilisation de l’authentification par paire de clés et rotation de clés¶

Le connecteur Kafka repose sur l’authentification par paire de clés au lieu de l’authentification par nom d’utilisateur et mot de passe. Cette méthode d’authentification nécessite une paire de clés de 2048 bits (minimum) RSA. Générez la paire de clés publiques-privées via OpenSSL. La clé publique est attribuée à l’utilisateur Snowflake défini dans le fichier de configuration.

Après avoir terminé les tâches d’authentification par paire de clés sur cette page et les tâches pour la rotation de paires de clés, évaluez la recommandation pour Externaliser les secrets, plus loin dans cette rubrique.

Pour configurer la paire de clés publiques/privées :

1. Depuis la ligne de commande d’une fenêtre de terminal, générez une clé privée.

   Vous pouvez générer une version chiffrée ou non chiffrée de la clé privée.

   Note

   Le connecteur Kafka prend en charge des algorithmes de chiffrement validés pour répondre aux exigences de la norme Federal Information Processing Standard (140-2) (FIPS 140-2). Pour plus d’informations, voir FIPS 140-2.

   Pour générer une version non chiffrée, utilisez la commande suivante :

   $ openssl genrsa -out rsa_key.pem 2048
   

   Copy

   Pour générer une version chiffrée, utilisez la commande suivante :

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

   Copy

   Où <algorithme> est un algorithme de chiffrement conforme à FIPS 140-2.

   Par exemple, pour spécifier AES 256 comme algorithme de chiffrement :

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

   Copy

   Si vous générez une version chiffrée de la clé privée, enregistrez la phrase secrète. Plus tard, vous spécifierez la phrase secrète dans la propriété snowflake.private.key.passphrase du fichier de configuration de Kafka.

   Exemple de clé privée PEM

   -----BEGIN ENCRYPTED PRIVATE KEY-----
   MIIE6TAbBgkqhkiG9w0BBQMwDgQILYPyCppzOwECAggABIIEyLiGSpeeGSe3xHP1
   wHLjfCYycUPennlX2bd8yX8xOxGSGfvB+99+PmSlex0FmY9ov1J8H1H9Y3lMWXbL
   ...
   -----END ENCRYPTED PRIVATE KEY-----
   

   Copy

2. Depuis la ligne de commande, générez la clé publique en faisant référence à la clé privée :

   En supposant que la clé privée soit chiffrée et contenue dans le fichier nommé rsa_key.p8, utilisez la commande suivante :

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

   Copy

   Exemple de clé publique PEM

   -----BEGIN PUBLIC KEY-----
   MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy+Fw2qv4Roud3l6tjPH4
   zxybHjmZ5rhtCz9jppCV8UTWvEXxa88IGRIHbJ/PwKW/mR8LXdfI7l/9vCMXX4mk
   ...
   -----END PUBLIC KEY-----
   

   Copy

3. Copiez les fichiers de clés publiques et privées dans un répertoire local en vue de leur stockage. Notez le chemin d’accès aux fichiers. La clé privée est stockée au format PKCS#8 (Public Key Cryptography Standards) et est chiffrée à l’aide de la phrase secrète que vous avez spécifiée à l’étape précédente ; toutefois, le fichier doit toujours être protégé contre tout accès non autorisé au moyen du mécanisme d’autorisation de fichier fourni par votre système d’exploitation. Il est de la responsabilité de l’utilisateur de sécuriser le fichier lorsqu’il n’est pas utilisé.

4. Connectez-vous à Snowflake. Attribuez la clé publique à l’utilisateur Snowflake en utilisant ALTER USER.

   For example:

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

   Copy

   Note

   • Seuls les administrateurs de sécurité (c’est-à-dire les utilisateurs ayant le rôle SECURITYADMIN) ou ayant un rôle supérieur peuvent modifier un utilisateur.

   • Excluez l’en-tête et le pied de page de la clé publique dans l’instruction SQL.

   Vérifiez l’empreinte de la clé publique de l’utilisateur en utilisant 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.                                  |
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   

   Copy

   Note

   La propriété RSA_PUBLIC_KEY_2_FP est décrite dans Configuration de la rotation de paires de clés.

5. Copiez et collez la clé privée complète dans le champ snowflake.private.key du fichier de configuration. Sauvegardez le fichier.

Mode distribué¶

À partir d’une fenêtre de terminal, exécutez la commande suivante :

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

Mode autonome¶

À partir d’une fenêtre de terminal, exécutez la commande suivante :

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