Comment fonctionne le Snowflake High Performance connector for Kafka¶

Noms de sujet, noms de table et noms de canal¶

En fonction des paramètres de configuration, le connecteur utilisera des noms différents pour la table de destination. Le nom de la table de destination est toujours dérivé du nom du sujet.

Comprendre RECORD_METADATA¶

Le connecteur remplit la structure RECORD_METADATA avec des métadonnées sur l’enregistrement Kafka. Ces métadonnées sont envoyées via la source de données Snowpipe Streaming à Snowflake, où elles deviennent disponibles dans les transformations de canal à l’aide de l’accesseur $1:RECORD_METADATA. La structure RECORD_METADATA est disponible en mode de canal défini par l’utilisateur et en mode de canal par défaut. Son contenu peut être enregistré dans la colonne de type. VARIANT ou des champs individuels peuvent être extraits et enregistrés dans des colonnes distinctes.

Exemple de canal avec des transformations et des métadonnées :

CREATE PIPE ORDERS AS
COPY INTO ORDERS_TABLE
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total,
    $1:RECORD_METADATA.topic AS source_topic,
    $1:RECORD_METADATA.offset::NUMBER AS kafka_offset,
    $1:RECORD_METADATA.SnowflakeConnectorPushTime::BIGINT AS ingestion_time
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Dans cet exemple :

• Le canal extrait des champs spécifiques du message Kafka (id_commande, nom_client, total_commande)

• Il capture également des champs de métadonnées (sujet, décalage et horodatage d’ingestion).

• Les valeurs peuvent être converties et/ou transformées selon les besoins.

Comment les champs de métadonnées sont renseignés¶

Le connecteur renseigne automatiquement les champs de métadonnées en fonction des propriétés de l’enregistrement Kafka et de la configuration du connecteur. Vous pouvez contrôler les champs de métadonnées inclus à l’aide de ces paramètres de configuration :

• snowflake.metadata.topic (par défaut : true) - Inclut le nom du sujet

• snowflake.metadata.offset.and.partition (par défaut : true) - Inclut le décalage et la partition

• snowflake.metadata.createtime (par défaut : true) - Inclut l’horodatage de l’enregistrement Kafka

• snowflake.metadata.all (par défaut : true) - Inclut toutes les métadonnées disponibles

Lorsque snowflake.metadata.all=true (par défaut), tous les champs de métadonnées sont renseignés. La définition des indicateurs de métadonnées individuels sur false exclut ces champs spécifiques de la structure RECORD_METADATA.

La colonne RECORD_METADATA contient les informations suivantes par défaut :

La quantité de métadonnées enregistrées dans la colonne RECORD_METADATA est configurable à l’aide des propriétés de configuration Kafka facultatives.

Les noms de champs et les valeurs sont sensibles à la casse.

Comment les enregistrements Kafka sont convertis avant l’ingestion¶

Avant que chaque ligne ne soit transmise à Snowpipe Streaming, le connecteur convertit la valeur de l’enregistrement Kafka Connect en Map<Chaîne, objet> dont les clés doivent correspondre aux noms de vos colonnes cibles (ou peuvent être transformées dans un canal défini par l’utilisateur). Les chaînes primaires, les tableaux d’octets ou les nombres doivent être encapsulés (par exemple en utilisant HoistField SMT) pour que le connecteur reçoive un objet structuré. Le convertisseur applique les règles suivantes :

• Les valeurs nulles sont traitées comme des tombstone. Ils sont ignorés lorsque behavior.on.null.values=IGNORE ou sinon ingéré comme objets JSON vides.

• Les champs numériques et booléens sont transmis tels quels. Les valeurs décimales dont la précision est supérieure à 38 sont sérialisées en tant que chaînes pour rester dans les limites de NUMBER de Snowflake.

• Les charges utiles byte[] et ByteBuffer sont des chaînes codées en Base64 ; par conséquent, stockez-les dans les colonnes VARIANT ou VARCHAR.

