Übersicht: Connectors

In der folgenden Tabelle sind die Kafka Connect-Connectortypen aufgeführt, die von Managed Service for Apache Kafka unterstützt werden. Mit diesen Connectors können Sie Apache Kafka in Ihre Anwendungen und in andere Google Cloud-Dienste einbinden.

Connector Beschreibung Anwendungsfälle
BigQuery-Senke Streamt Daten aus Kafka-Themen in eine BigQuery-Tabelle. Data-Warehouse-Prozess, Analysen
Cloud SQL for PostgreSQL-Quelle Liest Änderungen auf Zeilenebene aus einer Cloud SQL for PostgreSQL-Datenbank und schreibt sie in Kafka. Datenbankintegration, Überwachung von Datenbankereignissen und Reaktion darauf
Cloud Storage-Senke Streamt Daten aus Kafka-Themen in einen Cloud Storage-Bucket. Data-Lake-Aufnahme, Datenarchivierung
Allgemeine PostgreSQL-Quelle Liest Änderungen auf Zeilenebene aus einer PostgreSQL-Datenbank und schreibt sie in Kafka. Datenbankintegration, Überwachung von Datenbankereignissen und Reaktion darauf
MirrorMaker 2.0 (Prüfpunkt) Kopiert Nutzer-Offsets von einem Kafka-Cluster in einen anderen. Datenreplikation, Notfallwiederherstellung, Datenmigration
MirrorMaker 2.0 Heartbeat Generiert regelmäßige Heartbeat-Nachrichten und schreibt sie in ein Thema Überwachung des Clusterzustands, der Netzwerkverbindung oder der End-to-End-Replikationslatenz zwischen Clustern.
MirrorMaker 2.0 (Quelle) Repliziert Themen und Daten von einem Kafka-Cluster in einen anderen Kafka-Cluster. Datenreplikation, Notfallwiederherstellung, Datenmigration
Pub/Sub-Senke Streamt Daten aus Kafka-Themen in ein Pub/Sub-Thema. Dienstintegration, Echtzeitbenachrichtigungen
Pub/Sub-Quelle Streamt Nachrichten aus einem Pub/Sub-Abo in ein Kafka-Thema. Echtzeit-Datenaufnahme, ereignisgesteuerte Architekturen

Nutzer mit Conversion

Konverter sind für die Serialisierung und Deserialisierung von Kafka-Eintragsdaten verantwortlich. Sie übersetzen zwischen dem Rohbyteformat in Kafka-Themen und der internen, strukturierten Datendarstellung, die von Kafka Connect verwendet wird.

  • Bei Senken-Connectors deserialisieren Konverter Daten aus dem Wire-Format des Themas in das interne Datenformat von Kafka Connect, das der Connector zum Schreiben in das Zielsystem verwendet.

  • Bei Quell-Connectors serialisieren Konverter die Daten aus dem internen Datenformat von Kafka Connect in das angegebene Wire-Format für das Kafka-Thema.

Konverter sorgen dafür, dass der Connector die Kafka-Datensätze in einem Format liest oder schreibt, das mit dem externen System kompatibel ist.

Legen Sie beim Konfigurieren eines Connectors die folgenden Attribute fest:

  • Schlüsselkonverter (key.converter): Der Konverter, der beim Serialisieren und Deserialisieren von Kafka-Eintragsschlüsseln verwendet werden soll.

  • Wertkonverter (value.converter): Der Konverter, der beim Serialisieren und Deserialisieren von Kafka-Eintragswerten verwendet werden soll.

Wenn Sie keinen Konverter angeben, ist der Standardkonvertertyp org.apache.kafka.connect.converters.ByteArrayConverter. Dadurch werden die Daten im Rohbyteformat weitergegeben.

Unterstützte Konverter

Managed Service for Apache Kafka unterstützt die folgenden integrierten Konverter:

ConverterFormat
io.confluent.connect.avro.AvroConverter Apache Avro
org.apache.kafka.connect.converters.BooleanConverter Boolesch
org.apache.kafka.connect.converters.ByteArrayConverter

Byte-Array

Der Standardkonvertertyp. Der genaue Inhalt von Nachrichten wird über zwei Systeme hinweg beibehalten.
org.apache.kafka.connect.converters.DoubleConverter Doppelt
org.apache.kafka.connect.converters.FloatConverter Float
org.apache.kafka.connect.converters.IntegerConverter Ganzzahl
org.apache.kafka.connect.json.JsonConverter

JSON

Legen Sie für JSON-Daten ohne Schema auch value.converter.schemas.enable=false fest.
org.apache.kafka.connect.converters.LongConverter Lang
org.apache.kafka.connect.converters.ShortConverter Kurz
org.apache.kafka.connect.storage.StringConverter String

Die Wahl des Konverters hängt vom Connector-Typ und den Daten ab, die Sie in Kafka speichern. Weitere Informationen finden Sie in der Dokumentation zum jeweiligen Connector.

Tasks

Ein Connector überträgt Daten, indem er eine oder mehrere Aufgaben erstellt, die parallel ausgeführt werden. Wenn Sie eine Obergrenze für die Anzahl der Aufgaben festlegen möchten, die ein Connector erstellt, legen Sie das Konfigurationsattribut tasks.max des Connectors fest. Der Connector erstellt möglicherweise weniger Aufgaben als diesen Wert.

Wenn Sie den Wert von tasks.max erhöhen, kann sich der Durchsatz verbessern, aber auch der Ressourcenverbrauch (CPU und Arbeitsspeicher) steigen. Der optimale Wert hängt von der Arbeitslast und den Ressourcen ab, die Ihren Connect-Cluster-Workern zugewiesen sind. Bei Senken-Connectors kann die Anzahl der Kafka-Themenpartitionen auch die Parallelität beeinflussen.

Richtlinie für Task-Neustart

Sie können die Richtlinie für den Neustart von Aufgaben eines Connectors festlegen, die das Verhalten bei einem Fehler bestimmt. Connectors unterstützen die folgenden Richtlinien:

  • Nie neu starten Der Connector startet fehlgeschlagene Tasks nicht neu. Diese Richtlinie ist das Standardverhalten. Das ist nützlich für die Fehlerbehebung oder in Situationen, in denen nach einem Fehler ein manueller Eingriff erforderlich ist.

  • Mit exponentiellem Backoff neu starten. Der Connector startet eine fehlgeschlagene Aufgabe nach einer Verzögerung (dem sogenannten Backoff-Zeitraum) neu. Die Verzögerung erhöht sich mit jedem nachfolgenden Fehler exponentiell. Diese Richtlinie wird für die meisten Produktionsarbeitslasten empfohlen.

    Wenn Sie die exponentielle Backoff-Richtlinie verwenden, legen Sie auch Werte für den minimalen und maximalen Backoff fest. Der Mindest-Backoff sollte größer als 60 Sekunden und der maximale Backoff kleiner als 7.200 Sekunden sein.

Transformationen und Prädikate

Managed Service for Apache Kafka unterstützt die standardmäßigen Kafka Connect-Transformationen und Prädikate.

Mit Transformationen können Sie einzelne Nachrichten ändern, bevor sie an Managed Service for Apache Kafka (für Quell-Connectors) oder an das externe System (für Senken-Connectors) gesendet werden. Sie können eine Transformation verwenden, um sensible Daten zu maskieren, Zeitstempel hinzuzufügen oder Felder umzubenennen.

Mit Prädikaten können Sie Daten anhand bestimmter Bedingungen filtern und so festlegen, auf welche Nachrichten eine Transformation basierend auf den Nachrichteneigenschaften angewendet wird.

Wenn Sie beispielsweise einen Sink-Connector so konfigurieren möchten, dass Nachrichten mit einem DoNotProcess-Header-Schlüssel ignoriert werden, fügen Sie die folgende Konfiguration hinzu:

transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess

Diese Konfiguration hat folgende Auswirkungen:

  1. Konfiguriert ein Prädikat mit dem Namen hasKey vom Typ org.apache.kafka.connect.transforms.predicates.HasHeaderKey. Dieses Prädikat stimmt mit allen Nachrichten überein, die einen Header mit dem Schlüssel DoNotProcess enthalten.

  2. Konfiguriert eine Transformation mit dem Namen dropMessage vom Typ org.apache.kafka.connect.transforms.Filter. Bei dieser Transformation werden alle Nachrichten verworfen, die mit dem konfigurierten Prädikat übereinstimmen.

  3. Verknüpft die Transformation mit dem Prädikat hasKey. So wird sichergestellt, dass nur Nachrichten mit dem Header-Schlüssel DoNotProcess von der Transformation verworfen werden.

Nächste Schritte

Apache Kafka® ist eine eingetragene Marke der Apache Software Foundation oder deren Tochtergesellschaften in den USA und/oder anderen Ländern.