Event Streams mit IBM Cloud Object Storage über die Kubernetes Service verbinden

Holen Sie sich die Kafka in einem Kubernetes Service zur Ausführung. Dann starten Sie den IBM Cloud® Object Storage Sink Connector, um Daten aus Kafka in Event Streams in einer Instanz des IBM Cloud® Object Storage zu archivieren.

Der Connector verarbeitet Nachrichtenstapel aus Kafka und lädt die Nachrichtendaten als Objekte in ein Bucket im Cloud Object Storage -Service hoch.

Schritt 1. Voraussetzungen installieren

Stellen Sie sicher, dass die folgende Software und die folgenden Services installiert sind:

Event Streams-Instanz mit Plan "Standard" oder "Enterprise". Sie müssen Berechtigungsnachweise erstellen.
Eine Instanz des Cloud Object Storage -Service mit mindestens einem Bucket.
IBM Cloud® Kubernetes Service-Cluster. Zu Testzwecken können Sie einen Cluster kostenfrei einrichten.

Außerdem benötigen Sie CLI-Zugriff auf Ihren Cluster. Weitere Informationen finden Sie in CLI und API einrichten.
Aktuelle Version von Kubectl.
Git

Schritt 2. kafka-connect-Repositorys klonen

Klonen Sie die folgenden beiden Repositorys, die die erforderlichen Dateien enthalten:

Schritt 3. Kafka Connect-Konfiguration erstellen

Sie müssen diese Konfiguration nur einmal einrichten. Event Streams speichert sie für die zukünftige Verwendung.

Navigieren Sie im Projekt event-streams-samples zur Datei kafka-connect/IKS directory, bearbeiten Sie die connect-distributed.properties Datei, und ersetzen Sie <BOOTSTRAP_SERVERS> an einer Stelle und <APIKEY> an drei Stellen mit Ihren Event Streams -Berechtigungsnachweisen.

Geben Sie <BOOTSTRAP_SERVERS> als durch Kommas getrennte Liste an. Wenn sie nicht gültig sind, erhalten Sie einen Fehler.

Ihr Wert für <APIKEY> wird auf Ihrem System als Klartext angezeigt, bei der Übertragung mit Push-Operation an IBM Cloud® Kubernetes Service jedoch als geheimer Schlüssel behandelt.

Kafka Connect kann aus Gründen der Zuverlässigkeit und Skalierbarkeit mehrere Worker ausführen. Wenn Ihr Kubernetes Service Cluster mehrere Knoten hat und Sie möchten mehrere Connect-Worker möchten, bearbeiten Sie die Datei kafka-connect.yamlund den Eintrag replicas: 1.
Führen Sie anschließend die folgenden Befehle aus:

Führen Sie den folgenden Befehl aus, um einen geheimen Schlüssel zu erstellen:
```
kubectl create secret generic connect-distributed-config --from-file=connect-distributed.properties
```
Führen Sie den folgenden Befehl aus, um eine ConfigMap zu erstellen:
```
kubectl create configmap connect-log4j-config --from-file=connect-log4j.properties
```

Schritt 4. Kafka Connect bereitstellen

Wenden Sie die Konfiguration in der Datei kafka-connect.yamlan, indem Sie den folgenden Befehl ausführen:

kubectl apply -f ./kafka-connect.yaml

Schritt 5. Prüfen Sie, ob Kafka Connect ausgeführt wird

Um zu überprüfen, ob Kafka Connect ausgeführt wird, leiten Sie den Port wie im folgenden Beispiel an den Kafkaconnect-Service an Port 8083 weiter:

kubectl port-forward service/kafkaconnect-service 8083

Lassen Sie das Terminal, das Sie für die Portweiterleitung verwendet haben, geöffnet und verwenden Sie ein anderes Terminal für die nächsten Schritte.

Die Connect-REST-API ist dann unter http://localhost:8083verfügbar. Wenn Sie weitere Informationen über die API wünschen, siehe Kafka Connect REST-Schnittstelle.

Sie verfügen nun über die Kafka Connect-Laufzeit, die in Kubernetes Serviceimplementiert und ausgeführt wird. Als nächstes konfigurieren und starten Sie den Konnektor Object Storage.

Schritt 6. JSON-Datei für cos-sink konfigurieren

Bearbeiten Sie die Dateicos-sink.jsonim Verzeichnis kafka-connect-ibmcos-sink/config/, sodass Ihre erforderlichen Eigenschaften mindestens mit Ihren Informationen ausgefüllt werden. Obwohl die Konfigurationseigenschaften cos.object.deadline.seconds, cos.interval.seconds und cos.object.records als optional aufgeführt sind, müssen Sie mindestens eine dieser Eigenschaften auf einen nicht standardmäßigen Wert setzen.