• Les tableaux restent des tableaux et les objets imbriqués restent des cartes imbriquées. Déclarez les colonnes VARIANT lorsque vous comptez sur le canal par défaut pour atterrir les données imbriquées telles quelles.

• Les cartes avec des clés autres que des chaînes sont émises sous forme de paires [key, value] car les noms des colonnes Snowflake doivent être du texte.

• Les en-têtes et les clés d’enregistrement sont copiés dans RECORD_METADATA chaque fois que les indicateurs de métadonnées appropriés sont activés.

Si vous avez besoin que le corps du message entier soit conservé sous la forme d’une seule colonne, encapsulez-le dans un nouveau champ de premier niveau à l’aide des SMTs. Voir Colonne RECORD_CONTENT héritée pour le modèle de transformation.

Mode de canal défini par l’utilisateur¶

Dans ce mode, vous contrôlez totalement la transformation des données et le mappage des colonnes.

** Quand utiliser ce mode :**

• Vous avez besoin de noms de colonnes personnalisés différents des noms de champs JSON.

• Vous devez appliquer des transformations de données (conversion de type, masquage, filtrage).

• Vous souhaitez un contrôle total sur la façon dont les données sont mappées aux colonnes.

Mode de canal par défaut¶

Dans ce mode, le connecteur utilise un canal par défaut nommé {tableName}-STREAMING et mappe les champs d’enregistrement Kafka aux colonnes de table correspondant par leur nom (insensible à la casse).

** Quand utiliser ce mode :**

• Les noms de votre clé d’enregistrement Kafka correspondent aux noms de colonne souhaités.

• Vous n’avez pas besoin de transformations de données personnalisées.

• Vous souhaitez une configuration simplifiée.

Mappage des clés d’enregistrement Kafka avec des colonnes de table avec le mode de canal par défaut

Lorsque vous utilisez le mode de canal par défaut, le connecteur utilise un canal par défaut nommé {tableName}-STREAMING et mappent les clés de premier niveau du contenu directement aux colonnes de la table en utilisant une correspondance insensible à la casse.

Utilisation du mode canal par défaut - exemple¶

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  "@&$#* includes special characters": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  record_metadata VARIANT,
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  "has cat" BOOLEAN,
  "!@&$#* includes special characters" BOOLEAN,
  skills VARIANT,
  family VARIANT
);

Utilisation du mode de canal défini par l’utilisateur - exemples¶

Cet exemple montre comment configurer et utiliser des canaux définis par l’utilisateur avec des transformations de données personnalisées.

CREATE TABLE ORDERS (
  order_id NUMBER,
  customer_name VARCHAR,
  order_total NUMBER,
  order_date TIMESTAMP_NTZ,
  source_topic VARCHAR
);

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total::NUMBER,
    $1:order_date::TIMESTAMP_NTZ,
    $1:RECORD_METADATA.topic
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

snowflake.topic2table.map=kafka-orders-topic:ORDERS

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  " @&$#* has Łułósżź": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  has_cat BOOLEAN,
  weird_field_name BOOLEAN,
  skills VARIANT,
  family VARIANT
);

