Paramétrage du connecteur Openflow pour Kafka¶

Conditions préalables¶

1. Assurez-vous d’avoir consulté Connecteur Openflow Snowflake pour Kafka.

2. Assurez-vous d’avoir Configuration d’Openflow - BYOC ou de Configurer Openflow - Déploiements Snowflake.

3. Si vous utilisez Openflow - Déploiements Snowflake, assurez-vous d’avoir examiné la configuration des domaines requis et d’avoir accordé l’accès aux domaines requis pour le connecteur Kafka. Le connecteur doit pouvoir se connecter à tous les brokers Kafka du cluster.

Paramétrage du compte Snowflake¶

En tant qu’administrateur de compte Snowflake, effectuez les tâches suivantes :

1. Créez un nouvel utilisateur du service Snowflake avec le type SERVICE.

2. Créez un nouveau rôle ou utilisez un rôle existant et accordez les:ref:privilèges de base de données <label-database_privileges>.

   Le connecteur nécessite un utilisateur pour créer la table de destination. Assurez-vous que l’utilisateur dispose des privilèges nécessaires pour gérer les objets Snowflake :

   Objet	Privilège	Remarques
   Base de données	USAGE
   Schéma	USAGE
   Table	OWNERSHIP	Nécessaire pour que le connecteur ingère des données dans une table.

   See moreShow less

   Expand

   Snowflake recommande de créer un utilisateur et un rôle distincts pour chaque cluster Kafka pour un meilleur contrôle d’accès.

   Vous pouvez utiliser le script suivant pour créer et configurer un rôle personnalisé (exigence : SECURITYADMIN ou équivalent) :

   CopyExpand

   USE ROLE securityadmin;
   CREATE ROLE openflow_kafka_connector_role_1;
   
   GRANT USAGE ON DATABASE kafka_db TO ROLE openflow_kafka_connector_role_1;
   GRANT USAGE ON SCHEMA kafka_schema TO ROLE openflow_kafka_connector_role_1;
   

   Show lessSee more

   Scroll to top

   Note

   Les privilèges doivent être accordés directement au rôle de connecteur et ne peuvent pas être hérités.

3. Configurer la table de destination

   Snowflake recommande fortement d’utiliser l’évolution de schéma côté serveur pour les modifications de schéma et une :doc:` table d’erreurs pour la journalisation des erreurs DML</user-guide/data-load-overview>`. L’exemple suivant montre comment créer une table et ajouter les autorisations OWNERSHIP adaptées.

   CopyExpand

   USE ROLE openflow_kafka_connector_role_1;
   
   CREATE TABLE kafka_db.kafka_schema.<DESTINATION_TABLE_NAME> (
     kafkaMetadata variant
   )
   ENABLE_SCHEMA_EVOLUTION = TRUE
   ERROR_LOGGING = TRUE;
   
   USE ROLE securityadmin;
   GRANT OWNERSHIP ON TABLE existing_table1 TO ROLE openflow_kafka_connector_role_1;
   

   Show lessSee more

   Scroll to top

   Ce connecteur prend en charge la détection et l’évolution automatiques des schémas. La structure des tables dans Snowflake est définie et évolue automatiquement pour prendre en charge la structure des nouvelles données chargées par le connecteur. Elle mappe automatiquement les clés de premier niveau du contenu de l’enregistrement aux colonnes de la table correspondant par leur nom (non sensible à la casse).

   Lorsque l’évolution de schéma est activée, Snowflake peut automatiquement développer la table de destination en ajoutant de nouvelles colonnes détectées dans le flux entrant et en supprimant des contraintes NOT NULL pour prendre en charge de nouveaux modèles de données. Pour plus d’informations, consultez Évolution du schéma de la table.

   Si ENABLE_SCHEMA_EVOLUTION n’est pas activé, vous devez créer le schéma manuellement en étendant la définition de table. Le connecteur tente de faire correspondre les clés de premier niveau du contenu de l’enregistrement aux colonnes de la table par leur nom. Si les clés du fichier JSON ne correspondent pas aux colonnes de la table, le connecteur ignore les clés.

