Problembehandlung des Kafka-Konnektors¶

Schritt 1: COPY-Verlauf für die Tabelle anzeigen¶

Fragen Sie den Ladeaktivitätsverlauf für die Zieltabelle ab. Weitere Informationen dazu finden Sie unter Ansicht COPY_HISTORY. Wenn die COPY_HISTORY-Ausgabe keine erwarteten Dateien enthält, fragen Sie einen früheren Zeitraum ab. Wenn die Dateien Duplikate früherer Dateien waren, wurde die Aktivität möglicherweise im Ladeverlauf aufgezeichnet, als versucht wurde, die ursprünglichen Dateien zu laden. In der Spalte STATUS wird gezeigt, ob ein bestimmter Satz von Dateien geladen, teilweise geladen oder nicht geladen wurde. Die Spalte FIRST_ERROR_MESSAGE bietet einen Hinweis auf die Ursache, warum ein Ladeversuch nur teilweise erfolgreich war oder fehlgeschlagen ist.

Der Kafka-Konnektor verschiebt Dateien in den der Zieltabelle zugeordneten Stagingbereich. Die Syntax zum Verweisen auf einen Tabellen-Stagingbereich lautet @[namespace.]%table_name.

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

Beispiel:

LIST @mydb.public.%mytable;

Die Dateinamen haben eines der folgenden Formate. Die Bedingungen, die zu jedem Format führen, sind in der Tabelle beschrieben:

Die folgenden Abschnitte enthalten Anweisungen zur Lösung von Problemen mit dem jeweiligen Dateityp:

Schritt 2: Protokolldatei des Kafka-Konnektors analysieren¶

Wenn die COPY_HISTORY-Ansicht keinen Datensatz zum Datenladevorgang enthält, analysieren Sie die Protokolldatei für den Kafka-Konnektor. Der Konnektor schreibt Ereignisse in die Protokolldatei. Beachten Sie, dass der Snowflake-Kafka-Konnektor dieselbe Protokolldatei mit allen Kafka-Konnektor-Plugins teilt. Name und Speicherort dieser Protokolldatei sollten sich in Ihrer Kafka Connect-Konfigurationsdatei befinden. Weitere Informationen dazu finden Sie in der Dokumentation zu Ihrer Apache Kafka-Software.

Durchsuchen Sie die Kafka-Konnektor-Protokolldatei nach Snowflake-spezifischen Fehlermeldungen. Die meisten Nachrichten weisen die Zeichenfolge ERROR auf und enthalten den Dateinamen com.snowflake.kafka.connector..., was das Auffinden der Nachrichten erleichtert.

Folgende Fehler können möglicherweise auftreten:

Konfigurationsfehler:

Mögliche Fehlerursachen:

• Der Konnektor verfügt nicht über die richtigen Informationen, um das Thema zu abonnieren.

• Der Konnektor verfügt nicht über die richtigen Informationen zum Schreiben in die Snowflake-Tabelle (z. B. ist das Schlüsselpaar für die Authentifizierung möglicherweise falsch).

Beachten Sie, dass der Kafka-Konnektor seine Parameter überprüft. Der Konnektor gibt für jeden inkompatiblen Konfigurationsparameter einen Fehler aus. Die Fehlermeldung wird in die Protokolldatei des Kafka Connect-Clusters geschrieben. Wenn Sie von einem Konfigurationsproblem ausgehen, überprüfen Sie die Fehler in dieser Protokolldatei.

Lesefehler:

Der Konnektor konnte aus folgenden Gründen möglicherweise nicht von Kafka lesen:

• Kafka oder Kafka Connect werden möglicherweise nicht ausgeführt.

• Die Nachricht wurde möglicherweise noch nicht gesendet.

• Die Nachricht wurde möglicherweise gelöscht (abgelaufen).

Schreibfehler (Stagingbereich):

Mögliche Fehlerursachen:

• Unzureichende Berechtigungen auf der Stagingbereich.

• Der Stagingbereich hat nicht genügend Platz.

• Der Stagingbereich wurde gelöscht.

• Ein anderer Benutzer oder Prozess hat unerwartete Dateien in den Stagingbereich geschrieben.

Schreibfehler (Tabelle):

Mögliche Fehlerursachen:

• Unzureichende Berechtigungen für die Tabelle.

Schritt 3: Kafka Connect überprüfen¶

Wenn in der Kafka-Verbindungsprotokolldatei kein Fehler gemeldet wird, überprüfen Sie Kafka Connect. Anweisungen zur Problembehandlung finden Sie in der Dokumentation Ihres Apache Kafka-Softwareanbieters.

Doppelte Zeilen mit derselben Themenpartition und demselben Offset¶

Beim Laden von Daten mit Version 1.4 des Kafka-Konnektors (oder höher) können doppelte Zeilen in der Zieltabelle mit derselben Themenpartition und demselben Offset darauf hinweisen, dass der Ladevorgang das Standardausführungstimeout von 300.000 Millisekunden (300 Sekunden) überschritten hat. Überprüfen Sie die Kafka Connect-Protokolldatei auf folgenden Fehler, um die Ursache zu überprüfen:

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.

