Connexion de Event Streams à IBM Cloud Object Storage avec Kubernetes Service

Obtenir le runtime Kafka Connect pour l'exécuter dans un cluster Kubernetes Service Ensuite, démarrez le connecteur de puits IBM Cloud® Object Storage Pour archiver les données des sujets Kafka dans Event Streams vers une instance du service IBM Cloud® Object Storage.

Le connecteur consomme des lots de messages provenant de Kafka et charge les données du message en tant qu'objets dans un compartiment du service de cloud Object Storage.

Étape 1. Installation des prérequis

Vérifiez que le logiciel et les services suivants sont installés :

Une instance Event Streams - plan Standard ou Enterprise. Vous devez créer des données d'identification.
Une instance du service de cloud Object Storage avec au moins un compartiment.
Un cluster IBM Cloud® Kubernetes Service. Vous pouvez en mettre un à disposition gratuitement à des fins de test.

Vous avez également besoin d'un accès CLI à votre cluster. Pour plus d'informations, voir Configuration de l'interface de ligne de commande et de l'API.
Une version récente de Kubectl.
Git

Étape 2. Clonez les référentiels de connexion Kafka

Clonez les deux référentiels suivants qui contiennent les fichiers requis :

Étape 3. Création de votre configuration Kafka Connect

Vous devez configurer cette configuration une seule fois. Event Streams la conservera pour les utilisations ultérieures.

Dans le projet event-streams-samples, naviguez jusqu'au fichier kafka-connect/IKS directory, modifiez le fichier connect-distributed.properties et remplacez <BOOTSTRAP_SERVERS> à un endroit par <APIKEY> dans trois emplacements avec vos données d'identification Event Streams.

Indiquez <BOOTSTRAP_SERVERS> comme liste séparée par des virgules. Si elles ne sont pas valides, vous obtenez une erreur.

Votre <APIKEY> apparaît en texte clair sur votre machine, mais est secret lorsqu'il est envoyé vers IBM Cloud® Kubernetes Service.

Kafka Connect peut exécuter des noeuds worker multiples pour des raisons de fiabilité et d'évolutivité. Si votre cluster Kubernetes Service possède plusieurs nœuds et que vous souhaitez avoir plusieurs processus Connect, modifiez le fichier kafka-connect.yaml et l'entrée replicas: 1.

Exécutez ensuite les commandes suivantes :

Exécutez la commande suivante pour créer un secret :

kubectl create secret generic connect-distributed-config --from-file=connect-distributed.properties

Exécutez la commande suivante pour créer une mappe de configuration :

kubectl create configmap connect-log4j-config --from-file=connect-log4j.properties

Étape 4. Déployez Kafka Connect

Appliquez la configuration dans le fichier kafka-connect.yaml en exécutant la commande suivante :

kubectl apply -f ./kafka-connect.yaml

Étape 5. Validation de Kafka Connect en cours d'exécution

Pour vérifier que Kafka Connect est en cours d'exécution, réacheminez le port sur le port 8083 kafkaconnect-service, comme dans l'exemple suivant :

kubectl port-forward service/kafkaconnect-service 8083

Conservez le terminal que vous avez utilisé pour le transfert de port ouvert et utilisez un autre terminal pour les étapes suivantes.

L'API REST de Connect est alors disponible à l'adresse http://localhost:8083. Pour plus d'informations sur l'API, voir Interface REST deKafka Connect.

Par conséquent, vous disposez maintenant de l'environnement d'exécution Kafka Connect déployé et en cours d'exécution dans Kubernetes Service. Ensuite, configurez et démarrez le connecteur Object Storage

Étape 6. Configurer le fichier JSON cos-sink

Éditez le fichier cos-sink.json situé dans kafka-connect-ibmcos-sink/config/ pour indiquer vos informations au minimum pour les les propriétés requises. Bien que les propriétés de configuration cos.object.deadline.seconds, cos.interval.seconds et cos.object.records soient mentionnées comme étant facultatives, vous devez définir au moins l'une de ces propriétés sur une valeur autre que celle par défaut.

Propriétés du fichier cos-sink.json

Remplacez les espaces réservés du fichier cos-sink.json par vos propres valeurs.

Propriétés du fichier cos-sink.json
Propriétés du fichier	Description
cos.api.key	Obligatoire. Clé d'API utilisée pour se connecter à l'instance de service de cloud de Object Storage.
cos.bucket.location	Obligatoire. Emplacement du compartiment de service de cloud de Object Storage. Par exemple, pour un compartiment régional `eu-gb`, ou pour un compartiment global `eu`.
cos.bucket.name	Obligatoire. Nom du compartiment de service de cloud Object Storage dans lequel les données doivent être écrites.
cos.bucket.resiliency	Obligatoire. Résilience du compartiment de cloud de Object Storage. Doit contenir l'une des valeurs suivantes : cross-region, regional single-site.
cos.service.crn	Obligatoire. CRN pour l'instance de service de cloud de Object Storage. Veillez à entrer le nom de ressource de cloud correct: il s'agit de l'ID d'instance de ressource se terminant par deux points, par exemple, `crn:v1:staging:public:cloud-object-storage:global:a/8c226dc8c8bfb9bc3431515a16957954:b25fe12c-9cf5-4ee8-8285-2c7e6ae707f6::`.
cos.endpoint.visibility	Optionnel. Spécifiez public pour vous connecter au service Cloud Object Storage sur l'Internet public. Spécifiez private pour vous connecter à partir d'un connecteur qui s'exécute à l'intérieur du réseau IBM Cloud, par exemple, à partir d'un cluster IBM Cloud Kubernetes Service La valeur par défaut est "public".
cos.object.deadline.seconds	Optionnel. Nombre de secondes (mesuré comme le temps d'horloge pour l'instance de tâche Connect) entre la lecture du premier enregistrement de Kafka et l'écriture de tous les enregistrements lus jusqu'à présent dans un objet de cloud deObject Storage. Cette propriété peut être utile dans les situations où il y a de longues pauses entre les enregistrements Kafka produits pour un sujet. Il garantit que tous les enregistrements reçus par ce connecteur sont toujours écrits dans Object Storage dans le délai spécifié.
cos.object.interval.seconds	Optionnel. Le nombre de secondes (mesuré par les horodatages dans les enregistrements Kafka) entre la lecture du premier enregistrement de Kafka et l'écriture de tous les enregistrements lus jusqu'à présent dans un objet de cloud de Object Storage.
cos.object.records	Optionnel. Le nombre maximal d'enregistrements Kafka à associer dans un objet.

Obtenir les données d'identification Object Storage qui utilisent la console IBM Cloud

Recherchez votre service Object Storage sur le tableau de bord.
Cliquez sur la vignette de votre service.
Cliquez sur Données d'identification pour le service.
Cliquez sur Nouvelles données d'identification.
Entrez les détails de votre nouvelle donnée d'identification, par exemple un nom et un rôle, puis cliquez sur Ajouter. Une nouvelle donnée d'identification s'affiche dans la liste des données d'identification.
Cliquez sur cette donnée d'identification en utilisant Afficher les données d'identification pour afficher les détails au format JSON.

Etape 7. Démarrer le connecteur avec sa configuration

Exécutez la commande suivante pour démarrer le connecteur Object Storage avec la configuration que vous avez fournie à l'étape précédente.

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

Étape 8. Contrôler votre connecteur

Vous pouvez vérifier votre connecteur en allant à l'endroit suivant.

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

Si l'état indique que le connecteur ne s'exécute pas, redémarrez-le.

Étape 9. Supprimer votre connecteur

Vous pouvez utiliser la commande suivante pour supprimer un connecteur.

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