Guide pratique d'Apache Camel avec Quarkus : créer une application ETL

Je suis ravi de vous présenter une série d'articles sur Apache Camel. Dans ce premier article, plutôt que d'approfondir les complexités d'Apache Camel, je présenterai un cas d'utilisation pratique pour montrer ses capacités. Plus précisément, vous apprendrez à créer une application ETL (Extract, Transform, and Load) simple entre deux bases de données à l'aide d'Apache Camel.

Introduction à Apache Camel – Bref aperçu

Avant de plonger dans le cas d'utilisation pratique, présentons brièvement Apache Camel. Apache Camel est un framework d'intégration open source qui exploite les modèles d'intégration d'entreprise (EIP) pour faciliter l'intégration de divers systèmes.

Dans le monde d’aujourd’hui, de nombreux systèmes de types différents coexistent. Certains peuvent être des systèmes existants, tandis que d'autres sont nouveaux. Ces systèmes doivent souvent interagir et s'intégrer les uns aux autres, ce qui peut s'avérer difficile en raison des différentes implémentations et formats de message. Une solution consiste à écrire du code personnalisé pour combler ces différences, mais cela peut entraîner des difficultés de couplage étroit et de maintenance.

Au lieu de cela, Apache Camel offre une couche supplémentaire pour atténuer les différences entre les systèmes, ce qui entraîne un couplage lâche et une maintenance plus facile. Camel utilise une API (ou Java Domain Specific Language déclaratif) pour configurer les règles de routage et de médiation basées sur EIP.

Modèles d'intégration d'entreprise (EIP)

Pour comprendre Apache Camel, il est important de comprendre les « modèles d'intégration d'entreprise » (EIP). Le livre « Enterprise Integration Patterns » décrit un ensemble de modèles permettant de concevoir de grands systèmes basés sur des composants dans lesquels les composants peuvent s'exécuter dans le même processus ou sur des machines différentes. L'idée clé est que les systèmes doivent être orientés messages, avec des composants communiquant via des messages. Les modèles fournissent une boîte à outils pour mettre en œuvre ces communications (Figure 1).

Figure 1 – Éléments de base d'une solution d'intégration (enterpriseintegrationpatterns.com)

Terminologies clés dans Apache Camel

• EndPoint : Un point de terminaison est un canal par lequel les messages sont envoyés et reçus. Il sert d'interface entre un composant et le monde extérieur.

• Message : Un message est une structure de données utilisée pour la communication entre les systèmes, composée d'un en-tête et d'un corps. L'en-tête contient des métadonnées et le corps contient les données réelles.

• Canal : Un canal connecte deux points de terminaison, facilitant l'envoi et la réception de messages.

• Routeur : Un routeur dirige les messages d'un point de terminaison à un autre, déterminant le chemin du message.

• Traducteur : Un traducteur convertit les messages d'un format à un autre.

J'envisage de m'arrêter ici pour l'instant sur l'introduction à Apache Camel. Montrons maintenant comment créer une application ETL simple entre deux bases de données à l'aide d'Apache Camel.

Énoncé du problème

Supposons que nous ayons un système très chargé dans lequel un composant critique est la base de données. À un moment donné, nous devons traiter ces données en dehors des cas opérationnels habituels - former des modèles ML, générer des exportations, des graphiques, ou nous avons simplement besoin d'une partie des données. Bien sûr, cela alourdirait encore plus notre base de données opérationnelle, et à cette fin, il serait optimal de disposer d'un mécanisme avec lequel nous pourrions extraire les données nécessaires, les transformer sous la forme dont nous avons besoin et les stocker dans une autre base de données - autre que celui opérationnel. Avec cette stratégie, nous résolvons les problèmes de surcharge potentielle excessive de notre base opérationnelle. De plus, avec un tel mécanisme, nous pouvons effectuer cette opération à un moment où notre système n'est pas trop chargé (par exemple la nuit).

Présentation de la solution

La solution est présentée dans le schéma ci-dessous (Figure 2). Nous utiliserons Apache Camel pour créer une application ETL simple entre deux bases de données. L'application extraira les données d'une base de données source, les transformera et les chargera dans une base de données cible. Nous pouvons introduire différentes stratégies pour mettre en œuvre cette solution, en nous concentrant sur la manière d'extraire les données de la base de données source. Je suppose que les critères de sélection des données seront basés sur la date de modification du dossier. Cette stratégie offre également la possibilité d'extraire les données qui ont été modifiées.

Figure 2 – Synchronisation des données entre deux bases de données à l'aide d'Apache Camel

Les bases de données source et cible auront la structure de table suivante :

CREATE TABLE IF NOT EXISTS user
(
    id            serial PRIMARY KEY,
    username      VARCHAR(50)  NOT NULL,
    password      VARCHAR(50)  NOT NULL,
    email         VARCHAR(255) NOT NULL,
    created_at    timestamp default now()::timestamp without time zone,
    last_modified TIMESTAMP DEFAULT now()::timestamp without time zone
);

Dans la base de données cible, nous transformerons le nom d'utilisateur en majuscule avant de l'insérer.

