Praktischer Leitfaden für Apache Camel mit Quarkus: Erstellen einer ETL-Anwendung

Ich freue mich, eine Artikelserie über Apache Camel vorstellen zu können. In diesem ersten Beitrag werde ich mich nicht eingehend mit der Komplexität von Apache Camel befassen, sondern einen praktischen Anwendungsfall vorstellen, um seine Fähigkeiten zu demonstrieren. Konkret erfahren Sie, wie Sie mit Apache Camel eine einfache ETL-Anwendung (Extract, Transform, and Load) zwischen zwei Datenbanken erstellen.

Einführung in Apache Camel – Kurzer Überblick

Bevor wir uns mit dem praktischen Anwendungsfall befassen, stellen wir kurz Apache Camel vor. Apache Camel ist ein Open-Source-Integrationsframework, das Enterprise Integration Patterns (EIP) nutzt, um die Integration verschiedener Systeme zu erleichtern.

In der heutigen Welt existieren zahlreiche Systeme unterschiedlicher Art nebeneinander. Bei einigen handelt es sich möglicherweise um Legacy-Systeme, während andere neu sind. Diese Systeme müssen häufig miteinander interagieren und sich integrieren, was aufgrund unterschiedlicher Implementierungen und Nachrichtenformate eine Herausforderung darstellen kann. Eine Lösung besteht darin, benutzerdefinierten Code zu schreiben, um diese Unterschiede zu überbrücken. Dies kann jedoch zu engen Kopplungs- und Wartungsschwierigkeiten führen.

Stattdessen bietet Apache Camel eine zusätzliche Ebene zur Vermittlung der Unterschiede zwischen Systemen, was zu einer losen Kopplung und einer einfacheren Wartung führt. Camel verwendet eine API (oder eine deklarative Java Domain Specific Language), um Routing- und Vermittlungsregeln basierend auf EIP zu konfigurieren.

Unternehmensintegrationsmuster (EIP)

Um Apache Camel zu verstehen, ist es wichtig, „Enterprise Integration Patterns“ (EIP) zu verstehen. Das Buch „Enterprise Integration Patterns“ beschreibt eine Reihe von Mustern zum Entwerfen großer, komponentenbasierter Systeme, bei denen Komponenten im selben Prozess oder auf verschiedenen Maschinen ausgeführt werden können. Der Grundgedanke besteht darin, dass Systeme nachrichtenorientiert sein sollten, wobei die Komponenten über Nachrichten kommunizieren. Die Muster stellen ein Toolkit für die Implementierung dieser Kommunikation dar (Abbildung 1).

Abbildung 1 – Grundelemente einer Integrationslösung (enterpriseintegrationpatterns.com)

Wichtige Terminologien in Apache Camel

• EndPoint: Ein Endpunkt ist ein Kanal, über den Nachrichten gesendet und empfangen werden. Es dient als Schnittstelle zwischen einer Komponente und der Außenwelt.

• Nachricht: Eine Nachricht ist eine Datenstruktur, die für die Kommunikation zwischen Systemen verwendet wird und aus einem Header und einem Text besteht. Der Header enthält Metadaten und der Text enthält die eigentlichen Daten.

• Kanal: Ein Kanal verbindet zwei Endpunkte und erleichtert das Senden und Empfangen von Nachrichten.

• Router: Ein Router leitet Nachrichten von einem Endpunkt zum anderen und bestimmt den Nachrichtenpfad.

• Übersetzer: Ein Übersetzer konvertiert Nachrichten von einem Format in ein anderes.

Ich überlege, hier mit der Einführung in Apache Camel vorerst aufzuhören. Jetzt zeigen wir Ihnen, wie Sie mit Apache Camel eine einfache ETL-Anwendung zwischen zwei Datenbanken erstellen.

Problemstellung

Nehmen wir an, wir haben ein hoch ausgelastetes System, bei dem eine kritische Komponente die Datenbank ist. Irgendwann müssen wir uns mit diesen Daten außerhalb der üblichen Betriebsfälle befassen – Training von ML-Modellen, Generierung von Exporten, Diagrammen oder wir benötigen einfach einen Teil der Daten. Dies würde natürlich unsere Betriebsdatenbank noch mehr belasten, und zu diesem Zweck wäre es optimal, über einen Mechanismus zu verfügen, mit dem wir die erforderlichen Daten extrahieren, in die von uns benötigte Form umwandeln und in einer anderen Datenbank speichern können – einer anderen als die operative. Mit dieser Strategie lösen wir die Probleme einer übermäßigen potenziellen Überlastung unserer operativen Basis. Darüber hinaus können wir mit einem solchen Mechanismus diesen Vorgang zu einem Zeitpunkt durchführen, an dem unser System nicht zu stark belastet ist (z. B. nachts).

Lösungsübersicht

Die Lösung ist im Diagramm unten dargestellt (Abbildung 2). Wir werden Apache Camel verwenden, um eine einfache ETL-Anwendung zwischen zwei Datenbanken zu erstellen. Die Anwendung extrahiert Daten aus einer Quelldatenbank, wandelt sie um und lädt sie in eine Zieldatenbank. Wir können verschiedene Strategien zur Implementierung dieser Lösung einführen und uns dabei auf die Extraktion der Daten aus der Quelldatenbank konzentrieren. Ich gehe davon aus, dass die Kriterien für die Auswahl der Daten auf dem Änderungsdatum des Datensatzes basieren. Diese Strategie bietet die Möglichkeit, auch die geänderten Daten zu extrahieren.

Abbildung 2 – Synchronisieren von Daten zwischen zwei Datenbanken mit Apache Camel

Quell- und Zieldatenbanken haben die folgende Tabellenstruktur:

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
);

In der Zieldatenbank wandeln wir den Benutzernamen vor dem Einfügen in Großbuchstaben um.

Wir werden Camel Quarkus-Erweiterungen für verschiedene Camel-Komponenten verwenden. Konkret verwenden wir die Camel SQL-Komponente, um mit den Datenbanken zu interagieren. Die SQL-Komponente unterstützt die Durchführung von SQL-Abfragen, Einfügungen, Aktualisierungen und Löschungen.

Erstellen Sie zunächst eine Klasse, die RouteBuilder erweitert, und überschreiben Sie die Konfigurationsmethode:

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

Die Verwendung der Annotation @ApplicationScoped ist hier nicht zwingend erforderlich, aber ich möchte lieber angeben, dass die Klasse eine CDI-Bean ist und vom CDI-Container verwaltet werden sollte.

Wie ich oben erwähnt habe, werden wir die Camel SQL-Komponente verwenden, um mit den Datenbanken zu interagieren. Wir müssen die Camel SQL-Komponente konfigurieren, um eine Verbindung zu den Quell- und Zieldatenbanken herzustellen. Wir werden die Quarkus Agroal-Erweiterung verwenden, um die Datenquellen zu konfigurieren. Die Agroal-Erweiterung stellt einen Verbindungspool für die Datenquellen bereit. Wir werden die Datenquellen in der Datei application.properties konfigurieren.

#
# 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
#

Jetzt können wir die Camel SQL-Komponente so konfigurieren, dass sie eine Verbindung zu den Quell- und Zieldatenbanken herstellt. Wir werden die SQL-Komponente verwenden, um einen SQL-Endpunkt für die Quell- und Zieldatenbanken zu erstellen.

Die SQL-Komponente verwendet die folgende Endpunkt-URI-Notation:

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

Aber wir brauchen einen Mechanismus, um den Vorgang automatisch auszuführen. Wir werden die Timer-Komponente verwenden, um den ETL-Prozess alle Sekunden auszulösen. Die Timer-Komponente wird verwendet, um den Nachrichtenaustausch zu generieren, wenn ein Timer ausgelöst wird. Die Timer-Komponente verwendet die folgende Endpunkt-URI-Notation:

timer:name[?options]

In unserer Route verwenden wir die Konfiguration wie folgt:

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

{{etl.timer.delay}} und {{etl.timer.period}} sind die Konfigurationswerte, die wir in der Datei application.properties definieren werden.

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

Um die Daten vor dem Einfügen in die Zieldatenbank umzuwandeln, müssen wir unseren Übersetzer bereitstellen:

.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);
})

Die Prozessorschnittstelle wird verwendet, um Konsumenten des Nachrichtenaustauschs oder einen Nachrichtenübersetzer und andere Anwendungsfälle zu implementieren.

Und voilà, wir haben eine einfache ETL-Anwendung zwischen zwei Datenbanken mit Apache Camel.

Wenn Sie die Anwendung ausführen, sollten Sie die folgende Ausgabe in den Protokollen sehen:

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

Den vollständigen Quellcode der Anwendung finden Sie im GitHub-Repository.

Abschluss

Mit diesem Setup haben wir mit Apache Camel eine einfache ETL-Anwendung erstellt, die Daten aus einer Quelldatenbank extrahiert, sie transformiert und in eine Zieldatenbank lädt. Dieser Ansatz trägt dazu bei, die Belastung der Betriebsdatenbank zu reduzieren und ermöglicht uns die Datenextraktion außerhalb der Spitzenzeiten.

Das obige ist der detaillierte Inhalt vonPraktischer Leitfaden für Apache Camel mit Quarkus: Erstellen einer ETL-Anwendung. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!