In diesem Dokument wird beschrieben, wie Sie Apache Kafka und Pub/Sub mithilfe des Kafka-Connectors für Pub/Sub-Gruppen integrieren.
Informationen zum Kafka-Connector der Pub/Sub-Gruppe
Apache Kafka ist eine Open-Source-Plattform für das Streaming von Ereignissen. Es wird häufig in verteilten Architekturen verwendet, um die Kommunikation zwischen lose gekoppelten Komponenten zu ermöglichen. Pub/Sub ist ein verwalteter Dienst zum asynchronen Senden und Empfangen von Nachrichten. Wie bei Kafka können Sie Pub/Sub für die Kommunikation zwischen Komponenten in Ihrer Cloud-Architektur verwenden.
Mit dem Kafka-Connector für Pub/Sub-Gruppen können Sie diese beiden Systeme einbinden. Die folgenden Connectors sind in der Connector-JAR-Datei verpackt:
- Der Senken-Connector liest Datensätze aus einem oder mehreren Kafka-Themen und veröffentlicht sie in Pub/Sub.
- Der Quell-Connector liest Nachrichten aus einem Pub/Sub-Thema und veröffentlicht sie in Kafka.
Im Folgenden finden Sie einige Szenarien, in denen Sie den Kafka-Connector für Pub/Sub-Gruppen verwenden können:
- Sie migrieren eine Kafka-basierte Architektur zu Google Cloud.
- Sie haben ein Frontend-System, das Ereignisse in Kafka außerhalb von Google Cloud speichert, verwenden aber auch Google Cloud, um einige Ihrer Backend-Dienste auszuführen, die die Kafka-Ereignisse empfangen müssen.
- Sie erfassen Logs aus einer lokalen Kafka-Lösung und senden sie zur Datenanalyse an Google Cloud.
- Sie haben ein Front-End-System, das Google Cloud verwendet, aber Sie speichern Daten auch lokal mit Kafka.
Für den Connector ist Kafka Connect erforderlich, ein Framework für das Streaming von Daten zwischen Kafka und anderen Systemen. Wenn Sie den Connector verwenden möchten, müssen Sie Kafka Connect zusammen mit Ihrem Kafka-Cluster ausführen.
In diesem Dokument wird davon ausgegangen, dass Sie sowohl mit Kafka als auch Pub/Sub vertraut sind. Bevor Sie dieses Dokument lesen, sollten Sie eine der Pub/Sub-Kurzanleitungen durchgehen.
Der Pub/Sub-Connector unterstützt keine Integration zwischen Google Cloud IAM- und Kafka Connect-ACLs.
Erste Schritte mit dem Connector
In diesem Abschnitt werden Sie durch die folgenden Aufgaben geführt:- Konfigurieren Sie den Kafka-Connector der Pub/Sub-Gruppe.
- Ereignisse von Kafka an Pub/Sub senden
- Nachrichten von Pub/Sub an Kafka senden.
Vorbereitung
Kafka installieren
Folgen Sie der Kurzanleitung zu Apache Kafka, um eine Kafka-Datei mit einem einzelnen Knoten auf Ihrem lokalen Computer zu installieren. Führen Sie die folgenden Schritte in der Kurzanleitung aus:
- Laden Sie den neuesten Kafka-Release herunter und extrahieren Sie ihn.
- Starten Sie die Kafka-Umgebung.
- Erstellen Sie ein Kafka-Thema.
Authentifizieren
Der Kafka-Connector der Pub/Sub-Gruppe muss sich bei Pub/Sub authentifizieren, um Pub/Sub-Nachrichten zu senden und zu empfangen. So richten Sie die Authentifizierung ein:
- Melden Sie sich bei Ihrem Google Cloud-Konto an. Wenn Sie mit Google Cloud noch nicht vertraut sind, erstellen Sie ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.
- Installieren Sie die Google Cloud CLI.
-
Führen Sie folgenden Befehl aus, um die gcloud CLI zu initialisieren:
gcloud init
-
Google Cloud-Projekt erstellen oder auswählen.
-
Erstellen Sie ein Google Cloud-Projekt:
gcloud projects create PROJECT_ID
Ersetzen Sie
PROJECT_ID
durch einen Namen für das Google Cloud-Projekt, das Sie erstellen. -
Wählen Sie das von Ihnen erstellte Google Cloud-Projekt aus:
gcloud config set project PROJECT_ID
Ersetzen Sie
PROJECT_ID
durch den Namen Ihres Google Cloud-Projekts.
-
-
Erstellen Sie lokale Anmeldedaten zur Authentifizierung für Ihr Google-Konto:
gcloud auth application-default login
-
Gewähren Sie Ihrem Google-Konto Rollen. Führen Sie den folgenden Befehl für jede der folgenden IAM-Rollen einmal aus:
roles/pubsub.admin
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE
- Ersetzen Sie
PROJECT_ID
durch Ihre Projekt-ID. - Ersetzen Sie
EMAIL_ADDRESS
durch Ihre E-Mail-Adresse. - Ersetzen Sie
ROLE
durch jede einzelne Rolle.
- Ersetzen Sie
- Installieren Sie die Google Cloud CLI.
-
Führen Sie folgenden Befehl aus, um die gcloud CLI zu initialisieren:
gcloud init
-
Google Cloud-Projekt erstellen oder auswählen.
-
Erstellen Sie ein Google Cloud-Projekt:
gcloud projects create PROJECT_ID
Ersetzen Sie
PROJECT_ID
durch einen Namen für das Google Cloud-Projekt, das Sie erstellen. -
Wählen Sie das von Ihnen erstellte Google Cloud-Projekt aus:
gcloud config set project PROJECT_ID
Ersetzen Sie
PROJECT_ID
durch den Namen Ihres Google Cloud-Projekts.
-
-
Erstellen Sie lokale Anmeldedaten zur Authentifizierung für Ihr Google-Konto:
gcloud auth application-default login
-
Gewähren Sie Ihrem Google-Konto Rollen. Führen Sie den folgenden Befehl für jede der folgenden IAM-Rollen einmal aus:
roles/pubsub.admin
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE
- Ersetzen Sie
PROJECT_ID
durch Ihre Projekt-ID. - Ersetzen Sie
EMAIL_ADDRESS
durch Ihre E-Mail-Adresse. - Ersetzen Sie
ROLE
durch jede einzelne Rolle.
- Ersetzen Sie
Connector-JAR-Datei herunterladen
Laden Sie die Connector-JAR-Datei auf Ihren lokalen Computer herunter. Weitere Informationen finden Sie in der GitHub-Readme-Datei unter Connector erwerben.
Konfigurationsdateien des Connectors kopieren
Klonen Sie das GitHub-Repository für den Connector oder laden Sie es herunter.
git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git cd java-pubsub-group-kafka-connector
Kopieren Sie den Inhalt des Verzeichnisses
config
in das Unterverzeichnisconfig
Ihrer Kafka-Installation.cp config/* [path to Kafka installation]/config/
Diese Dateien enthalten Konfigurationseinstellungen für den Connector.
Kafka Connect-Konfiguration aktualisieren
- Wechseln Sie zu dem Verzeichnis, das die heruntergeladene Kafka Connect-Binärdatei enthält.
- Öffnen Sie im Binärverzeichnis von Kafka Connect die Datei
config/connect-standalone.properties
in einem Texteditor. - Wenn das
plugin.path property
auskommentiert ist, entfernen Sie das Kommentarzeichen. Aktualisieren Sie
plugin.path property
, um den Pfad zur Connector-JAR-Datei aufzunehmen.Beispiel:
plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar
Legen Sie für das Attribut
offset.storage.file.filename
einen lokalen Dateinamen fest. Im eigenständigen Modus verwendet Kafka diese Datei zum Speichern von Offset-Daten.Beispiel:
offset.storage.file.filename=/tmp/connect.offsets
Ereignisse von Kafka an Pub/Sub weiterleiten
In diesem Abschnitt wird beschrieben, wie Sie den Senken-Connector starten, Ereignisse in Kafka veröffentlichen und dann die weitergeleiteten Nachrichten von Pub/Sub lesen.
Verwenden Sie die Google Cloud CLI, um ein Pub/Sub-Thema mit einem Abo zu erstellen.
gcloud pubsub topics create PUBSUB_TOPIC gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC
Ersetzen Sie Folgendes:
- PUBSUB_TOPIC: Der Name eines Pub/Sub-Themas, das die Nachrichten von Kafka empfangen soll.
- PUBSUB_SUBSCRIPTION: Der Name eines Pub/Sub-Abos für das Thema.
Öffnen Sie die Datei
/config/cps-sink-connector.properties
in einem Texteditor. Fügen Sie Werte für die folgenden Attribute hinzu, die in den Kommentaren mit"TODO"
gekennzeichnet sind:topics=KAFKA_TOPICS cps.project=PROJECT_ID cps.topic=PUBSUB_TOPIC
Ersetzen Sie Folgendes:
- KAFKA_TOPICS: Eine durch Kommas getrennte Liste von Kafka-Themen, aus denen gelesen werden soll.
- PROJECT_ID: Das Google Cloud-Projekt, das Ihr Pub/Sub-Thema enthält.
- PUBSUB_TOPIC: Das Pub/Sub-Thema, das die Nachrichten von Kafka empfangen soll.
Führen Sie im Kafka-Verzeichnis den folgenden Befehl aus:
bin/connect-standalone.sh \ config/connect-standalone.properties \ config/cps-sink-connector.properties
Folgen Sie den Schritten in der Kurzanleitung zu Apache Kafka, um einige Ereignisse in Ihr Kafka-Thema zu schreiben.
Verwenden Sie die gcloud CLI, um die Ereignisse aus Pub/Sub zu lesen.
gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack
Nachrichten von Pub/Sub an Kafka weiterleiten
In diesem Abschnitt wird beschrieben, wie Sie den Quell-Connector starten, Nachrichten in Pub/Sub veröffentlichen und die weitergeleiteten Nachrichten von Kafka lesen.
Verwenden Sie die gcloud CLI, um ein Pub/Sub-Thema mit einem Abo zu erstellen.
gcloud pubsub topics create PUBSUB_TOPIC gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC
Ersetzen Sie Folgendes:
- PUBSUB_TOPIC: Der Name eines Pub/Sub-Themas.
- PUBSUB_SUBSCRIPTION: Der Name eines Pub/Sub-Abos.
Öffnen Sie die Datei
/config/cps-source-connector.properties
in einem Texteditor. Fügen Sie Werte für die folgenden Attribute hinzu, die in den Kommentaren mit"TODO"
gekennzeichnet sind:kafka.topic=KAFKA_TOPIC cps.project=PROJECT_ID cps.subscription=PUBSUB_SUBSCRIPTION
Ersetzen Sie Folgendes:
- KAFKA_TOPIC: Die Kafka-Themen, die die Pub/Sub-Nachrichten empfangen sollen.
- PROJECT_ID: Das Google Cloud-Projekt, das Ihr Pub/Sub-Thema enthält.
- PUBSUB_TOPIC: Das Pub/Sub-Thema.
Führen Sie im Kafka-Verzeichnis den folgenden Befehl aus:
bin/connect-standalone.sh \ config/connect-standalone.properties \ config/cps-source-connector.properties
Verwenden Sie die gcloud CLI, um eine Nachricht in Pub/Sub zu veröffentlichen.
gcloud pubsub topics publish PUBSUB_TOPIC --message="message 1"
Lesen Sie die Nachricht von Kafka. Folgen Sie den Schritten in der Kurzanleitung zu Apache Kafka, um die Nachrichten zum Kafka-Thema zu lesen.
SMS-Conversion
Ein Kafka-Eintrag enthält einen Schlüssel und einen Wert, bei denen es sich um Bytearrays mit variabler Länge handelt. Optional kann ein Kafka-Eintrag auch Header enthalten, bei denen es sich um Schlüssel/Wert-Paare handelt. Eine Pub/Sub-Nachricht besteht aus zwei Hauptteilen: dem Nachrichtentext und null oder mehr Schlüssel/Wert-Attributen.
Kafka Connect verwendet Converter, um Schlüssel und Werte zu und von Kafka zu synchronisieren. Legen Sie die folgenden Attribute in den Konfigurationsdateien des Connectors fest, um die Serialisierung zu steuern:
key.converter
: Der Converter, der zum Serialisieren von Datensatzschlüsseln verwendet wird.value.converter
: Der Converter, der zum Serialisieren von Datensatzwerten verwendet wird.
Der Text einer Pub/Sub-Nachricht ist ein ByteString
-Objekt, sodass die effizienteste Konvertierung das direkte Kopieren der Nutzlast ist. Aus diesem Grund empfehlen wir die Verwendung eines Converters, der nach Möglichkeit primitive Datentypen (Ganzzahl, Gleitkommazahl, String- oder Byteschema) erzeugt, um die Deserialisierung und Reserialisierung desselben Nachrichtentextes zu verhindern.
Konvertierung von Kafka zu Pub/Sub
Der Senken-Connector konvertiert Kafka-Einträge so in Pub/Sub-Nachrichten:
- Der Kafka-Eintragsschlüssel wird als Attribut mit dem Namen
"key"
in der Pub/Sub-Nachricht gespeichert. - Standardmäßig löscht der Connector alle Header im Kafka-Eintrag. Wenn Sie jedoch die Konfigurationsoption
headers.publish
auftrue
setzen, schreibt der Connector die Header als Pub/Sub-Attribute. Der Connector überspringt alle Header, die die Pub/Sub-Beschränkungen für Nachrichtenattribute überschreiten. - Bei Ganzzahl-, Float-, String- und Byteschemas übergibt der Connector die Bytes des Kafka-Eintragswerts direkt an den Pub/Sub-Nachrichtentext.
- Bei Strukturschemas schreibt der Connector jedes Feld als Attribut der Pub/Sub-Nachricht. Wenn das Feld beispielsweise
{ "id"=123 }
lautet, hat die resultierende Pub/Sub-Nachricht das Attribut"id"="123"
. Der Feldwert wird immer in einen String umgewandelt. Map- und Strukturtypen werden innerhalb einer Struktur nicht als Feldtypen unterstützt. - Bei Kartenschemas schreibt der Connector jedes Schlüssel/Wert-Paar als Attribut der Pub/Sub-Nachricht. Wenn die Zuordnung beispielsweise
{"alice"=1,"bob"=2}
ist, hat die resultierende Pub/Sub-Nachricht die beiden Attribute"alice"="1"
und"bob"="2"
. Die Schlüssel und Werte werden in Strings konvertiert.
Struktur- und Kartenschemas weisen einige zusätzliche Verhaltensweisen auf:
Optional können Sie ein bestimmtes Strukturfeld oder einen Zuordnungsschlüssel als Nachrichtentext angeben. Dazu legen Sie das Konfigurationsattribut
messageBodyName
fest. Der Wert des Felds oder Schlüssels wird alsByteString
im Nachrichtentext gespeichert. Wenn SiemessageBodyName
nicht festlegen, ist der Nachrichtentext für Struktur- und Kartenschemas leer.Bei Arraywerten unterstützt der Connector nur primitive Arraytypen. Die Folge der Werte im Array wird zu einem einzelnen
ByteString
-Objekt verkettet.
Konvertierung von Pub/Sub zu Kafka
Der Quell-Connector konvertiert Pub/Sub-Nachrichten so in Kafka-Einträge:
Kafka-Eintragsschlüssel: Der Schlüssel ist standardmäßig auf
null
festgelegt. Optional können Sie mit der Konfigurationsoptionkafka.key.attribute
ein Pub/Sub-Nachrichtenattribut angeben, das als Schlüssel verwendet werden soll. In diesem Fall sucht der Connector nach einem Attribut mit diesem Namen und setzt den Datensatzschlüssel auf den Attributwert. Wenn das angegebene Attribut nicht vorhanden ist, wird der Eintragsschlüssel aufnull
gesetzt.Wert des Kafka-Eintrags: Der Connector schreibt den Eintragswert so:
Wenn die Pub/Sub-Nachricht keine benutzerdefinierten Attribute hat, schreibt der Connector den Pub/Sub-Nachrichtentext direkt in den Wert des Kafka-Eintrags als Typ
byte[]
. Dazu verwendet er den vonvalue.converter
angegebenen Converter.Wenn die Pub/Sub-Nachricht benutzerdefinierte Attribute hat und
kafka.record.headers
den Wertfalse
hat, schreibt der Connector eine Struktur in den Eintragswert. Die Struktur enthält ein Feld für jedes Attribut und ein Feld namens"message"
, dessen Wert der (in Byte gespeicherte) Pub/Sub-Nachrichtentext ist:{ "message": "<Pub/Sub message body>", "<attribute-1>": "<value-1>", "<attribute-2>": "<value-2>", .... }
In diesem Fall müssen Sie einen
value.converter
verwenden, der mitstruct
-Schemas kompatibel ist, z. B.org.apache.kafka.connect.json.JsonConverter
.Wenn die Pub/Sub-Nachricht benutzerdefinierte Attribute hat und
kafka.record.headers
den Werttrue
hat, schreibt der Connector die Attribute als Kafka-Eintragsheader. Der Pub/Sub-Nachrichtentext wird mithilfe des durchvalue.converter
angegebenen Converters direkt alsbyte[]
-Typ in den Wert des Kafka-Eintrags geschrieben.
Kafka-Eintragsheader Standardmäßig sind die Header leer, sofern Sie
kafka.record.headers
nicht auftrue
festlegen.
Konfigurationsoptionen
Zusätzlich zu den von der Kafka Connect API bereitgestellten Konfigurationen unterstützt der Kafka-Connector für Pub/Sub-Gruppen die folgenden Konfigurationen.
Konfigurationsoptionen für Senken-Connectors
Der Senken-Connector unterstützt die folgenden Konfigurationsoptionen.
Einstellung | Datentyp | Beschreibung |
---|---|---|
connector.class |
String |
Erforderlich. Die Java-Klasse für den Connector. Für den Pub/Sub-Senken-Connector muss der Wert com.google.pubsub.kafka.sink.CloudPubSubSinkConnector sein.
|
cps.endpoint |
String |
Der zu verwendende Pub/Sub-Endpunkt. Standardeinstellung: |
cps.project |
String |
Erforderlich. Die Google Cloud, die das Pub/Sub-Thema enthält. |
cps.topic |
String |
Erforderlich. Das Pub/Sub-Thema, in dem Kafka-Einträge veröffentlicht werden sollen. |
gcp.credentials.file.path |
String |
Optional. Der Pfad zu einer Datei, in der Google Cloud-Anmeldedaten zur Authentifizierung von Pub/Sub Lite gespeichert sind. |
gcp.credentials.json |
String |
Optional. Ein JSON-Blob, das Google Cloud zur Authentifizierung von Pub/Sub Lite enthält. |
headers.publish |
Boolean |
Geben Sie bei Standardeinstellung: |
maxBufferBytes |
Long |
Die maximale Anzahl von Byte, die auf einer Kafka-Partition eines Themas empfangen werden sollen, bevor sie in Pub/Sub veröffentlicht werden. Standardwert: 10000000. |
maxBufferSize |
Integer |
Die maximale Anzahl von Datensätzen, die auf einer Kafka-Themenpartition empfangen werden sollen, bevor sie in Pub/Sub veröffentlicht werden. Der Standardwert is 100. |
maxDelayThresholdMs |
Integer |
Die maximale Wartezeit in Millisekunden, bis Der Standardwert is 100. |
maxOutstandingMessages |
Long |
Die maximale Anzahl von Datensätzen, die ausstehend sein dürfen, einschließlich unvollständiger und ausstehender Batches, bevor der Verlag oder Webpublisher die weitere Veröffentlichung blockiert. Standardeinstellung: |
maxOutstandingRequestBytes |
Long |
Die maximale Gesamtzahl der Byte, die ausstehend sind, einschließlich unvollständiger und ausstehender Batches, bevor der Verlag oder Webpublisher die weitere Veröffentlichung blockiert. Standardeinstellung: |
maxRequestTimeoutMs |
Integer |
Das Zeitlimit für einzelne Veröffentlichungsanfragen an Pub/Sub in Millisekunden. Standardwert: 10.000. |
maxTotalTimeoutMs |
Integer |
Das gesamte Zeitlimit in Millisekunden für einen Aufruf zur Veröffentlichung in Pub/Sub, einschließlich Wiederholungen. Standardeinstellung: 60.000. |
metadata.publish |
Boolean |
Bei Standardeinstellung: |
messageBodyName |
String |
Gibt bei Verwendung eines Struktur- oder Zuordnungswertschemas den Namen eines Felds oder Schlüssels an, der als Pub/Sub-Nachrichtentext verwendet werden soll. Siehe Konvertierung von Kafka zu Pub/Sub. Standardeinstellung: |
orderingKeySource |
String |
Gibt an, wie der Sortierungsschlüssel in der Pub/Sub-Nachricht festgelegt wird. Kann einer der folgenden Werte sein:
Standardeinstellung: |
topics |
String |
Erforderlich. Eine durch Kommas getrennte Liste von Kafka-Themen, aus denen gelesen werden soll. |
Konfigurationsoptionen für Quell-Connector
Der Quell-Connector unterstützt die folgenden Konfigurationsoptionen.
Einstellung | Datentyp | Beschreibung |
---|---|---|
connector.class |
String |
Erforderlich. Die Java-Klasse für den Connector. Für den Pub/Sub-Quell-Connector muss der Wert com.google.pubsub.kafka.source.CloudPubSubSourceConnector sein.
|
cps.endpoint |
String |
Der zu verwendende Pub/Sub-Endpunkt. Standardeinstellung: |
cps.makeOrderingKeyAttribute |
Boolean |
Wenn Standardeinstellung: |
cps.maxBatchSize |
Integer |
Die maximale Anzahl von Nachrichten, die pro Pull-Anfrage an Pub/Sub als Batch verarbeitet werden sollen. Standardeinstellung: 100 |
cps.project |
String |
Erforderlich. Das Google Cloud-Projekt, das das Pub/Sub-Thema enthält. |
cps.subscription |
String |
Erforderlich. Der Name des Pub/Sub-Abos, aus dem Nachrichten abgerufen werden sollen. |
gcp.credentials.file.path |
String |
Optional. Der Pfad zu einer Datei, in der Google Cloud-Anmeldedaten zur Authentifizierung von Pub/Sub Lite gespeichert sind. |
gcp.credentials.json |
String |
Optional. Ein JSON-Blob, das Google Cloud zur Authentifizierung von Pub/Sub Lite enthält. |
kafka.key.attribute |
String |
Das Pub/Sub-Nachrichtenattribut, das als Schlüssel für in Kafka veröffentlichte Nachrichten verwendet werden soll. Wenn Standardeinstellung: |
kafka.partition.count |
Integer |
Die Anzahl der Kafka-Partitionen für das Kafka-Thema, in dem Nachrichten veröffentlicht werden. Dieser Parameter wird ignoriert, wenn das Partitionsschema Der Standardwert ist 1. |
kafka.partition.scheme |
String |
Das Schema zum Zuweisen einer Nachricht zu einer Partition in Kafka. Kann einer der folgenden Werte sein:
Standardeinstellung: |
kafka.record.headers |
Boolean |
Wenn |
kafka.topic |
String |
Erforderlich. Das Kafka-Thema, das Nachrichten von Pub/Sub empfängt. |
Support
Wenn Sie Hilfe benötigen, erstellen Sie ein Support-Ticket. Erstellen Sie für allgemeine Fragen und Diskussionen ein Problem im GitHub-Repository.
Nächste Schritte
- Unterschiede zwischen Kafka und Pub/Sub
- Weitere Informationen zum Kafka-Connector für Pub/Sub-Gruppen
- Weitere Informationen finden Sie im GitHub-Repository für den Kafka-Connector für Pub/Sub-Gruppe.