Snowflake Connector for Kafka : installer et configurer¶

Téléchargement des fichiers du connecteur Kafka¶

Téléchargez le fichier JAR du connecteur Kafka à l’un des emplacements suivants :

Hub Confluent:

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

Le paquet inclut toutes les dépendances requises pour utiliser une clé privée chiffrée ou non chiffrée pour l’authentification par paire de clés. Pour plus d’informations, voir Utilisation de l’authentification par paire de clés et rotation de clés plus loin dans cette rubrique.

Maven Central Repository:

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

Lorsque vous utilisez cette version, vous devez télécharger les bibliothèques de cryptographie Bouncy Castle (fichiers JAR) :

• https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

• https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

Téléchargez ces fichiers dans le même dossier local que le fichier JAR du connecteur Kafka.

Le code source du connecteur est disponible sur https://github.com/snowflakedb/snowflake-kafka-connector.

Installation du connecteur Kafka¶

Installez le connecteur Kafka en suivant les instructions fournies pour installer d’autres connecteurs :

https://docs.confluent.io/current/connect/userguide.html

Installer Apache Kafka¶

1. Téléchargez le paquet Kafka à partir du site officiel de Kafka.

2. Dans une fenêtre de terminal, accédez au répertoire dans lequel vous avez téléchargé le fichier de paquet.

3. Exécutez la commande suivante pour décompresser le fichier kafka_<scala_version>-<kafka_version>.tgz :

   CopyExpand

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

   Show lessSee more

   Scroll to top

Installer le JDK¶

Installez et configurez le kit de développement Java (JDK) version 11 ou supérieure. Snowflake teste avec l’édition standard (SE) de JDK. L’édition entreprise (EE) devrait être compatible mais n’a pas été testée.

Si vous avez déjà installé le JDK, vous pouvez ignorer cette section.

1. Téléchargez le JDK du site web Oracle JDK.

2. Installez ou décompressez le JDK.

3. En suivant les instructions pour votre système d’exploitation, définissez la variable d’environnement JAVA_HOME pour qu’elle pointe vers le répertoire contenant le JDK.

Téléchargement des fichiers JAR du connecteur Kafka¶

1. Téléchargez le fichier JAR du connecteur Kafka à partir du référentiel central Maven :

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

2. Téléchargez les fichiers jar de la bibliothèque de cryptographie Bouncy Castle :

   • https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

   • https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

3. Si vos données Kafka sont diffusées au format Apache Avro, téléchargez le fichier JAR (1.11.4) :

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

Le code source du connecteur est disponible sur https://github.com/snowflakedb/snowflake-kafka-connector.

Installation du connecteur Kafka¶

Copiez les fichiers JAR que vous avez téléchargés dans Installation du connecteur pour Apache Kafka open source dans le dossier <kafka_dir>/libs.

Propriétés de schématisation et de validation¶

Ces propriétés contrôlent la manière dont le connecteur traite et valide les données. Pour les nouvelles installations, les valeurs par défaut fonctionnent bien. Si vous migrez depuis v3, examinez Migrer du connecteur Kafka v3 vers v4 pour obtenir des conseils sur les valeurs à utiliser.

snowflake.enable.schematization

Contrôle si les enregistrements entrants sont schématisés en colonnes de tables individuelles ou inclus dans des colonnes VARIANT héritées.

Lorsque true (par défaut), les champs d’enregistrement sont mappés aux colonnes de tables individuelles par leur nom. Lorsque false, le connecteur stocke les enregistrements dans deux colonnes VARIANT (RECORD_CONTENT et RECORD_METADATA), correspondant au comportement v3.

Par défaut:

true

snowflake.validation

Contrôles où la validation des données et l’évolution des schémas sont effectuées.

server_side (par défaut) : La validation est effectuée par le backend Snowflake, aligné avec le comportement COPY et Snowpipe. Les enregistrements invalides sont capturés dans les tables d’erreurs. Prend en charge à la fois les modes de canal par défaut et les modes de canal définis par l’utilisateur.

client_side : Le connecteur valide les types de données et la compatibilité des schémas avant d’envoyer les lignes à Snowflake. Prend en charge la file d’attente de lettres mortes (DLQ) pour les enregistrements invalides. Ne fonctionne qu’avec le mode de canal par défaut.

Pour plus de détails, voir Validation et traitement des erreurs.

Par défaut:

server_side

Propriétés de migration et de compatibilité¶

Ces propriétés sont pertinentes lors de la migration depuis v3. Pour les nouvelles installations, vous pouvez les ignorer et définir snowflake.streaming.validate.compatibility.with.classic=false.

