Ü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 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 und Reaktion auf Datenbankereignisse
Cloud Storage-Senke Streamt Daten aus Kafka-Themen in einen Cloud Storage-Bucket. Data Lake-Aufnahme, Datenarchivierung
MirrorMaker 2.0 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
Allgemeine PostgreSQL-Quelle Liest Änderungen auf Zeilenebene aus einer PostgreSQL-Datenbank und schreibt sie in Kafka. Datenbankintegration, Überwachung und Reaktion auf Datenbankereignisse

Nutzer mit Conversion

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

  • Bei Sink-Connectors werden Daten aus dem Wire-Format des Themas in das interne Datenformat von Kafka Connect deserialisiert, 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.

Wenn Sie einen Connector konfigurieren, legen Sie die folgenden Attribute fest:

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

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

Wenn Sie keinen Converter angeben, ist der Standard-Convertertyp org.apache.kafka.connect.converters.ByteArrayConverter. Dabei werden die Daten im Rohbyteformat weitergegeben.

Unterstützte Konverter

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

Nutzer mit ConversionFormat
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 in zwei Systemen 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 Auswahl des Converters hängt vom Connectortyp und den Daten ab, die Sie in Kafka speichern. Weitere Informationen finden Sie in der Dokumentation für den 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 dieser Wert.

Durch Erhöhen des Werts von tasks.max kann der Durchsatz verbessert werden, aber auch der Ressourcenverbrauch (CPU und Arbeitsspeicher) steigt. Der optimale Wert hängt von der Arbeitslast und den Ressourcen ab, die den Connect-Cluster-Workern zugewiesen sind. Bei Senken-Connectors kann sich die Anzahl der Kafka-Topic-Partitionen auch auf die Parallelität auswirken.

Richtlinie für Task-Neustart

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

  • Nie neu starten. Der Connector startet fehlgeschlagene Tasks nicht neu. Das ist das Standardverhalten. Das ist nützlich für das Debugging oder in Situationen, in denen nach einem Fehler manuelles Eingreifen erforderlich ist.

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

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

Transformationen und Vorhersagen

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 Ziel-Connectors) gesendet werden. Sie können eine Transformation verwenden, um vertrauliche 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 angewendet wird.

Wenn Sie beispielsweise einen Senken-Connector so konfigurieren möchten, dass Nachrichten mit dem Header-Schlüssel DoNotProcess 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 entspricht allen Nachrichten, 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 dem konfigurierten Prädikat entsprechen.

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

Nächste Schritte

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