Snowflake Connector for Kafka: Installation und Konfiguration¶

Kafka-Konnektordateien herunterladen¶

Laden Sie die JAR-Datei des Kafka-Konnektors von einem der folgenden Speicherorte herunter:

Confluent Hub:

https://www.confluent.io/hub/

Das Paket enthält alle Abhängigkeiten, die zur Verwendung eines verschlüsselten oder unverschlüsselten privaten Schlüssels für die Schlüsselpaar-Authentifizierung erforderlich sind. Weitere Informationen dazu finden Sie unter Verwenden von Schlüsselpaar-Authentifizierung und Schlüsselrotation (unter diesem Thema).

Zentrales Maven-Repository:

https://mvnrepository.com/artifact/com.snowflake

Wenn Sie diese Version verwenden, müssen Sie die Bouncy Castle-Kryptographie-Bibliotheken (JAR-Dateien) herunterladen:

• https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

• https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

Laden Sie diese Dateien in denselben lokalen Ordner wie die JAR-Datei des Kafka-Konnektors herunter.

Der Quellcode des Konnektors ist unter https://github.com/snowflakedb/snowflake-kafka-connector verfügbar.

Kafka-Konnektor installieren¶

Installieren Sie den Kafka-Konnektor gemäß den Anweisungen, die für die Installation anderer Konnektoren bereitgestellt wurden:

https://docs.confluent.io/current/connect/userguide.html

Apache Kafka installieren¶

1. Laden Sie das Kafka-Paket von der offiziellen Kafka-Website herunter.

2. Wechseln Sie in einem Terminalfenster in das Verzeichnis, in das Sie die Paketdatei heruntergeladen haben.

3. Führen Sie den folgenden Befehl aus, um die Datei kafka_<scala_version>-<kafka_version>.tgz zu dekomprimieren:

   CopyExpand

   tar xzvf kafka_<scala_version>-<kafka_version>.tgz
   

   Show lessSee more

   Scroll to top

JDK installieren¶

Installieren und konfigurieren Sie das Java Development Kit (JDK) Version 11 oder höher. Snowflake wurde mit der Standard Edition (SE) des JDK getestet. Die Enterprise Edition (EE) ist voraussichtlich kompatibel, wurde jedoch nicht getestet.

Wenn Sie bereits zuvor das JDK installiert haben, können Sie diesen Abschnitt überspringen.

1. Laden Sie das JDK von der Oracle JDK-Website herunter.

2. Installieren oder dekomprimieren Sie das JDK.

3. Stellen Sie gemäß den Anweisungen für Ihr Betriebssystem die Umgebungsvariable JAVA_HOME so ein, dass sie auf das Verzeichnis mit dem JDK verweist.

JAR-Dateien des Kafka-Konnektors herunterladen¶

1. Laden Sie die JAR-Datei des Kafka-Konnektors vom Maven Central Repository herunter:

   https://mvnrepository.com/artifact/com.snowflake

2. Laden Sie die JAR-Dateien für die Bouncy Castle-Kryptographie-Bibliothek herunter:

   • https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

   • https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

3. Wenn Ihre Kafka-Daten im Apache Avro-Format gestreamt werden, laden Sie die Avro-JAR-Datei (1.11.4) herunter:

   • https://mvnrepository.com/artifact/org.apache.avro/avro

Der Quellcode des Konnektors ist unter https://github.com/snowflakedb/snowflake-kafka-connector verfügbar.

Kafka-Konnektor installieren¶

Kopieren Sie die JAR-Dateien, die Sie in Installieren des Konnektors für Open Source Apache Kafka heruntergeladen haben, in den Ordner <kafka_dir>/libs.

Schematisierungs- und Validierungseigenschaften¶

Diese Eigenschaften steuern, wie der Konnektor Daten verarbeitet und validiert. Für neue Installationen funktionieren die Standardwerte gut. Wenn Sie von v3 migrieren, lesen Sie Migration vom Kafka-Konnektor v3 auf v4, um Hinweise zu erhalten, welche Werte verwendet werden sollten.

snowflake.enable.schematization

Steuert, ob eingehende Datensätze in einzelne Tabellenspalten schematisiert oder in Legacy-VARIANT-Spalten eingeschlossen werden.

Wenn true (Standard), werden Datensatzfelder einzelnen Tabellenspalten über Namen zugeordnet. Wenn false, speichert der Konnektor Datensätze in zwei VARIANT-Spalten (RECORD_CONTENT und RECORD_METADATA), was dem Verhalten von v3 entspricht.

Standard:

true

snowflake.validation

Steuert, wo Datenvalidierung und Schemaentwicklung durchgeführt werden.

server_side (Standard): Die Validierung wird vom Snowflake-Backend ausgeführt, was dem COPY- und Snowpipe-Verhalten entspricht. Ungültige Datensätze werden in Fehlertabellen erfasst. Unterstützt sowohl den Standard-Pipe- als auch den benutzerdefinierten Pipe-Modus.

client_side: Der Konnektor überprüft Datentypen und Schemakompatibilität, bevor Zeilen an Snowflake gesendet werden. Unterstützt eine Warteschlange für nicht zustellbare Meldungen (DLQ) für ungültige Datensätze. Funktioniert nur mit dem Standard-Pipe-Modus.

Weitere Details dazu finden Sie unter Validierung und Fehlerbehandlung.

Standard:

server_side

Migrations- und Kompatibilitätseigenschaften¶

Diese Eigenschaften sind relevant bei der Migration von v3. Bei Neuinstallationen können Sie diese überspringen und snowflake.streaming.validate.compatibility.with.classic=false einstellen.

snowflake.compatibility.enable.autogenerated.table.name.sanitization

Steuert, wie automatisch generierte Tabellennamen von Themennamen abgeleitet werden.

Wenn false (die Standardeinstellung), werden Themennamen unverändert für Tabellennamen verwendet, wobei Groß- und Kleinschreibung und Sonderzeichen beibehalten werden. Tabellennamen werden als Bezeichner in Anführungszeichen erstellt.

Wenn``true``, werden ungültige Zeichen in Snowflake-Bezeichnern durch Unterstriche ersetzt, Namen werden in Großbuchstaben geschrieben und ein Hash-Code wird zur Eindeutigkeit angehängt. Dies entspricht dem Verhalten von v3.

Standard:

false

snowflake.compatibility.enable.column.identifier.normalization

Steuert, wie Spaltenbezeichner behandelt werden.

Wenn false (die Standardeinstellung), behalten Spaltenbezeichner die Groß-/Kleinschreibung und Sonderzeichen unverändert bei.

Wenn true, werden Spaltenbezeichner auf Großbuchstaben normalisiert, was dem Verhalten von v3 entspricht.

Standard:

false

snowflake.streaming.validate.compatibility.with.classic

Aktiviert eine Startvalidierung, die prüft, ob alle migrationsbezogenen Konfigurationen explizit festgelegt sind. Wenn true, schlägt der Konnektor beim Start mit einem beschreibenden Fehler fehl, wenn eine der folgenden Konfigurationen fehlt oder mit dem Verhalten von v3 nicht kompatibel ist:

• snowflake.validation muss client_side sein.

• snowflake.compatibility.enable.column.identifier.normalization muss true sein.

• snowflake.compatibility.enable.autogenerated.table.name.sanitization muss true sein.

• snowflake.enable.schematization muss explizit auf true oder false gesetzt sein (der Standard wurde zwischen v3 und v4 geändert).

• snowflake.streaming.classic.offset.migration muss explizit festgelegt sein.

• snowflake.streaming.classic.offset.migration.include.connector.name muss explizit festgelegt sein (wenn die Offset-Migration strict oder best_effort ist).

Dies verhindert, dass versehentlich v4 mit einer kopierten v3-Konfiguration ausgeführt wird, ohne die grundlegenden Änderungen zu überprüfen. Legen Sie false fest, um diese Prüfung zu überspringen, nachdem Sie Ihre Konfiguration bestätigt haben.

Bemerkung

Bei Neuinstallationen (keine Migration von v3) setzen Sie den Wert auf false. Der Kompatibilitätsvalidierer wird nur benötigt, wenn ein Upgrade von einer bestehenden v3-Bereitstellung ausgeführt wird.

Weitere Informationen dazu finden Sie unter Migration vom Kafka-Konnektor v3 auf v4.

Standard:

true

Eigenschaften der Offset-Migration¶

Diese Eigenschaften steuern, wie v4 bestätigte Offsets von v3 Snowpipe Streaming (SSv1)-Kanälen migriert. Diese sind nur relevant, wenn Sie von einem v3-Konnektor migrieren, der snowflake.ingestion.method=SNOWPIPE_STREAMING verwendet hat. Wenn Sie vom v3-Snowpipe-Modus (dateibasierte Datenaufnahme) migrieren, setzen Sie snowflake.streaming.classic.offset.migration auf skip.

snowflake.streaming.classic.offset.migration

Steuert, wie v4 Offsets von v3 Snowpipe Streaming (SSv1)-Kanälen migriert.

strict: v4 sucht den bestätigten Offset des v3 SSv1-Kanals und wird ab diesem Punkt fortgesetzt. Wenn der SSv1-Kanal nicht gefunden wird, schlägt der Konnektor mit einem Fehler fehl.

best_effort: v4 versucht, den bestätigten Offset des v3 SSv1-Kanals zu ermitteln. Wenn der Kanal nicht gefunden wird, greift v4 auf den Offset der Kafka-Verbrauchergruppe zurück.

skip (Standard): Es wird keine SSv1-Offset-Migration durchgeführt. v4 verwendet den Offset der Kafka-Verbrauchergruppe. Verwenden Sie dies, wenn Sie vom v3 Snowpipe-Modus (nicht Snowpipe Streaming) migrieren.

Standard:

skip

snowflake.streaming.classic.offset.migration.include.connector.name

Steuert, ob die SSv1-Kanalnamensuche den Namen des Konnektors enthält. Dieser muss mit der Konfiguration Ihres v3-Konnektors übereinstimmen. In v3 steuerte die Eigenschaft snowflake.streaming.channel.name.include.connector.name, ob der Name des Konnektors im Kanalnamen enthalten ist.

Auf true setzen, wenn Ihr v3-Konnektor snowflake.streaming.channel.name.include.connector.name=true aufwies oder wenn Sie den Kafka-Konnektor der Version 2.1.0 oder 2.1.1 ausführten (diese Versionen enthielten standardmäßig den Namen des Konnektors). Andernfalls auf false setzen.

Nur erforderlich, wenn snowflake.streaming.classic.offset.migration auf strict oder best_effort gesetzt ist.

Standard:

Keine (muss explizit festgelegt werden, wenn die Offset-Migration aktiv ist)

Fehlerbehandlung von Eigenschaften¶

errors.tolerance

Steuert, wie der Konnektor auf Fehler während der Erfassung reagiert.

none (Standard): Die Konnektor-Aufgabe schlägt beim ersten Fehler fehl. Bei der serverseitigen Validierung ist die Fehlererkennung asynchron, sodass einige Datensätze nach dem fehlerhaften noch erfasst werden können, bevor die Aufgabe fehlschlägt.

all: Der Konnektor nimmt weiterhin Daten auf. Bei der clientseitigen Validierung werden ungültige Datensätze an die DLQ weitergeleitet (falls konfiguriert) oder ohne Meldung gelöscht.

Warnung

Die Einstellung von errors.tolerance=all ohne Konfiguration eines DLQ-Themas führt dazu, dass ungültige Datensätze ohne Meldung gelöscht werden, wenn die clientseitige Validierung verwendet wird. Dies kann zu Datenverlusten führen.

Standard:

none

errors.deadletterqueue.topic.name

Der Name des Kafka-Themas für die Warteschlange für nicht zustellbare Meldungen (DLQ). Nur wirksam, wenn snowflake.validation=client_side und errors.tolerance=all.

Standard:

leer (DLQ deaktiviert)

errors.log.enable

Wenn true, werden Fehler mit Details zur fehlgeschlagenen Operation und den Datensatzeigenschaften protokolliert.

Standard:

false

enable.task.fail.on.authorization.errors

Wenn true, schlägt die Konnektoraufgabe bei Autorisierungsfehlern von Snowflake sofort fehl. Wenn false, versucht der Konnektor es erneut.

Standard:

false

Zwischenspeichern von Eigenschaften¶

snowflake.cache.table.exists

Aktiviert das Caching für das Vorhandensein von Tabellen, wodurch die Anzahl der Metadaten-Abfragen an Snowflake reduziert wird.

Standard:

true

snowflake.cache.table.exists.expire.ms

Cache-Ablaufzeit in Millisekunden für die Überprüfung auf das Vorhandensein von Tabellen.

Standard:

300000 (5 Minuten)

snowflake.cache.pipe.exists

Aktiviert das Caching für die Überprüfung auf das Vorhandensein von Pipes.

Standard:

true

snowflake.cache.pipe.exists.expire.ms

Cache-Ablaufzeit in Millisekunden für die Überprüfung auf das Vorhandensein von Pipes.

Standard:

300000 (5 Minuten)

Überwachungs- und Diagnoseeigenschaften¶

jmx

Aktiviert JMX MBeans für Konnektor-Metriken. Weitere Informationen dazu finden Sie unter Monitor the Snowflake Connector for Kafka.

Standard:

true

enable.mdc.logging

Aktiviert MDC (Mapped Diagnostic Kontext), um den Konnektorkontext den Protokollmeldungen voranzustellen, was nützlich ist, wenn mehrere Konnektor-Instanzen ausgeführt werden.

Standard:

false

snowflake.streaming.metadata.connectorPushTime

Wenn true, ist der SnowflakeConnectorPushTime-Zeitstempel in RECORD_METADATA enthalten. Dieses Feld erfasst, wann der Konnektor einen Datensatz zur Erfassung gepuffert hat, und ist nützlich, um die End-to-End-Latenz abzuschätzen.

Standard:

true

Erweiterte Eigenschaften¶

snowflake.streaming.client.provider.override.map

Überschreibt die Eigenschaften von Snowpipe Streaming-Clients. Format: key1:value1,key2:value2. Verwenden Sie diese Option nur nach Rücksprache mit dem Snowflake-Support.

Standard:

empty

Andere Eigenschaften¶

snowflake.private.key.passphrase

Wenn der Wert dieses Parameters nicht leer ist, verwendet Kafka diesen Ausdruck, um zu versuchen, den privaten Schlüssel zu entschlüsseln.

tasks.max

Anzahl der Aufgaben. Entspricht in der Regel der Zahl der CPU-Kerne, die auf die Workerknoten im Kafka Connect-Cluster verteilt sind. Um eine optimale Leistung zu erzielen, empfiehlt Snowflake, die Anzahl der Tasks auf die Gesamtzahl der Kafka-Partitionen festzulegen, jedoch nicht die Anzahl der CPU-Kerne zu überschreiten. Eine hohe Anzahl von Tasks kann zu einem erhöhten Arbeitsspeicherverbrauch und häufigen Anpassungen führen.

snowflake.topic2table.map

Durch Kommas getrennte Liste der Zuordnungen von Themen zu Tabellen im topic:table-Format. Unterstützt Regex-Muster für Themennamen. Die regulären Ausdrücke dürfen nicht mehrdeutig sein – jedes abgeglichene Thema darf nur mit einer einzigen Zieltabelle übereinstimmen.

Sowohl Themen- als auch Tabellennamen können in doppelte Anführungszeichen gesetzt werden, um Sonderzeichen (Doppelpunkte, Kommas, Leerzeichen) zu unterstützen. Tabellennamen, die nicht in Anführungszeichen gesetzt sind, werden in Großbuchstaben geschrieben. Bei Tabellennamen in Anführungszeichen wird die Groß-/Kleinschreibung beibehalten.

Detaillierte Beispiele, einschließlich Regex-Muster, Viele-zu-eins-Zuordnungen und Anführungszeichen, finden Sie unter :ref:` Explizite Zuordnung von Themen zu Tabellen<label-kafkahp_explicit_topic_to_table_mapping>`.

Beispiel:

CopyExpand

snowflake.topic2table.map=topic1:low_range,topic2:low_range,"my:topic":"My_Table"

Show lessSee more

Scroll to top

value.converter.schema.registry.url

Wenn Sie das Avro-Format und einen Schema-Registrierungsdienst verwenden, sollte dies die URL des Schema-Registrierungsdienstes sein. Andernfalls sollte das Feld leer sein.

value.converter.break.on.schema.registry.error

Wenn Sie Avro-Daten aus dem Schema-Registrierungsdienst laden, bestimmt diese Eigenschaft, ob der Kafka-Konnektor keine Datensätze mehr verbrauchen soll, wenn beim Abrufen der Schema-ID ein Fehler auftritt. Der Standardwert ist false. Setzen Sie den Wert auf true, um dieses Verhalten zu aktivieren.

jvm.proxy.host

Damit der Snowflake-Kafka-Konnektor über einen Proxyserver auf Snowflake zugreifen kann, legen Sie diesen Parameter fest, um den Host dieses Proxyservers anzugeben.

jvm.proxy.port

Damit der Snowflake-Kafka-Konnektor über einen Proxyserver auf Snowflake zugreifen kann, legen Sie diesen Parameter fest, um den Port dieses Proxyservers anzugeben.

jvm.proxy.username

Benutzername, der sich beim Proxyserver authentifiziert.

jvm.proxy.password

Kennwort für den Benutzernamen, der sich beim Proxyserver authentifiziert.

snowflake.jdbc.map

Beispiel: "snowflake.jdbc.map": "networkTimeout:20,tracing:WARNING"

Weitere JDBC-Eigenschaften (siehe Übersicht der Verbindungsparameter für den JDBC-Treiber) werden nicht validiert. Diese zusätzlichen Eigenschaften werden nicht validiert und dürfen weder überschrieben noch anstelle der erforderlichen Eigenschaften verwendet werden, wie z. B: jvm.proxy.xxx, snowflake.user.name, snowflake.private.key, snowflake.schema.name und ähnliche Eigenschaften.

Geben Sie eine der folgenden Kombinationen an:

• tracing-Eigenschaft zusammen mit der Variable JDBC_TRACE env

• database-Eigenschaft zusammen mit snowflake.database.name

Führt zu einem mehrdeutigen Verhalten und das Verhalten wird durch den JDBC-Treiber bestimmt.

value.converter.basic.auth.credentials.source

Wenn Sie das Avro-Datenformat verwenden und einen sicheren Zugriff auf die Kafka-Schemaregistrierung benötigen, setzen Sie diesen Parameter auf die Zeichenfolge „USER_INFO“, und geben Sie einen Wert für den unten beschriebenen Parameter value.converter.basic.auth.user.info an. Andernfalls lassen Sie diesen Parameter aus.

value.converter.basic.auth.user.info

Wenn Sie das Avro-Datenformat verwenden und einen sicheren Zugriff auf die Kafka-Schemaregistrierung benötigen, setzen Sie diesen Parameter auf die Zeichenfolge „<Benutzer-ID>:<Kennwort>“, und geben Sie einen Wert für den oben beschriebenen Parameter value.converter.basic.auth.credentials.source an. Andernfalls lassen Sie diesen Parameter aus.

snowflake.metadata.createtime

Wenn der Wert auf FALSE festgelegt ist, wird der Eigenschaftswert CreateTime in den Metadaten in der Spalte RECORD_METADATA weggelassen. Der Standardwert ist TRUE.

snowflake.metadata.topic

Wenn der Wert auf FALSE festgelegt ist, wird der Eigenschaftswert topic in den Metadaten in der Spalte RECORD_METADATA weggelassen. Der Standardwert ist TRUE.

