Comment ingérer des données dans Elasticsearch via Apache Camel ?

Elasticsearch vous permet d’indexer des données rapidement et en toute flexibilité. Essayez-le gratuitement dans le cloud ou exécutez-le en local pour découvrir à quel point l’indexation peut être simple.

L'ingestion de données dans Elasticsearch à l'aide d'Apache Camel est un processus qui combine la robustesse d'un moteur de recherche et la flexibilité d'un cadre d'intégration. Dans cet article, nous allons voir comment Apache Camel peut simplifier et optimiser l'ingestion de données dans Elasticsearch. Pour illustrer cette fonctionnalité, nous allons mettre en œuvre une application d'introduction qui démontre, étape par étape, comment configurer et utiliser Apache Camel pour envoyer des données à Elasticsearch.

Qu'est-ce qu'Apache Camel ?

Apache Camel est un cadre d'intégration open-source qui simplifie la connexion de divers systèmes, permettant aux développeurs de se concentrer sur la logique commerciale sans se préoccuper des complexités de la communication entre les systèmes. Le concept central de Camel est "routes," qui définit le chemin suivi par un message de l'origine à la destination, en incluant éventuellement des étapes intermédiaires telles que des transformations, des validations et des filtrages.

Architecture d'Apache Camel

Camel utilise les composants "" pour se connecter à différents systèmes et protocoles, tels que les bases de données et les services de messagerie, et les points d'extrémité "" pour représenter les points d'entrée et de sortie des messages. Ces concepts offrent une conception modulaire et flexible, facilitant la configuration et la gestion d'intégrations complexes de manière efficace et évolutive.

Utilisation d'Elasticsearch et d'Apache Camel

Nous allons montrer comment configurer une application Java simple qui utilise Apache Camel pour ingérer des données dans un cluster Elasticsearch. Les processus de création, de mise à jour et de suppression de données dans Elasticsearch à l'aide de routes définies dans Apache Camel seront également abordés.

1. Ajout de dépendances

La première étape de la configuration de cette intégration consiste à ajouter les dépendances nécessaires au fichier pom.xml de votre projet. Cela inclut les bibliothèques Apache Camel et Elasticsearch. Nous utiliserons la nouvelle bibliothèque Java API Client, nous devons donc importer le composant camel-elasticsearch et la version doit être la même que celle de la bibliothèque camel-core.

Si vous souhaitez utiliser le Java Low level Rest Client, vous devez utiliser le composant Elasticsearch Low level Rest Client.

2. Configuration et exécution du contexte Camel

La configuration commence par la création d'un nouveau contexte Camel à l'aide de la classe DefaultCamelContext, qui sert de base à la définition et à l'exécution des itinéraires. Ensuite, nous configurons le composant Elasticsearch, qui permettra à Apache Camel d'interagir avec un cluster Elasticsearch. L'instance ESlasticsearchComponent est configurée pour se connecter à l'adresse localhost:9200, qui est l'adresse par défaut d'un cluster Elasticsearch local. Dans le cas d'un environnement nécessitant une authentification, il convient de lire la documentation relative à la configuration du composant et à l'activation de l'authentification de base, appelée ". Configurer le composant et activer l'authentification de base".

Ce composant est ensuite ajouté au contexte Camel, ce qui permet aux routes définies d'utiliser ce composant pour effectuer des opérations dans Elasticsearch.

Ensuite, les itinéraires sont ajoutés au contexte. Nous allons créer des itinéraires pour l'indexation, la mise à jour et la suppression de documents en masse.

3. Configuration des itinéraires Camel

Indexation des données

La première route que nous allons configurer est destinée à l'indexation des données. Nous utiliserons un fichier JSON contenant un catalogue de films. La route sera configurée pour lire le fichier situé à l'adresse src/main/resources/movies.json, désérialiser le contenu JSON en objets Java, puis appliquer une stratégie d'agrégation pour combiner plusieurs messages en un seul, permettant ainsi des opérations par lots dans Elasticsearch. La taille de 500 éléments par message a été configurée, c'est-à-dire que le bulk indexera 500 films à la fois.