CREATE PIPE PERSON_DATA AS
COPY INTO PERSON_DATA
FROM (
  SELECT
    $1:city,
    $1:age,
    $1:married,
    $1['has cat'] AS has_cat,
    $1[' @&$#* has Łułósżź'] AS weird_field_name,
    $1:skills,
    $1:family
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Mappage explicite de sujet à table¶

Lorsque vous configurez le paramètre snowflake.topic2table.map, le connecteur fonctionne en mode de mappage explicite. Ce mode vous permet de :

• Mapper plusieurs sujets Kafka à une seule table Snowflake.

• Utiliser des noms de tables personnalisés différents des noms de sujets.

• Appliquer des modèles regex pour faire correspondre plusieurs sujets.

Format de configuration :

Le paramètre snowflake.topic2table.map accepte une liste de mappages sujets à tables séparés par des virgules au format :

topic1:table1,topic2:table2,topic3:table3

Exemples de configurations :

Mappage direct des sujets

snowflake.topic2table.map=orders:ORDER_TABLE,customers:CUSTOMER_TABLE

Correspondance de modèles Regex

snowflake.topic2table.map=.*_cat:CAT_TABLE,.*_dog:DOG_TABLE

Cette configuration mappe tous les sujets se terminant par _cat (comme orange_cat, calico_cat) à la table CAT_TABLE et tous les chapitres se terminant par _dog à la table DOG_TABLE.

Plusieurs sujets à une table

snowflake.topic2table.map=topic1:shared_table,topic2:shared_table,topic3:other_table

Cette configuration mappe à la fois topic1 et topic2 dans shared_table, tandis que topic3 est mappé à other_table.

Colonne RECORD_CONTENT héritée¶

Dans les anciennes versions du connecteur, lorsque la fonction de schématisation était désactivée, le connecteur créait une table de destination avec deux colonnes : RECORD_CONTENT et RECORD_METADATA. La colonne RECORD_CONTENT contenait la totalité du contenu du message Kafka dans la colonne de type VARIANT. La colonne RECORD_METADATA est toujours prise en charge, mais la colonne RECORD_CONTENT n’est plus créée par le connecteur. La même fonctionnalité peut être obtenue en utilisant les transformations SMT (voir des exemples plus loin dans cette section). La clé RECORD_CONTENT n’est plus disponible dans les transformations PIPE. Par exemple, cette définition PIPE ne fonctionnera pas par défaut :

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:RECORD_CONTENT
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Si vous voulez que l’intégralité du contenu du message Kafka soit enregistrée dans une seule colonne, ou si vous avez besoin de traiter l’intégralité du contenu du message Kafka dans une transformation PIPE, vous pouvez utiliser la transformation SMT suivante qui place l’intégralité du contenu du message Kafka dans le champ personnalisé souhaité :

transforms=wrapKafkaMessageContent
transforms.wrapKafkaMessageContent.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.wrapKafkaMessageContent.field=your_top_level_field_name

Cette transformation placera l’intégralité du contenu du message Kafka dans un champ personnalisé nommé your_top_level_field_name. Vous pouvez ensuite accéder à l’intégralité du contenu du message Kafka en utilisant l’accesseur $1:your_top_level_field_name dans votre transformation PIPE.

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:your_top_level_field_name
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Sinon, si vous souhaitez enregistrer à la fois l’intégralité des métadonnées et du contenu dans une seule table à l’aide du canal par défaut, ne créez pas de canal personnalisé ; au lieu de cela, créez uniquement une table avec deux colonnes : RECORD_CONTENT et your_top_level_field_name.

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

Pour en savoir plus sur la transformation HoistField$Value, voir la documentation Kafka.

Gestion des erreurs de canal de streaming et des files d’attente de lettres mortes¶

Dans la version 4.0.0-rc4, le connecteur inspecte le statut du canal Snowpipe Streaming avant de valider les décalages. Si Snowflake signale des lignes rejetées (rowsErrorCount > 0), the connector now raises a fatal error (ERROR_5030) when erreurs.tolerance=none so that data issues cannot go unnoticed. To allow ingestion to continue while you triage bad rows, set error.tolerance=tout

errors.tolerance=all

Limitations de la tolérance aux pannes avec le connecteur¶

Les sujets Kafka peuvent être configurés avec une limite d’espace de stockage ou de durée de conservation.

• Si le système est hors ligne pendant une durée supérieure à la durée de conservation, les enregistrements expirés ne seront pas chargés. De même, si la limite d’espace de stockage de Kafka est dépassée, certains messages ne seront pas remis.

• Si les messages du sujet Kafka sont supprimés, ces modifications risquent de ne pas être reflétées dans la table Snowflake.