Eigenschaften in der Datei "cos-sink.json"

Ersetzen Sie die Platzhalter in der Datei cos-sink.jsondurch Ihre eigenen Werte.

Eigenschaften in der Datei cos-sink.json""
Dateieigenschaften	Beschreibung
cos.api.key	Erforderlich. Der API-Schlüssel, der für die Verbindung zur Cloud- Object Storage -Serviceinstanz verwendet wird.
cos.bucket.location	Erforderlich. Position des Cloud Object Storage -Service-Buckets. Beispielsweise für eine regionales Bucket `eu-gb`oder für eine Globales Bucket `eu`.
cos.bucket.name	Erforderlich. Name des Cloud Object Storage -Service-Buckets, in das Daten geschrieben werden sollen.
cos.bucket.resiliency	Erforderlich. Ausfallsicherheit des Cloud Object Storage -Buckets. Der Wert muss "cross-region" (regionsübergreifend), "regional" (regional) oder "single-site" (einzelner Standort) lauten.
cos.service.crn	Erforderlich. CRN für die Cloud Object Storage -Serviceinstanz. Stellen Sie sicher, dass Sie den richtigen CRN eingeben: Dies ist die Ressourceninstanz-ID, die mit Doppelpunkten endet, z. B. `crn:v1:staging:public:cloud-object-storage:global:a/8c226dc8c8bfb9bc3431515a16957954:b25fe12c-9cf5-4ee8-8285-2c7e6ae707f6::`.
cos.endpoint.visibility	Optional. Geben Sie public an, um eine Verbindung zum Cloud-Service Object Storage über das öffentliche Internet herzustellen. Geben Sie "privat" an, um eine Verbindung von einem Connector herzustellen, der innerhalb des IBM Cloud ausgeführt wird, z. B. von einem IBM Cloud Kubernetes Service. Der Standardwert ist "public".
cos.object.deadline.seconds	Optional. Die Anzahl Sekunden (als gemessene Wanduhrzeit für die Connect-Taskinstanz) zwischen dem Lesen des ersten Datensatzes aus Kafka und dem Schreiben aller bisher gelesenen Datensätze in ein Cloud Object Storage -Objekt. Diese Eigenschaft kann in Situationen nützlich sein, in denen lange Pausen zwischen der Produktion von Kafka für ein Thema liegen. Sie stellt sicher, dass alle Datensätze, die von diesem Connector empfangen werden, immer innerhalb der angegebenen Zeit in Object Storage geschrieben werden.
cos.object.interval.seconds	Optional. Die Anzahl der Sekunden (gemessen anhand der Zeitmarken in Kafka-Datensätzen) zwischen dem Lesen des ersten Datensatzes aus Kafkaund dem Schreiben aller bisher gelesenen Datensätze in ein Cloud Object Storage -Objekt.
cos.object.records	Optional. Die maximale Anzahl von Kafka -Datensätzen, die in einem Objekt kombiniert werden können.

Object Storage -Berechtigungsnachweise abrufen, die die IBM Cloud -Konsole verwenden

Suchen Sie Ihren Object Storage-Dienst im Dashboard.
Klicken Sie auf die Servicekachel.
Klicken Sie auf Serviceberechtigungsnachweise.
Klicken Sie auf Neuer Berechtigungsnachweis.
Geben Sie die Details zu Ihrem neuen Berechtigungsnachweis, z. B. einen Namen und eine Rolle, ein und klicken Sie auf Hinzufügen. In der Berechtigungsnachweisliste wird ein neuer Berechtigungsnachweis angezeigt.
Klicken Sie auf diesen Berechtigungsnachweis, indem Sie Berechtigungsnachweise anzeigen verwenden, um die Details im JSON-Format anzuzeigen.

Schritt 7. Connector mit seiner Konfiguration starten

Führen Sie den folgenden Befehl aus, um den Object Storage -Connector mit der Konfiguration zu starten, die Sie im vorherigen Schritt angegeben haben.

curl -X POST -H "Content-Type: application/json" http://localhost:8083/connectors --data "@./cos-sink.json"

Schritt 8. Connector überwachen

Sie können Ihren Anschluss überprüfen, indem Sie die folgende Adresse aufrufen.

http://localhost:8083/connectors/cos-sink/status

Falls für den Connector nicht der Status "running" (aktiv) angegeben ist, starten Sie den Connector erneut.

Schritt 9. Connector löschen

Mit dem folgenden Befehl können Sie einen Connector löschen.

curl -X DELETE
http://localhost:8083/connectors/cos-sink