Apache Kafka avec DLQ et métadonnées¶

Fonctionnalités clés¶

Le connecteur Apache Kafka avec DLQ et métadonnées offre des fonctionnalités complètes :

• Prise en charge de la file d’attente de lettres mortes (DLQ) pour la gestion des messages en échec

• Colonne RECORD_METADATA avec les métadonnées des messages Kafka

• Configuration de la schématisation - activer ou désactiver la détection des schémas

• Prise en charge des tables Iceberg avec évolution du schéma

• Prise en charge de plusieurs formats de messages - JSON et AVRO

• Intégration du registre des schémas pour les messages AVRO

• Mappage sujet-table avec des modèles avancés

• Prise en charge de l’authentificationSASL

Paramètres spécifiques¶

Outre les paramètres communs décrits à l’adresse Paramétrage du connecteur Openflow pour Kafka, ce connecteur comprend des contextes de paramètres supplémentaires pour les fonctions avancées.

Comportement de schématisation¶

Le comportement du connecteur change en fonction du paramètre Activation de la schématisation :

Détection et évolution des schémas¶

Le connecteur prend en charge la détection et l’évolution des schémas. La structure des tables dans Snowflake peut être définie et évoluer automatiquement pour supporter la structure des nouvelles données chargées par le connecteur.

Sans détection et évolution du schéma, la table Snowflake chargée par le connecteur ne comporte que deux colonnes OBJECT: RECORD_CONTENT et RECORD_METADATA.

Lorsque la détection et l’évolution des schémas sont activées, Snowflake peut détecter le schéma des données en continu et charger les données dans des tables qui correspondent automatiquement à n’importe quel schéma défini par l’utilisateur. Snowflake permet également d’ajouter de nouvelles colonnes ou de supprimer la NOT NULL contrainte des colonnes manquantes dans les nouveaux fichiers de données.

La détection des schémas avec le connecteur est possible avec ou sans registre de schémas fourni. Si vous utilisez le registre des schémas (Avro), la colonne sera créée avec les types de données définis dans le registre des schémas fourni. En l’absence de registre de schéma (JSON), le type de données sera déduit sur la base des données fournies.

Le ARRAY JSON n’est pas pris en charge pour une schématisation plus poussée.

File d’attente lettres mortes (DLQ)¶

La fonctionnalité DLQ permet de traiter les messages qui ne peuvent être traités avec succès :

Support de table Iceberg¶

Le connecteur Openflow for Kafka peut ingérer des données dans une table Apache Iceberg™ gérée par Snowflake lorsque Iceberg activé est réglé sur true.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    record_metadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

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

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    id INT,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    RECORD_CONTENT OBJECT(
        id INT,
        body_temperature FLOAT,
        name STRING,
        approved_coffee_types ARRAY(STRING),
        animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
        date_added DATE
    )
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

Cas d’utilisation¶

Ce connecteur est idéal pour :

• Environnements de production exigeant DLQ

• Lignage des données et audit où les métadonnées Kafka sont importantes

• Traitement de messages complexes avec des exigences d’évolution du schéma

• Intégration de la table Iceberg

Si vous avez besoin d’une ingestion plus simple, sans métadonnées ni fonctions DLQ, optez plutôt pour les connecteurs Apache Kafka pour le format de données JSON/AVRO.