snowflake.compatibility.enable.autogenerated.table.name.sanitization

Contrôle la manière dont les noms de tables automatiquement générés sont dérivés des noms de rubriques.

Lorsque false (par défaut), les noms de sujets sont utilisés tels quels pour les noms de tables, en préservant la casse et les caractères spéciaux. Les noms de tables sont créés sous forme d’identificateurs cités.

Lorsque true, les caractères d’identificateurs Snowflake invalides sont remplacés par des traits de soulignement, les noms sont en majuscules, et un code de hachage est ajouté pour l’unicité. Cela correspond au comportement v3.

Par défaut:

false

snowflake.compatibility.enable.column.identifier.normalization

Contrôle la manière dont les identificateurs de colonnes sont traités.

Lorsque false (par défaut), les identificateurs de colonnes conservent les majuscules et les caractères spéciaux tels quels.

Lorsque true, les identificateurs de colonnes sont normalisés en majuscules, correspondant au comportement v3.

Par défaut:

false

snowflake.streaming.validate.compatibility.with.classic

Active la validation du démarrage qui vérifie si toutes les configurations liées à la migration sont explicitement définies. Lorsque true, le connecteur échoue au démarrage avec une erreur descriptive si l’une des configurations suivantes est manquante ou incompatible avec le comportement v3 :

• snowflake.validation doit être client_side

• snowflake.compatibility.enable.column.identifier.normalization doit être true

• snowflake.compatibility.enable.autogenerated.table.name.sanitization doit être true

• snowflake.enable.schematization doit être explicitement définie sur true ou false (la valeur par défaut a changé entre v3 et v4)

• snowflake.streaming.classic.offset.migration doit être explicitement définie

• snowflake.streaming.classic.offset.migration.include.connector.name doit être explicitement définie (lorsque la migration de décalage est strict ou best_effort)

Cela empêche l’exécution accidentelle de v4 avec une configuration v3 copiée sans examiner les modifications de rupture. Définissez cette valeur sur false pour ignorer cette vérification après avoir confirmé votre configuration.

Note

Pour les nouvelles installations (qui ne migrent pas depuis v3), définissez cette valeur sur false. Le validateur de compatibilité n’est nécessaire que lors de la mise à niveau à partir d’un déploiement v3 existant.

Pour plus d’informations, voir Migrer du connecteur Kafka v3 vers v4.

Par défaut:

true

Propriétés de migration de décalage¶

Ces propriétés contrôlent la manière dont v4 migre les décalages validés des canaux Snowpipe Streaming (SSv1) v3. Elles ne sont pertinentes que lors de la migration à partir d’un connecteur v3 qui utilisait snowflake.ingestion.method=SNOWPIPE_STREAMING. Si vous migrez à partir du mode Snowpipe v3 (ingestion basée sur des fichiers), définissez snowflake.streaming.classic.offset.migration sur skip.

snowflake.streaming.classic.offset.migration

Contrôle la manière dont v4 migre les décalages des canaux Snowpipe Streaming (SSv1) v3.

strict : v4 recherche le décalage validé à partir du canal SSv1 v3 et reprend à partir de là. Si le canal SSv1 n’est pas trouvé, le connecteur échoue avec une erreur.

best_effort : v4 tente de rechercher le décalage validé à partir du canal SSv1 v3. Si le canal n’est pas trouvé, v4 revient au décalage du groupe de consommateurs Kafka.

skip (par défaut) : Aucune migration de décalage SSv1 n’est effectuée. v4 utilise le décalage du groupe de consommateurs Kafka. Utilisez cette option lorsque vous migrez à partir du mode Snowpipe v3 (et non de Snowpipe Streaming).

Par défaut:

skip

snowflake.streaming.classic.offset.migration.include.connector.name

Contrôle si la recherche du nom du canal SSv1 inclut le nom du connecteur. Cela doit correspondre à la manière dont votre connecteur v3 a été configuré. Dans la v3, la propriété snowflake.streaming.channel.name.include.connector.name contrôlait si le nom du connecteur était inclus dans le nom du canal.

Définissez-la sur true si votre connecteur v3 disposait de snowflake.streaming.channel.name.include.connector.name=true ou si vous exécutiez la version 2.1.0 ou 2.1.1 du connecteur Kafka (ces versions incluaient le nom du connecteur par défaut). Définissez-la sur false sinon.

Nécessaire uniquement lorsque snowflake.streaming.classic.offset.migration est strict ou best_effort.

Par défaut:

none (doit être explicitement défini lorsque la migration de décalage est active)

Propriétés de traitement des erreurs¶

errors.tolerance

Contrôle la manière dont le connecteur répond aux erreurs lors de l’ingestion.

none (par défaut) : La tâche du connecteur échoue à la première erreur. Avec la validation côté serveur, la détection des erreurs est asynchrone, de sorte que quelques enregistrements après celui qui est cassé peuvent encore être ingérés avant l’échec de la tâche.

all : Le connecteur poursuit l’ingestion des données. Avec la validation côté client, les enregistrements invalides sont acheminés vers la DLQ (si configurée) ou supprimés de manière silencieuse.

Avertissement

Définir errors.tolerance=all sans configurer de sujet DLQ entraîne la suppression silencieuse des enregistrements invalides lors de l’utilisation de la validation côté client. Cela peut entraîner une perte de données.

Par défaut:

none

errors.deadletterqueue.topic.name

Nom du sujet Kafka pour la file d’attente de lettres mortes. Effectif uniquement lorsque snowflake.validation=client_side et errors.tolerance=all.

Par défaut:

vide (DLQ désactivée)

errors.log.enable

Lorsque true, les erreurs sont journalisées avec les détails de l’opération qui a échoué et les propriétés de l’enregistrement.

Par défaut:

false

enable.task.fail.on.authorization.errors

Lorsque true, la tâche du connecteur échoue immédiatement en cas d’erreurs d’autorisation de Snowflake. Lorsque false, le connecteur effectue une nouvelle tentative.

Par défaut:

false

Mise en cache des propriétés¶

snowflake.cache.table.exists

Active la mise en cache pour les contrôles d’existence des tables, réduisant le nombre de requêtes de métadonnées vers Snowflake.

Par défaut:

true

snowflake.cache.table.exists.expire.ms

Temps d’expiration du cache en millisecondes pour les contrôles d’existence des tables.

Par défaut:

300000 (5 minutes)

snowflake.cache.pipe.exists

Active la mise en cache pour les contrôles d’existence des canaux.

Par défaut:

true

snowflake.cache.pipe.exists.expire.ms

Temps d’expiration du cache en millisecondes pour les contrôles d’existence des canaux.

Par défaut:

300000 (5 minutes)

Propriétés de surveillance et de diagnostic¶

jmx

Active JMXMBeans pour les métriques du connecteur. Pour plus d’informations, voir Surveiller le Snowflake Connector for Kafka.

Par défaut:

true

enable.mdc.logging

Active MDC (Mapped Diagnostic Context) pour ajouter le contexte du connecteur aux messages de journalisation, ce qui est utile lors de l’exécution de plusieurs instances de connecteur.

Par défaut:

false

snowflake.streaming.metadata.connectorPushTime

Lorsque true, comprend l’horodatage SnowflakeConnectorPushTime dans RECORD_METADATA. Ce champ enregistre quand le connecteur a mis en mémoire tampon un enregistrement à des fins d’ingestion et est utile pour estimer la latence de bout en bout.

Par défaut:

true

Propriétés avancées¶

snowflake.streaming.client.provider.override.map

Remplacements des propriétés du client Snowpipe Streaming. Format : key1:value1,key2:value2. À utiliser uniquement après avoir consulté le Support Snowflake.

Par défaut:

empty

Autres propriétés¶

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

Liste des correspondances entre sujet et table, séparées par des virgules, au format topic:table. Prend en charge les modèles regex pour les noms de sujets. Les expressions régulières ne peuvent pas être ambiguës ; tout sujet correspondant doit correspondre à une seule table cible.

Les noms de sujets et de tables peuvent être entre guillemets doubles pour prendre en charge les caractères spéciaux (deux-points, virgules, espaces). Les noms de tables qui ne sont pas entre guillemets sont en majuscules. Les noms de tables entre guillemets conservent la casse.

Pour des exemples détaillés incluant des modèles regex, des correspondances plusieurs-à-un et l’utilisation des guillemets, consultez Correspondance explicite entre sujet et table.

Exemple :

CopyExpand

snowflake.topic2table.map=topic1:low_range,topic2:low_range,"my:topic":"My_Table"

Show lessSee more

Scroll to top

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.

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 et des propriétés similaires.

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.

Externaliser les secrets¶

Snowflake recommande fortement d’externaliser des secrets tels que la clé privée et de les stocker sous une forme chiffrée ou dans un service de gestion de clés tel que AWS Key Management Service (KMS), Microsoft Azure Key Vault, ou HashiCorp Vault. Pour ce faire, utilisez une implémentation ConfigProvider sur votre cluster Kafka Connect.

Pour plus d’informations, voir la description de ce service par Confluent.