snowflake.metadata.offset.and.partition

Wenn der Wert auf FALSE festgelegt ist, werden die Eigenschaftswerte Offset und Partition in den Metadaten in der Spalte RECORD_METADATA weggelassen. Der Standardwert ist TRUE.

snowflake.metadata.all

Wenn der Wert auf FALSE gesetzt ist, sind die Metadaten in der Spalte RECORD_METADATA vollständig leer. Der Standardwert ist TRUE.

transforms

Geben Sie an, dass vom Kafka-Konnektor festgestellte Tombstone-Datensätze übersprungen und nicht in die Zieltabelle geladen werden sollen. Ein Tombstone-Datensatz ist definiert als ein Datensatz, bei dem das gesamte Wertefeld null ist.

Setzen Sie den Eigenschaftswert auf "tombstoneHandlerExample".

Bemerkung

Verwenden Sie diese Eigenschaft nur mit den Kafka Community-Konvertern (d. h. value.converter-Eigenschaftswert) (z. B. org.apache.kafka.connect.json.JsonConverter oder org.apache.kafka.connect.json.AvroConverter). Um die Behandlung von Tombstone-Datensätzen mit den Snowflake-Konvertern zu verwalten, verwenden Sie stattdessen die Eigenschaft behavior.on.null.values.

transforms.tombstoneHandlerExample.type

Erforderlich beim Einstellen der transforms-Eigenschaft.

Setzen Sie den Eigenschaftswert auf "io.confluent.connect.transforms.TombstoneHandler"

behavior.on.null.values

Geben Sie an, wie der Kafka-Konnektor Tombstone-Datensätze behandeln soll. Ein Tombstone-Datensatz ist definiert als ein Datensatz, bei dem das gesamte Wertefeld null ist. Für Snowpipe wird diese Eigenschaft von der Kafka-Konnektor-Version 1.5.5 und höher unterstützt. Für Snowpipe Streaming wird diese Eigenschaft von der Kafka-Konnektor-Version 2.1.0 und höher unterstützt.

Diese Eigenschaft unterstützt die folgenden Werte:

DEFAULT

Wenn der Kafka-Konnektor auf einen Tombstone-Datensatz stößt, fügt er in die Inhaltsspalte eine leere JSON-Zeichenfolge ein.

IGNORE

Der Kafka-Konnektor überspringt Tombstone-Datensätze und fügt keine Zeilen für diese Datensätze ein.

Der Standardwert ist DEFAULT.

Bemerkung

Die Erfassung von Tombstone-Datensätzen variiert je nach Erfassungsmethode:

• Für Snowpipe verwendet der Kafka-Konnektor nur Snowflake-Konverter. Um die Behandlung von Tombstone-Datensätzen mit den Kafka-Community-Konvertern zu verwalten, verwenden Sie stattdessen die Eigenschaften transform und transforms.tombstoneHandlerExample.type.

• Für Snowpipe Streaming verwendet der Kafka-Konnektor ausschließlich Community-Konverter.

An Kafka-Broker gesendete Datensätze dürfen nicht NULL sein, da diese Datensätze vom Kafka-Konnektor verworfen werden, was zu fehlenden Offsets führt. Die fehlenden Offsets machen den Kafka-Konnektor in bestimmten Anwendungsfällen funktionsunfähig. Es wird empfohlen, dass Sie anstelle von NULL-Datensätzen Tombstone-Datensätze verwenden.

Externalisieren von Geheimnissen¶

Snowflake empfiehlt dringend, Geheimnisse wie den privaten Schlüssel zu externalisieren und in verschlüsselter Form bzw. in einem Schlüsselverwaltungsdienst wie AWS KMS, Azure Key Vault oder HashiCorp Vault zu speichern. Dies kann mithilfe einer ConfigProvider-Implementierung in Ihrem Kafka Connect-Cluster erreicht werden.

Weitere Informationen dazu finden Sie in der Beschreibung des Confluent-Dienstes.