4. (Facultatif) Configurer un gestionnaire de secrets

   Snowflake recommande vivement cette étape. Configurez un gestionnaire de secrets pris en charge par Openflow, par exemple AWS, Azure et Hashicorp, et stockez les clés publiques et privées dans le magasin de secrets.

   1. Déterminez comment vous vous authentifierez auprès du gestionnaire de secrets une fois celui-ci configuré. SurAWS, Snowflake recommande d’utiliser le rôle d’instance EC2 associé à Openflow, de sorte qu’aucun autre secret ne doit être conservé.

   2. Configurez un fournisseur de paramètres associé à ce gestionnaire de secrets dans Openflow, à partir du menu hamburger en haut à droite. Accédez à Paramètres du contrôleur > Fournisseur de paramètres et récupérez les valeurs de vos paramètres.

   3. Référencez tous les identifiants de connexion avec les chemins de paramètres associés afin qu’aucune valeur sensible ne doive être conservée dans Openflow.

5. Octroyer l’accès aux utilisateurs

   Pour tout autre utilisateur de Snowflake qui a besoin d’accéder aux données ingérées brutes par le connecteur (par exemple, pour un traitement personnalisé dans Snowflake), accordez à ces utilisateurs le rôle créé à l’étape 1.

Définir le connecteur¶

En tant qu’ingénieur des données, effectuez les tâches suivantes pour installer et configurer le connecteur :

Comprendre la colonne KAFKAMETADATA¶

Le connecteur remplit la structure KAFKAMETADATA avec des métadonnées sur l’enregistrement Kafka. La structure contient les informations suivantes :

Mesure de la latence d’ingestion¶

Pour le suivi des modifications, le traitement incrémentiel et les requêtes Time Travel basées sur l’heure de modification des lignes, la fonctionnalité ROW_TIMESTAMP peut être utilisée.

Activez-la en exécutant la commande suivante sur votre table de destination :

ALTER TABLE <DESTINATION_TABLE> SET ROW_TIMESTAMP = TRUE;

Une fois les horodatages de lignes activés, les tables exposent la colonne METADATA$ROW_LAST_COMMIT_TIME, qui renvoie l’horodatage de la dernière modification de chaque ligne.

Pour plus d’informations, voir METADATA$ROW_LAST_COMMIT_TIME.

Utilisation du connecteur avec des tables Apache Iceberg™¶

Le connecteur peut ingérer des données dans des tables Apache Iceberg™ gérées par Snowflake, mais doit répondre aux exigences suivantes :

• Vous devez avoir obtenu le privilège USAGE sur le volume externe associé à votre table Apache Iceberg™.

• Vous devez créer une table Apache Iceberg™ avant d’exécuter le connecteur.

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

{
  "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"
}

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
  kafkaMetadata OBJECT(
    topic STRING,
    partition INTEGER,
    offset INTEGER,
    key STRING,
    headers variant,
    timestamp INTEGER
  ),
  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;

Utilisation du connecteur avec des tables interactives¶

Les tables interactives sont un type spécial de table Snowflake optimisée pour les requêtes à faible latence et à forte concurrence. Pour plus d’informations, voir Tables interactives et entrepôts interactifs Snowflake.

1. Créez une table interactive :

   CopyExpand

   CREATE INTERACTIVE TABLE REALTIME_METRICS (
     metric_name VARCHAR,
     metric_value NUMBER,
     source_topic VARCHAR,
     timestamp TIMESTAMP_NTZ
   ) AS (SELECT
         $1:M_NAME::VARCHAR,
         $1:M_VALUE::NUMBER,
         $1:RECORD_METADATA.topic::VARCHAR,
         $1:RECORD_METADATA.timestamp::TIMESTAMP_NTZ
         from TABLE(DATA_SOURCE(TYPE => 'STREAMING')));
   

   Show lessSee more

   Scroll to top