Nous utiliserons les extensions Camel Quarkus pour divers composants Camel. Plus précisément, nous utiliserons le composant Camel SQL pour interagir avec les bases de données. Le composant SQL prend en charge l'exécution de requêtes, d'insertions, de mises à jour et de suppressions SQL.

Tout d'abord, créez une classe qui étend RouteBuilder et remplacez la méthode de configuration :

@ApplicationScoped
public class UserRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        // your code here
    }
}

L'utilisation de l'annotation @ApplicationScoped n'est pas obligatoire ici, mais je préfère indiquer que la classe est un bean CDI et doit être gérée par le conteneur CDI.

Comme je l'ai mentionné ci-dessus, nous utiliserons le composant Camel SQL pour interagir avec les bases de données. Nous devons configurer le composant Camel SQL pour nous connecter aux bases de données source et cible. Nous utiliserons l'extension Quarkus Agroal pour configurer les sources de données. L'extension Agroal fournit un pool de connexions pour les sources de données. Nous allons configurer les sources de données dans le fichier application.properties.

#
# Source Database Configuration
quarkus.datasource.source_db.db-kind=postgresql
quarkus.datasource.source_db.jdbc.url=jdbc:postgresql://localhost:5001/demo
quarkus.datasource.source_db.username=test
quarkus.datasource.source_db.password=password1
#
#
# Target Database Configuration
quarkus.datasource.target_db.db-kind=postgresql
quarkus.datasource.target_db.jdbc.url=jdbc:postgresql://localhost:6001/demo
quarkus.datasource.target_db.username=test
quarkus.datasource.target_db.password=password1
#

Nous pouvons maintenant configurer le composant Camel SQL pour se connecter aux bases de données source et cible. Nous utiliserons le composant SQL pour créer un point de terminaison SQL pour les bases de données source et cible.

Le composant SQL utilise la notation URI de point de terminaison suivante :

sql:select * from table where id=# order by name[?options]

Mais nous avons besoin d'un mécanisme pour exécuter l'opération automatiquement. Nous utiliserons le composant timer pour déclencher le processus ETL toutes les secondes. Le composant timer est utilisé pour générer des échanges de messages lorsqu'un timer se déclenche. Le composant timer utilise la notation URI de point de terminaison suivante :

timer:name[?options]

Dans notre itinéraire, nous utilisons la configuration comme suit :

 from("timer://userSync?delay={{etl.timer.delay}}&period={{etl.timer.period}}")

Les {{etl.timer.delay}} et {{etl.timer.period}} sont les valeurs de configuration que nous définirons dans le fichier application.properties.

etl.timer.period=10000
etl.timer.delay=1000

Afin de transformer les données avant de les insérer dans la base de données cible, nous devons fournir à notre traducteur :

.process(exchange -> {
    final Map<String, Object> rows = exchange.getIn().getBody(Map.class);

    final String userName = (String) rows.get("username");

    final String userNameToUpperCase = userName.toUpperCase();

    log.info("User name: {} converted to upper case: {}", userName, userNameToUpperCase);

    rows.put("username", userNameToUpperCase);
})

L'interface du processeur est utilisée pour implémenter des consommateurs d'échanges de messages ou pour implémenter un traducteur de messages et d'autres cas d'utilisation.

Et voilà, nous avons une simple application ETL entre deux bases de données utilisant Apache Camel.

Lorsque vous exécutez l'application, vous devriez voir le résultat suivant dans les journaux :

2024-06-09 13:15:49,257 INFO  [route1] (Camel (camel-1) thread #1 - timer://userSync) Extracting Max last_modified value from source database
2024-06-09 13:15:49,258 INFO  [route1] (Camel (camel-1) thread #1 - timer://userSync) No record found in target database
2024-06-09 13:15:49,258 INFO  [route2] (Camel (camel-1) thread #1 - timer://userSync) The last_modified from source DB: 
2024-06-09 13:15:49,274 INFO  [route2] (Camel (camel-1) thread #1 - timer://userSync) Extracting records from source database
2024-06-09 13:15:49,277 INFO  [org.iqn.cam.rou.UserRoute] (Camel (camel-1) thread #1 - timer://userSync) User name: john_doe converted to upper case: JOHN_DOE
2024-06-09 13:15:49,282 INFO  [org.iqn.cam.rou.UserRoute] (Camel (camel-1) thread #1 - timer://userSync) User name: jane_smith converted to upper case: JANE_SMITH
2024-06-09 13:15:49,283 INFO  [org.iqn.cam.rou.UserRoute] (Camel (camel-1) thread #1 - timer://userSync) User name: alice_miller converted to upper case: ALICE_MILLER

Vous pouvez retrouver le code source complet de l'application dans le référentiel GitHub.

Conclusion

Avec cette configuration, nous avons créé une application ETL simple utilisant Apache Camel qui extrait les données d'une base de données source, les transforme et les charge dans une base de données cible. Cette approche permet de réduire la charge sur la base de données opérationnelle et nous permet d'effectuer l'extraction de données en dehors des heures de pointe.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!