This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:1061)

Um den Fehler zu beheben, ändern Sie in der Kafka-Konfigurationsdatei (z. B. <Kafka-Verzeichnis>/config/connect-distributed.properties) eine der folgenden Eigenschaften:

consumer.max.poll.interval.ms

Erhöhen Sie das Ausführungstimeout auf 900000 (900 Sekunden).

consumer.max.poll.records

Verringern Sie die Anzahl der mit jeder Operation geladenen Datensätze auf 50.

Failure in Streaming Channel Offset Migration Response Error Code: 5023¶

Beim Upgrade auf die Konnektor-Version v2.1.0 (oder höher) wurde eine Änderung am Namensformat des Snowpipe Streaming-Kanals vorgenommen. Folglich findet die Logik, die Informationen über zuvor mit Commit bestätigte Offsets ermittelt, keine Informationen über zuvor mit Commit bestätigte Offsets. Dies äußert sich in der folgenden Ausnahme:

com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException: [SF_KAFKA_CONNECTOR] Exception: Failure in Streaming Channel Offset Migration Response Error Code: 5023

Detail: Streaming Channel Offset Migration from Source to Destination Channel has no/invalid response, please contact Snowflake Support

Message: Snowflake experienced a transient exception, please retry the migration request.

Um diesen Fehler zu beheben, fügen Sie in der Datei der Kafka-Konfiguration (z. B. <kafka_dir>/config/connect-distributed.properties) die folgende Eigenschaft der Konfiguration hinzu:

enable.streaming.channel.offset.migration

Deaktivieren Sie die automatische Offset-Migration, indem Sie sie auf false festlegen.

Konfigurieren des Konnektors zur Unterstützung verschiedener Themen¶

Wir haben ein Problem mit einer einzelnen Kafka-Konnektor-Instanz festgestellt, die eine große Anzahl von Themen unterstützt, von denen jedes mehrere Partitionen hat. Die Konfiguration des Konnektors, obwohl sie gültig zu sein schien, führte zu einem endlosen Zyklus der erneuten Abgleichung, ohne die Möglichkeit, Daten in Snowflake zu importieren. Das Problem betraf den Snowpipe Streaming-Modus (snowflake.ingestion.method=SNOWPIPE_STREAMING), aber die Richtlinien gelten auch für den Snowpipe Modus (snowflake.ingestion.method=SNOWPIPE). Das Problem manifestiert sich in der Datei dadurch, dass diese Protokollmeldung wiederholt aufgezeichnet wird:

[Worker-xyz] [timestamp] INFO [my-connector|task-id] [SF_INGEST] Channel is marked as closed

Dies kann typischerweise passieren, wenn Sie Ihren Konnektor so konfigurieren, dass er Themen über Regex einliest. Wir empfehlen die Anwendung der folgenden Optionen auf die Kafka-Konfigurationsdatei (z. B. <kafka_dir>/config/connect-distributed.properties):

consumer.override.partition.assignment.strategy

Konfigurieren Sie die Strategie für die Zuweisung von Partitionen an Tasks als org.apache.kafka.clients.consumer.CooperativeStickyAssignor - dies führt zu einer gleichmäßigen Verteilung der aufgenommenen Kanäle auf die verfügbaren Aufgaben und verringert das Risiko einer Neuverteilung.

tasks.max

Die Anzahl der instanziierten Aufgaben pro Konnektor sollte die Anzahl der verfügbaren CPUs nicht überschreiten - der zugrunde liegende Treiber implementiert einen Drosselungsmechanismus auf der Basis der verfügbaren CPUs. Eine steigende Anzahl paralleler Anfragen erhöht nicht nur den Druck auf den Speicher Ihres Systems, sondern führt auch zu längeren Verarbeitungszeiten für Einsätze, was direkt zu fehlenden Heartbeats des Konnektors führt.

Timeout Wenn es um die Timeout-Werte des Konnektors geht, gibt es eine Reihe von Eigenschaftseigenschaften, die diese direkt beeinflussen:

consumer.override.heartbeat.interval.ms

Defines how often the monitor thread (there is one associated with each task) will send heartbeat to Kafka. Die Voreinstellung ist 3000 ms, aber bei höherer Systembelastung können Sie mit einer Erhöhung auf 5000 ms experimentieren.

consumer.override.session.timeout.ms

Legt fest, wie lange der Broker wartet, bevor er davon annimmt, dass sich der Verbraucher in einem ungültigen Zustand befindet und versucht, den Abgleich erneut durchzuführen. Diese Einstellung sollte in der Regel 3-mal höher sein als das Heartbeat-Intervall. Wenn Sie also das Heartbeat-Intervall auf 5000 ms festgelegt haben, legen Sie diese Einstellung auf 15000 ms fest.

consumer.override.max.poll.interval.ms

