Dépannage du connecteur Kafka¶

Octets bruts¶

Le nom de fichier <connector_name>/<table_name>/<partition>/offset_(<key>/<value>_)<timestamp>.gz comprend le décalage exact de l’enregistrement qui n’a pas été converti des octets bruts au format de fichier source. Pour résoudre les problèmes, renvoyer l’enregistrement au connecteur Kafka comme un nouvel enregistrement.

Format du fichier source (Avro, JSON ou Protobuf)¶

Si Snowpipe n’a pas pu charger les données à partir de fichiers de la zone de préparation interne créée pour le sujet Kafka, le connecteur Kafka déplace les fichiers vers la zone de préparation de la table cible dans le format du fichier source.

Notez que si un ensemble de fichiers pose plusieurs problèmes, la colonne FIRST_ERROR_MESSAGE dans la sortie COPY_HISTORY indique seulement la première erreur rencontrée. Pour visualiser toutes les erreurs dans les fichiers, il est nécessaire de récupérer les fichiers de la zone de préparation de la table, de les télécharger vers une zone de préparation nommée, puis d’exécuter une instruction COPY INTO <table> avec l’option de copie VALIDATION_MODE définie sur RETURN_ALL_ERRORS. L’option de copie VALIDATION_MODE commande une instruction COPY pour valider les données à charger et retourner les résultats en fonction de l’option de validation spécifiée. Aucune donnée n’est chargée lorsque cette option de copie est spécifiée. Dans l’instruction, faites référence à l’ensemble des fichiers que vous avez tenté de charger à l’aide du connecteur Kafka.

Lorsque tous les problèmes liés aux fichiers de données sont résolus, vous pouvez charger les données manuellement à l’aide d’une ou plusieurs instructions COPY.

L’exemple suivant fait référence aux fichiers de données situés dans la zone de préparation de la table pour la table mytable de la base de données et du schéma mydb.public.

Pour valider les fichiers de données dans la zone de préparation de la table et résoudre les erreurs :

1. Répertoriez tous les fichiers situés dans la zone de préparation de la table avec LIST.

   Par exemple :

   LIST @mydb.public.%mytable;
   

   Copy

   Les exemples de cette section supposent que JSON est le format source des fichiers de données.

2. Téléchargez les fichiers créés par le connecteur Kafka sur votre machine locale en utilisant GET.

   Par exemple, téléchargez les fichiers dans un répertoire nommé data sur votre machine locale :

   Linux ou macOS:

   GET @mydb.public.%mytable file:///data/;
   

   Copy

   Microsoft Windows:

   GET @mydb.public.%mytable file://C:\data\;
   

   Copy

3. Créez une zone de préparation interne nommée en utilisant CREATE STAGE qui stocke les fichiers de données avec le même format que vos fichiers Kafka source.

   Par exemple, créer une zone de préparation interne nommée kafka_json qui stocke les fichiers JSON :

   CREATE STAGE kafka_json FILE_FORMAT = (TYPE = JSON);
   

   Copy

4. Téléchargez les fichiers que vous avez téléchargés à partir de la zone de préparation de la table en utilisant PUT.

   Par exemple, téléchargez les fichiers téléchargés dans le répertoire data de votre machine locale :

   Linux ou macOS:

   PUT file:///data/ @mydb.public.kafka_json;
   

   Copy

   Microsoft Windows:

   PUT file://C:\data\ @mydb.public.kafka_json;
   

   Copy

5. Créer une table temporaire avec deux colonnes de variantes à des fins de test. La table n’est utilisée que pour valider le fichier de données mis en zone de préparation. Aucune donnée n’est chargée dans la table. La table est automatiquement détruite lorsque la session utilisateur en cours se termine :

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. Récupérez toutes les erreurs rencontrées dans le fichier de données en exécutant une instruction COPY INTO *table* … VALIDATION_MODE = “RETURN_ALL_ERRORS”. L’instruction valide le fichier dans la zone de préparation indiquée. Aucune donnée n’est chargée dans la table :

   COPY INTO mydb.public.t1
     FROM @mydb.public.kafka_json
     FILE_FORMAT = (TYPE = JSON)
     VALIDATION_MODE = 'RETURN_ALL_ERRORS';
   

   Copy

7. Corrigez toutes les erreurs signalées dans les fichiers de données sur votre machine locale.

8. Téléchargez les fichiers corrigés vers la zone de préparation de la table ou vers la zone de préparation interne nommée en utilisant PUT.

   L’exemple suivant permet de télécharger les fichiers vers la zone de préparation de la table, en écrasant les fichiers existants :

   Linux ou macOS:

   PUT file:///tmp/myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

   Windows:

   PUT file://C:\temp\myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

9. Chargez les données dans la table cible à l’aide de COPY INTO table sans l’option VALIDATION_MODE.

   Vous pouvez éventuellement utiliser l’option de copie PURGE = TRUE pour supprimer les fichiers de données de la zone de préparation une fois les données chargées correctement, ou supprimer manuellement les fichiers de la zone de préparation de table à l’aide de REMOVE :

   COPY INTO mydb.public.mytable(RECORD_METADATA, RECORD_CONTENT)
     FROM (SELECT $1:meta, $1:content FROM @mydb.public.%mytable)
     FILE_FORMAT = (TYPE = 'JSON')
     PURGE = TRUE;
   

   Copy