Working with the 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.

• You want a simple configuration

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¶

In prior versions of the connector (3.x and earlier), when the schematization feature was disabled, the connector created a destination table with two columns: RECORD_CONTENT and RECORD_METADATA. The RECORD_CONTENT column contained the entire Kafka message content in a column of type VARIANT. The RECORD_METADATA column continues to be supported but the RECORD_CONTENT column is no longer created by the connector. The same functionality can be achieved using SMT transformations (see examples later in this section). The RECORD_CONTENT key is also no longer available in PIPE transformations. For example, this PIPE definition will not work by default:

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¶

The connector inspects the Snowpipe Streaming channel status before committing offsets in Kafka. If the connector detects that the rowsErrorCount property on channel has increased since the connector was started, it raises a fatal error (ERROR_5030) when errors.tolerance=none so that data issues don’t go unnoticed. To allow ingestion to continue while triaging bad rows, set errors.tolerance=all

errors.tolerance=all

Accorder l’utilisation sur un volume externe¶

Pour accorder le privilège USAGE sur le volume externe associé à votre table Apache Iceberg™ à votre rôle pour le connecteur Kafka, exécutez l’instruction suivante :

Par exemple, si votre table Iceberg utilise le volume externe kafka_external_volume et que le connecteur utilise le rôle kafka_connector_role, exécutez l’instruction suivante :

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kafka_external_volume TO ROLE kafka_connector_role;

Créer une table Apache Iceberg™ pour l’ingestion¶

Le connecteur ne crée pas automatiquement des tables Iceberg et ne prend pas en charge l’évolution des schémas. Avant d’exécuter le connecteur, vous devez créer manuellement une table Iceberg.

Lorsque vous créez une table Iceberg, vous pouvez utiliser des types de données Iceberg (notamment VARIANT) ou des types Snowflake compatibles </user-guide/tables-iceberg-data-types>.

Prenons l’exemple du message suivant :

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "options":
    {
        "can_walk": true,
        "can_talk": false
    },
    "date_added": "2024-10-15"
}

Pour créer une table Iceberg pour le message d’exemple, utilisez les instructions suivantes :

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id number(38,0),
    name varchar,
    body_temperature number(4,2),
    approved_coffee_types array(varchar),
    animals_possessed variant,
    options object(can_walk boolean, can_talk boolean),
    date_added date
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id INT,
    name string,
    body_temperature float,
    approved_coffee_types array(string),
    animals_possessed variant,
    date_added date,
    options object(can_walk boolean, can_talk boolean),
    )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy