Problembehandlung des Kafka-Konnektors¶

Raw-Bytes¶

Der Dateiname <Konnektorname>/<Tabellenname>/<Partition>/offset_(<Schlüssel>/<Wert>_)<Zeitstempel>.gz enthält den genauen Offset des Datensatzes, der nicht von Raw-Bytes in das Quelldateiformat konvertiert wurde. Um Probleme zu lösen, senden Sie den Datensatz als neuen Datensatz erneut an den Kafka-Konnektor.

Quelldateiformat (Avro, JSON oder Protobuf)¶

Wenn Snowpipe keine Daten aus Dateien in den internen Stagingbereich laden konnte, der für das Kafka-Thema erstellt wurde, verschiebt der Kafka-Konnektor die Dateien im Quelldateiformat in den Stagingbereich, der der Zieltabelle zugeordnet ist.

Wenn ein Satz von Dateien mehrere Probleme aufweist, wird in der Spalte FIRST_ERROR_MESSAGE der COPY_HISTORY-Ausgabe nur der erste aufgetretene Fehler angezeigt. Um alle Fehler in den Dateien anzuzeigen, ist es erforderlich, die Dateien aus dem Tabellen-Stagingbereich abzurufen, sie in einen benannten Stagingbereich hochzuladen und dann eine COPY INTO <Tabelle>-Anweisung auszuführen, wobei die Kopieroption VALIDATION_MODE auf RETURN_ALL_ERRORS gesetzt sein muss. Mit der Kopieroption VALIDATION_MODE wird die COPY-Anweisung angewiesen, die zu ladenden Daten zu validieren und Ergebnisse auf Basis der angegebenen Validierungsoption zurückzugeben. Bei Angabe dieser Kopieroption werden keine Daten geladen. Verweisen Sie in der Anweisung auf die Dateien, die Sie mit dem Kafka-Konnektor zu laden versucht haben.

Wenn Probleme mit den Datendateien behoben sind, können Sie die Daten manuell mit einer oder mehreren COPY-Anweisungen laden.

Das folgende Beispiel verweist auf Datendateien im Tabellen-Stagingbereich für die Tabelle mytable in der Datenbank mydb.public und im Schema.

So validieren Sie Datendateien im Tabellen-Stagingbereich und beheben Fehler:

1. Listen Sie mit LIST alle Dateien auf, die sich im Tabellen-Stagingbereich befinden.

   Beispiel:

   LIST @mydb.public.%mytable;
   

   Copy

   Die Beispiele in diesem Abschnitt gehen davon aus, dass JSON das Quellformat für die Datendateien ist.

2. Laden Sie die vom Kafka-Konnektor erstellten Dateien mit GET auf Ihren lokalen Computer herunter.

   Laden Sie die Dateien beispielsweise in ein Verzeichnis namens data auf Ihrem lokalen Computer herunter:

   Linux oder macOS:

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

   Copy

   Microsoft Windows:

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

   Copy

3. Erstellen Sie mit CREATE STAGE einen benannten internen Stagingbereich, der die Datendateien im gleichen Format speichert wie die Kafka-Quelldateien.

   Erstellen Sie beispielsweise einen internen Stagingbereich mit dem Namen kafka_json, der JSON-Dateien speichert:

   CREATE STAGE kafka_json FILE_FORMAT = (TYPE = JSON);
   

   Copy

4. Laden Sie die Dateien, die Sie aus dem Tabellen-Stagingbereich heruntergeladen haben, mit PUT hoch.

   Laden Sie die heruntergeladenen Dateien zum Beispiel in das Verzeichnis data auf Ihrem lokalen Computer hoch:

   Linux oder macOS:

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

   Copy

   Microsoft Windows:

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

   Copy

5. Erstellen Sie zu Testzwecken eine temporäre Tabelle mit zwei Variant-Spalten. Die Tabelle wird nur zur Validierung der Staging-Datendatei verwendet. Es werden keine Daten in die Tabelle geladen. Die Tabelle wird automatisch gelöscht, wenn die aktuelle Benutzersitzung endet:

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. Rufen Sie alle in der Datendatei aufgetretenen Fehler durch Ausführen einer COPY INTO *Tabelle* … VALIDATION_MODE = ‚RETURN_ALL_ERRORS‘-Anweisung ab. Die Anweisung validiert die Datei in dem angegebenen Stagingbereich. Es werden keine Daten in die Tabelle geladen:

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

   Copy

7. Beheben Sie alle gemeldeten Fehler in den Datendateien auf Ihrem lokalen Computer.

8. Laden Sie die korrigierten Dateien mit PUT entweder in den Tabellen-Stagingbereich oder in den benannten internen Stagingbereich hoch.

   Im folgenden Beispiel werden die Dateien in den Tabellen-Stagingbereich hochgeladen, wobei die vorhandenen Dateien überschrieben werden:

   Linux oder 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. Laden Sie die Daten mit COPY INTO Tabelle ohne die Option VALIDATION_MODE in die Zieltabelle.

   Sie können optional die Kopieroption PURGE = TRUE verwenden, um die Datendateien aus dem Stagingbereich zu löschen, sobald die Daten erfolgreich geladen wurden, oder Sie löschen die Dateien manuell mithilfe von REMOVE aus dem Tabellen-Stagingbereich:

   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