Route Elasticsearch Operation Bulk

Le lot de documents sera envoyé au point de terminaison de l'opération en bloc d'Elasticsearch. Cette approche garantit l'efficacité et la rapidité du traitement de grands volumes de données.

Mise à jour des données

La prochaine étape consistera à mettre à jour les documents. Nous avons indexé quelques films dans l'étape précédente et nous allons maintenant créer de nouvelles routes pour rechercher un document par code de référence et mettre à jour le champ "rating".

Nous mettons en place un contexte Camel (DefaultCamelContext), dans lequel un composant Elasticsearch est enregistré et une route personnalisée IngestionRoute est ajoutée. L'opération commence par l'envoi du code du document par le ProducerTemplate, qui lance la route à partir du point de terminaison direct:update-ingestion.

Ensuite, nous avons l'IngestionRoute, qui est le point d'entrée de ce flux. La route effectue plusieurs opérations en pipeline. Tout d'abord, une recherche dans Elasticsearch est effectuée pour localiser le document par code (direct:search-by-id), où le SearchByCodeProcessor assemble la requête sur la base du code. Ensuite, le document récupéré est traité par l'UpdateRatingProcessor, qui convertit le résultat en objets Movie, met à jour la classification du film avec une valeur spécifique et prépare le document mis à jour à renvoyer à Elasticsearch pour la mise à jour.

Le processeur SearchByCodeProcessor a été configuré uniquement pour exécuter la requête de recherche :

Le processeur UpdateRatingProcessor est responsable de la mise à jour du champ "rating".

Suppression des données

Enfin, l'itinéraire de suppression des documents est configuré. Ici, nous allons supprimer un document en utilisant son ID. Dans Elasticsearch, pour supprimer un document, il faut connaître l'identifiant du document, l'index dans lequel le document est stocké et exécuter une requête Delete. Dans Apache Camel, nous effectuerons cette opération en créant une nouvelle route, comme indiqué ci-dessous.

L'itinéraire part du point d'arrivée direct:op-delete, qui sert de point d'entrée. Lorsqu'un document doit être supprimé, son identifiant (_id) est reçu dans le corps du message. La route définit ensuite l'en-tête indexId avec la valeur de cet identifiant en utilisant le simple("${body}"), qui extrait l'_id du corps du message.

Enfin, le message est dirigé vers le point de terminaison spécifié par URI_DELETE_OPERATION, qui se connecte à Elasticsearch pour effectuer l'opération de suppression du document dans l'index correspondant.
Maintenant que nous avons créé la route, nous pouvons créer un contexte Camel (DefaultCamelContext), qui est configuré pour inclure le composant Elasticsearch.

Ensuite, la route de suppression, définie par la classe OperationDeleteRoute, est ajoutée au contexte. Une fois le contexte initialisé, une adresse ProducerTemplate est utilisée pour transmettre l'identifiant du document à supprimer à l'adresse direct:op-delete, qui déclenche la procédure de suppression.

Conclusion

L'intégration entre Apache Camel et Elasticsearch permet une ingestion robuste et efficace des données, en tirant parti de la flexibilité de Camel pour définir des itinéraires capables de gérer différents scénarios de manipulation des données, tels que l'indexation, la mise à jour et la suppression. Grâce à cette configuration, vous pouvez orchestrer et automatiser des processus complexes de manière évolutive, en veillant à ce que vos données soient gérées efficacement dans Elasticsearch. Cet exemple montre comment ces outils peuvent être utilisés ensemble pour créer une solution efficace et adaptable pour l'ingestion de données.

Références

• Apache Camel
• Architecture d'Apache Camel
• Agrégation Apache Camel
• Composant du dossier
• Composant Elasticsearch

Questions fréquentes