Considérations importantes :

• Les tables interactives ont des limitations et des restrictions de requêtes spécifiques. Consultez Tables interactives et entrepôts interactifs Snowflake avant de les utiliser avec le connecteur.

• Pour les tables interactives, toutes les transformations requises doivent être gérées dans la définition de la table.

• Des entrepôts interactifs sont nécessaires pour interroger efficacement les tables interactives.

Utilisation du connecteur avec un schéma défini par le client pour la table de destination¶

Le connecteur traite chaque enregistrement Kafka comme une ligne à insérer dans une table Snowflake. Par exemple, si vous avez un sujet Kafka avec le contenu du message structuré comme le fichier JSON :

{
  "order_id": 12345,
  "customer_name": "John",
  "order_total": 100.00,
  "isPaid": true
}

Par défaut, vous n’êtes pas obligé de spécifier tous les champs de JSON grâce à la fonctionnalité ENABLE_SCHEMA_EVOLUTION = TRUE. Toutefois, si vous préférez un schéma statique, il peut être créé en exécutant :

CREATE TABLE ORDERS (
  kafkaMetadata OBJECT,
  order_id NUMBER,
  customer_name VARCHAR,
  order_total FLOAT,
  ispaid BOOLEAN
);

Utilisation du connecteur avec un PIPE défini par le client¶

Si vous choisissez de créer votre propre canal, vous pouvez définir la logique de transformation des données dans l’instruction COPY INTO du canal. Vous pouvez renommer les colonnes si nécessaire et convertir les types de données selon vos besoins. Par exemple :

CREATE TABLE ORDERS (
  order_id VARCHAR,
  customer_name VARCHAR,
  order_total VARCHAR,
  ispaid VARCHAR
);

CREATE PIPE ORDERS AS
COPY INTO ORDERS
SELECT
  $1:order_id::STRING,
  $1:customer_name,
  $1:order_total::STRING,
  $1:isPaid::STRING
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'));

Lorsque vous définissez votre propre canal, les colonnes de votre table de destination ne doivent pas nécessairement correspondre aux clés JSON. Vous pouvez renommer les colonnes à l’aide des noms de votre choix et convertir les types de données selon vos besoins.

Pour ajuster le connecteur afin qu’il fonctionne avec un canal personnalisé, effectuez les tâches suivantes :

1. Faites un clic droit sur le processeur PublishSnowpipeStreaming utilisé dans votre flux d’ingestion Kafka dans le canevas Openflow.

2. Sélectionnez Configurer dans le menu contextuel.

3. Accédez à l’onglet Propriétés.

4. Dans le champ Type de destination, sélectionnez Canal.

5. Dans le champ Canal, tapez le nom de votre PIPE.

6. Sélectionnez Appliquer pour enregistrer la configuration.

Personnalisation de la gestion des erreurs¶

Le traitement des erreurs est réparti entre les échecs côté Openflow et les échecs côté serveur au sein du service Snowpipe Streaming.

• Erreurs Openflow (échecs côté client) : Des erreurs telles que des charges utiles impossibles à analyser ou des échecs de transformations personnalisées se produisent avant que les enregistrements n’atteignent Snowflake. Par défaut, ces enregistrements sont ignorés. Il est possible de traiter ces erreurs dans Openflow - utilisez FlowFiles de la relation d’échec d’analyse dans le processeur ConsumeKafka.

• Erreurs Snowpipe Streaming (échecs côté serveur) : Les erreurs des enregistrements qui atteignent Snowflake, mais qui sont incompatibles avec le schéma de la table de destination (par exemple, les discordances de type), sont capturées par l’infrastructure Snowflake. Lorsque la journalisation des erreurs est activée sur la table de destination (error_logging = true), ces lignes qui ont échoué sont automatiquement ingérées dans la table d’erreur de destination.

Réglage des performances¶

Optimisation des performances du connecteur Openflow pour Kafka