Definiert das maximale Intervall zwischen Aufrufen von poll() aus zugrunde liegendem Kafka. Die Zeit zwischen den Abfragen wird im Wesentlichen dem Konnektor zugeordnet, der einen Batch von Daten verarbeitet (einschließlich Hochladen in Snowflake und Bestätigen durch Commit). In Szenarios, in denen mehrere Aufgaben Daten verarbeiten, kann die zugrunde liegende Snowflake-Verbindung beginnen, Anfragen zu drosseln, was zu längeren Verarbeitungszeiten führt. Abhängig von Ihrem Szenario können Sie diesen Wert sogar auf 20 Minuten (1200000 ms) erhöhen - vor allem, wenn Sie den Konnektor mit einer großen Anzahl von Datensätzen starten, die aufgenommen werden sollen.

consumer.override.rebalance.timeout.ms

Wenn der Abgleich in einem Szenario mit einer großen Anzahl von Kanälen pro Aufgabe erneut durchgeführt wird, gibt es eine Menge zugrunde liegender Logik pro Kanal, um herauszufinden, wo die Verarbeitung fortgesetzt werden soll. Dieser Code wird sequenziell ausgeführt. Je mehr Kanäle pro Aufgabe, desto länger dauert das anfängliche Setup. Konfigurieren Sie diese Eigenschaft so, dass der Wert groß genug ist, damit jeder Kanal seine Initialisierung abschließen kann. Ein Wert von 3 Minuten (180000 ms) ist ein guter Ausgangspunkt.

Es ist auch wichtig, auf den verfügbaren Heap-Speicher für den Konnektor zu achten. Dies ist besonders wichtig in Szenarios, in denen mehrere Konnektoren gleichzeitig laufen oder ein Konnektor Daten aus mehreren Themen aufnimmt. Jede Partition eines Themas ist einem einzelnen Kanal zugeordnet und benötigt daher Speicher.

Stellen Sie sicher, dass Sie die Einstellungen für den Arbeitsspeicher Ihres Kafka-Verbindungsprozesses über die Xmx-Einstellung anpassen. Eine Möglichkeit, dies zu tun, besteht darin, die Umgebungsvariable KAFKA_OPTS zu definieren und entsprechend zu konfigurieren (KAFKA_OPTS=-Xmx4G).

Datei-Cleaner löscht Dateien unerwartet¶

Wenn Sie den Kafka-Konnektor mit SNOWPIPE verwenden, kann es zu einem Problem kommen, wenn Sie Daten aus mehreren Themen in eine einzige Tabelle einlesen. Wenn Ihre Konfiguration den Eintrag snowflake.topic2table.map nicht enthält oder es eine 1:1-Zuordnung zwischen dem Thema und der Tabelle gibt, ist dieses Problem nicht relevant.

Der Kafka-Konnektor erzeugt Dateien mit Datensätzen, die in einen Stagingbereich hochgeladen werden sollen. Diese Dateien sind nach dem folgenden Muster formatiert: snowflake_kafka_connector_<connector-name>_stage_<table-name>/<connector-name>/<table-name>/<partition-id>/<low-watermark>_<high-watermark>_<timestamp>.json.gz. Das Problem liegt im Speicherort <partition-id>: Wenn mehrere Themen Daten in eine einzige Tabelle geladen werden, sind Duplikate im Wert partition-id wahrscheinlich. Dies ist bei einer normalen Konnektoroperation kein Problem. Wenn der Konnektor jedoch neu gestartet oder neu abgeglichen wird, könnte der Cleaner-Prozess die in den Stagingbereich geladenen (aber noch nicht eingelesenen) Dateien fälschlicherweise der falschen Partition zuordnen und beschließen, sie zu löschen, was zu einem Datenverlust führen könnte.

Der Konnektor mit der Version 2.4.x korrigiert dieses Problem, indem er den Hashcode des Ausgangsthemas in die partition-id einfügt, um eindeutige Dateinamen zu gewährleisten, die genau mit der Partition eines einzelnen Themas übereinstimmen. Diese Korrektur ist standardmäßig aktiviert – snowflake.snowpipe.stageFileNameExtensionEnabled – und betrifft nur Konfigurationen, bei denen eine Zieltabelle mehr als einmal unter snowflake.topic2table.map aufgeführt ist.

Wenn Ihre Konfiguration von dieser Funktionalität betroffen ist, kann es vorkommen, dass veraltete Dateien in Ihren Stagingbereich hochgeladen werden. Wenn der Konnektor startet, prüft er, ob Ihr Stagingbereich solche Dateien enthält. Sie müssen nach den Protokolleinträgen suchen, die mit NOTE: For table beginnen, gefolgt von der Liste der erkannten Dateien.

Sie können auch manuell prüfen, ob einige Dateien im Stagingbereich betroffen sind:

1. Suchen Sie den betroffenen Stagingbereich:

   show stages like 'snowflake_kafka_connector%<your table name>';
   

   Copy

2. Liste der Stagingbereich-Dateien:

   list @<your stage name> pattern = '.+/<your-table-name>/[0-9]{1,4}/[0-9]+_[0-9]+_[0-9]+\.json\.gz$';
   

   Copy

Der obige Befehl listet alle Dateien auf, die dem Stagingbereich Ihrer Tabelle entsprechen und deren Partition-IDs im Bereich 0–9999 liegen. Diese Dateien werden nicht mehr importiert, so dass Sie sie herunterladen oder löschen können.