Debezium: Change Data Capture für Apache Kafka

In der heutigen datengetriebenen Welt ist die Fähigkeit, Daten nahtlos zwischen verschiedenen Systemen zu bewegen, von unschätzbarem Wert. Eines der Schlüsselelemente in modernen Datenarchitekturen ist Apache Kafka, ein leistungsstarkes Tool für die Verarbeitung von Streaming-Daten. Doch wie integriert man Daten aus einer Vielzahl von Datenbanken zuverlässig und in nahezu Echtzeit in Kafka? Hier kommt Debezium ins Spiel, ein Open-Source-Datenstreaming-Tool, das genau dieses Problem löst.

Debezium ermöglicht es uns, Änderungen in unseren Datenbanken zu erfassen und diese Änderungen als Event-Stream nach Kafka zu leiten. Dieser Prozess, bekannt als Change Data Capture (CDC), ist entscheidend für die Erstellung reaktiver Anwendungen, die in Echtzeit auf Datenänderungen reagieren müssen.

Während es eine Vielfalt von Werkzeugen gibt um Daten nach Kafka zu befördern, möchten wir eine besondere Anmerkung zum JDBC Source Connector machen. In unseren Schulungen weisen wir stets darauf hin, dass wir mit diesem speziellen Connector vorsichtig sein sollten. Der Grund dafür liegt unter anderem in seiner simplen Abfrage-Methodik, wie zum Beispiel SELECT * FROM TABLE WHERE id/timestamp > ?, die uns vor Herausforderungen stellt, wenn es darum geht, UPDATEs effektiv zu erfassen und DELETEs zu bemerken.

Außerdem freuen sich unsere Datenbank-Admins nicht unbedingt über diese Zusatzlast auf der Datenbank, aber es gibt eine bessere Lösung: Nahezu jede Datenbank hat intern ein Log, in das jede Änderung geschrieben wird. Manche Datenbanken nennen es das Transaktions-Log manche das Commit-Log oder manche Write Ahead Log.

Datenbanken nutzen dieses Log, um für höhere Zuverlässigkeit zu sorgen oder auch um Daten zwischen Replicas zu replizieren. Mithilfe von Debezium können wir uns an dieses Log anschließen und Daten aus der Datenbank dann schreiben, wenn sie sich verändern, also bei einem INSERT einem UPDATE und gar einem DELETE.

Dabei ist Debezium lediglich ein Kafka Connect Plugin und unterstützt zahlreiche Datenbanksysteme wie PostgreSQL, MySQL, Oracle, Microsoft SQL Server oder gar MongoDB. Da Debezium nur aus dem Transaktions-Log liest (zumindest bei den meisten Datenbanken) und keine Queries abfeuert, ist der Overhead Debeziums meistens vernachlässigbar.

Der Nachteil ist aber, dass ein paar Vorbereitungen zu tun sind. Lasst uns das gemeinsam anhand des Beispiels PostgreSQL anschauen. Im Lab ist das schon vorbereitet. Als Erstes müssen wir das Log Level für den Write Ahead Log (wal_level) erhöhen. Der Default ist replica, welches ausreicht, um Daten auf Read-Only Replicas zu replizieren. Aber wir benötigen für Debezium ein paar mehr Daten und erhöhen das Level auf logical. Danach müssen wir aber Postgres neu starten.

Als Nächstes müssen wir unserem Datenbanknutzer die Rechte geben, Daten zu replizieren. Dafür fügen wir dem Nutzer einfach die Rolle REPLICATION hinzu.

Natürlich umschifft Debezium dabei nicht alle Sicherheitsmechanismen und gibt Zugriff auf alle Daten in allen Tabellen, sondern es erstellt eine sogenannte Publication, in die die Daten für die benötigten Tabellen gesammelt sind. Das wird aber automatisch von Debezium erledigt, und wir müssen es nicht manuell tun.

Als Nächstes konfigurieren wir uns den Connector. Wir müssen die Datenbank Zugangsdaten setzen und definieren, welche Tabellen verbunden beziehungsweise nicht verbunden werden sollen. Zu guter Letzt setzen wir das Topic Präfix, denn Debezium erstellt uns Topics nach dem Muster prefix.schema.table. Das können wir aber auch ändern. Hier ist ein Beispiel für eine einfache Konfiguration:

Dieser Connector schreibt alle Daten aus allen Tabellen des public-Schemas der user-Datenbank auf localhost nach Kafka. Dabei werden die Kafka-Topics nach dem folgenden Muster benannt: dbz.public.[table-name].

Nachdem wir also den Connector erstellt haben, fängt Debezium an, die Daten zu replizieren. Das Problem ist aber, dass im Write Ahead Log nicht die Daten für alle Ewigkeit aufbewahrt werden. Deshalb macht Debezium beim ersten Start einen initialen Snapshot aller Daten in allen angefragten Tabellen. Und erst danach fängt der Change Data Capture-Teil an. Debezium garantiert uns sogar, dass dies konsistent passiert.

Wie sehen aber die Daten aus, die Debezium schreibt? Denn Debezium schreibt nicht nur die Veränderung, sondern ein paar mehr Informationen. Das Schema (schema) der Daten wird auch in jeder Nachricht gespeichert.

Wir können eine Schema Registry benutzen, um diese Daten nicht jedes Mal mitzuschicken. Dazu aber etwas mehr in einem späteren Modul. Der eigentliche Payload (payload) besteht aus einigen Metadaten zur Datenquelle, dem source Block, der Operation(op), die diese Nachricht ausgelöst hat. Also war das ein Create (c), Update (u), Delete (d) oder ein Read (r)? Reads passieren aber nur beim Snapshot und natürlich den Timestamp (ts_ms). Mehr Infos findet ihr im Debezium Tutorial.

Die wichtigsten Felder sind aber before, also der Zustand des Eintrags vor der Operation und after, der Zustand des Eintrags nach der Operation. Aber oft interessieren uns die meisten Daten hier gar nicht, dafür können wir auch das Event Flattening SMT benutzen, um die ganzen Zusatzdaten nicht mitzuschreiben. Dann wird Debezium lediglich den after Payload nach Kafka schreiben.