Kafka Connect - Comment relier Kafka Connect à Kafka et créer un connecteur

Objectif

Kafka Connect déplace des données entre vos topics Kafka et des systèmes externes grâce à des connecteurs prêts à l'emploi : les connecteurs source font entrer des données dans Kafka, les connecteurs sink copient les données de Kafka vers un autre système.

Ce guide explique comment relier un service Kafka Connect à votre service Kafka, puis comment créer, suivre et gérer un connecteur avec l'API OVHcloud. L'exemple copie les messages d'un topic Kafka dans une collection Public Cloud Databases pour MongoDB.

Prérequis

• Disposer d'un projet Public Cloud dans votre compte OVHcloud
• Disposer d'un service Kafka avec un topic, ici orders, dont les messages sont des documents JSON
• Disposer d'un service Kafka Connect dans le même projet Public Cloud
• Disposer d'un service Public Cloud Databases pour MongoDB et d'un utilisateur MongoDB ayant le rôle readWrite sur la base cible (pour l'exemple)
• Disposer d'un accès à l'API OVHcloud

En pratique

Étape 1 - Relier Kafka Connect à votre service Kafka

Kafka Connect lit et écrit dans un service Kafka du même projet grâce à une intégration. Créez-la sur votre service Kafka Connect :

{
  "type": "kafkaConnect",
  "sourceServiceId": "<ID du service Kafka>",
  "destinationServiceId": "<ID du service Kafka Connect>"
}

L'intégration est prête immédiatement :

{
  "id": "<ID de l'intégration>",
  "sourceServiceId": "<ID du service Kafka>",
  "destinationServiceId": "<ID du service Kafka Connect>",
  "type": "kafkaConnect",
  "status": "READY"
}

Kafka Connect crée ensuite 3 topics sur votre service Kafka pour stocker sa configuration, ses offsets et son état : __connect_configs-<ID de l'intégration>, __connect_offsets-<ID de l'intégration> et __connect_status-<ID de l'intégration>. Ces topics restent sur votre service Kafka après la suppression du service Kafka Connect.

Étape 2 - Autoriser Kafka Connect à joindre le système cible

Un connecteur se connecte au système externe depuis un nœud de votre service Kafka Connect. Pour un service Kafka Connect public, trouvez l'adresse IP de chaque nœud en résolvant son nom d'hôte, affiché dans le champ nodes du détail du service :

dig +short kafkaconnect-xxxxxxxx-xxxxxxxxx-1.database.cloud.ovh.net

Ajoutez ensuite ces adresses IP aux IP autorisées du service cible, ici le service MongoDB. Ces adresses IP peuvent changer au fil du temps : si le connecteur ne parvient plus à joindre le système cible, résolvez à nouveau les noms d'hôte et mettez à jour les IP autorisées. Pour plus d'informations, consultez le guide « Capacités et limitations pour Analytics avec Kafka Connect ».

Étape 3 - Choisir un connecteur

Listez les connecteurs disponibles sur votre service Kafka Connect, puis consultez le détail de chacun :

{
  "id": "<ID de la capacité du connecteur>",
  "name": "MongoDB Kafka Sink Connector",
  "author": "MongoDB",
  "documentationUrl": "https://docs.mongodb.com/kafka-connector/current/",
  "preview": false,
  "type": "sink",
  "version": "1.16.0",
  "latest": true
}

Conservez l'id du connecteur à utiliser. Le champ documentationUrl renvoie vers la documentation du connecteur, qui liste ses propriétés de configuration. Vous pouvez aussi parcourir la liste des connecteurs disponibles.

Étape 4 - Créer le connecteur

Créez le connecteur avec l'id de sa capacité, un nom et sa configuration :

{
  "name": "orders-to-mongodb",
  "connectorId": "<ID de la capacité du connecteur>",
  "configuration": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "topics": "orders",
    "connection.uri": "mongodb+srv://<utilisateur>:<mot de passe>@<hôte MongoDB>/admin?replicaSet=replicaset&tls=true",
    "database": "shop",
    "collection": "orders",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

Dans cette configuration :

• topics est le topic Kafka à lire ;
• connection.uri, database et collection définissent où les messages sont écrits dans MongoDB ;
• les convertisseurs lisent les clés des messages comme des chaînes de caractères et leurs valeurs comme des documents JSON sans schéma.

Kafka Connect vérifie la configuration à la création du connecteur. Si le système cible est injoignable, par exemple parce que l'adresse IP de Kafka Connect n'est pas encore autorisée, l'API renvoie une erreur 400 avec le message Unable to connect to the server.

Chaque message du topic orders est désormais écrit sous forme de document dans la collection shop.orders. Par exemple, le message {"amount":42} devient le document { "_id": ObjectId("<ID du document>"), "amount": 42 }.

Étape 5 - Suivre le connecteur

Dans les appels suivants, connectorId est l'id renvoyé à la création du connecteur, et non l'ID de la capacité utilisé à l'étape précédente. Consultez l'état du connecteur, puis listez ses tâches et l'état de chacune :

{"id":0,"trace":"","status":"RUNNING"}

Les tâches effectuent le travail du connecteur : un connecteur fonctionne lorsque ses tâches sont à l'état RUNNING.

Étape 6 - Mettre en pause, reprendre, redémarrer ou supprimer le connecteur

Mettez en pause, reprenez, redémarrez ou supprimez un connecteur :

Aller plus loin

Capacités et limitations pour Analytics avec Kafka Connect

Connecteurs disponibles pour Kafka Connect

Votre avis nous intéresse !

Nous serions ravis de répondre à vos questions et apprécions tout retour que vous pourriez nous faire.

Pour une formation ou une assistance technique sur la mise en œuvre de nos solutions, contactez votre commercial ou consultez la page Professional Services pour obtenir un devis et faire analyser votre projet par nos experts.

Vous êtes sur Discord ? Rejoignez notre chaîne via https://discord.gg/ovhcloud et interagissez directement avec l'équipe qui développe notre service Analytics !

Échangez avec notre communauté d'utilisateurs.