In meiner Arbeit stoße ich häufig auf neue technische Lösungen und Softwareprodukte, über die es im deutschsprachigen Internet nur wenig Informationen gibt. Mit diesem Artikel möchte ich eine solche Lücke schließen und ein Beispiel aus meiner jüngsten Praxis vorstellen, als es darum ging, die Übertragung von CDC-Ereignissen aus zwei beliebten Datenbanksystemen (PostgreSQL und MongoDB) an ein Kafka-Cluster mittels Debezium einzurichten. Ich hoffe, dass dieser Überblick auf der Basis der durchgeführten Arbeiten auch anderen hilfreich sein wird.
Was ist Debezium und was bedeutet CDC?
ist ein Vertreter der Kategorie CDC-Software (), genauer gesagt, handelt es sich um eine Sammlung von Connectoren für verschiedene Datenbanksysteme, die mit dem Apache Kafka Connect-Framework kompatibel sind.
Das das unter der Apache License v2.0 lizenziert ist und von Red Hat gesponsert wird. Die Entwicklung begann im Jahr 2016, und derzeit wird die Unterstützung für folgende Datenbanksysteme offiziell angeboten: MySQL, PostgreSQL, MongoDB, SQL Server. Es gibt auch Connectoren für Cassandra und Oracle, jedoch befinden sich diese derzeit im Status „Early Access“, und neue Versionen garantieren keine Rückwärtskompatibilität.
Im Vergleich zu traditionellen Ansätzen, bei denen Anwendungen direkt aus Datenbanken lesen, bietet CDC wesentliche Vorteile wie das Streamen von Datenänderungen auf Zeilenebene mit niedriger Latenz sowie hohe Zuverlässigkeit und Verfügbarkeit. Letztere beiden Punkte werden durch die Verwendung eines Kafka-Clusters als Speicher für CDC-Ereignisse erreicht.
Ein weiterer Vorteil ist, dass eine einheitliche Modellierung für die Speicherung von Ereignissen verwendet wird. Dadurch muss sich die Endanwendung nicht um die spezifischen Anforderungen verschiedener Datenbanksysteme kümmern.
Dank des Einsatzes eines Nachrichtenbrokers eröffnet sich die Möglichkeit zur horizontalen Skalierung von Anwendungen, die Datenänderungen verfolgen. Dabei wird der Einfluss auf die Datenquelle minimiert, da die Daten nicht direkt aus der Datenbank, sondern aus dem Kafka-Cluster abgerufen werden.
Über die Architektur von Debezium
Die Verwendung von Debezium lässt sich einfach wie folgt beschreiben:
Datenbank (als Datenquelle) → Connector in Kafka Connect → Apache Kafka → Consumer
Zur Veranschaulichung folgt hier ein Diagramm von der Projektseite:
Allerdings gefällt mir dieses Schema nicht besonders, da der Eindruck entsteht, dass nur der Einsatz eines Sink-Connectors möglich ist.
In Wirklichkeit sieht die Situation jedoch anders aus: Das Befüllen Ihres Data Lakes (das letzte Glied im obigen Schema) ist nicht die einzige Anwendungsmöglichkeit von Debezium. Die in Apache Kafka gesendeten Ereignisse können von Ihren Anwendungen genutzt werden, um verschiedene Situationen zu lösen. Beispielsweise:
- Entfernung nicht mehr relevanter Daten aus dem Cache;
- Versand von Benachrichtigungen;
- Aktualisierungen von Suchindizes;
- ein gewisses Äquivalent zu Audit-Logs;
- …
Falls Sie eine Anwendung in Java haben und nicht die Notwendigkeit/Möglichkeit besteht, einen Kafka-Cluster zu nutzen, gibt es auch die Möglichkeit, über den zu arbeiten. Der offensichtliche Vorteil besteht darin, dass man auf zusätzliche Infrastruktur (in Form eines Connectors und Kafka) verzichten kann. Diese Lösung wurde jedoch mit Version 1.1 als veraltet erklärt und wird nicht mehr empfohlen (in zukünftigen Releases könnte die Unterstützung eingestellt werden).
In diesem Artikel wird die von den Entwicklern empfohlene Architektur behandelt, die Ausfallsicherheit und Skalierbarkeit bietet.
Konfiguration des Connectors
Um die Veränderungen des wertvollsten Guts – der Daten – zu verfolgen, benötigen wir:
- eine Datenquelle, die MySQL ab Version 5.7, PostgreSQL 9.6+ oder MongoDB 3.2+ sein kann ();
- einen Apache Kafka-Cluster;
- eine Kafka Connect-Instanz (Versionen 1.x, 2.x);
- einen konfigurierten Debezium-Connector.
Die ersten beiden Punkte, d.h. der Installationsprozess für die DBMS und Apache Kafka, sind nicht Teil dieses Artikels. Für diejenigen, die alles in einer Sandbox bereitstellen möchten, gibt es im offiziellen Repository mit Beispielen eine fertige .
Hier konzentrieren wir uns ausführlicher auf die letzten beiden Punkte.
0. Kafka Connect
Im Folgenden werden alle Konfigurationsbeispiele im Kontext des von den Debezium-Entwicklern veröffentlichten Docker-Images betrachtet. Es enthält alle notwendigen Plugin-Dateien (Connectoren) und ermöglicht die Konfiguration von Kafka Connect über Umgebungsvariablen.
Sollte Kafka Connect von Confluent verwendet werden, müssen die Plugins der erforderlichen Connectoren manuell in das Verzeichnis hinzugefügt werden, das in plugin.path oder über die Umgebungsvariable CLASSPATH. Die Einstellungen für den Kafka Connect Worker und die Connectoren werden über Konfigurationsdateien definiert, die als Argumente an den Startbefehl des Workers übergeben werden. Weitere Informationen finden Sie in .
Der gesamte Prozess zur Einrichtung von Debezium mit einem Connector umfasst zwei Schritte. Lassen Sie uns jeden von ihnen betrachten:
1. Konfiguration des Kafka Connect Frameworks
Für das Streaming von Daten in ein Apache Kafka-Cluster im Kafka Connect Framework werden spezifische Parameter festgelegt, wie zum Beispiel:
- Verbindungsparameter für das Cluster,
- Themen, in denen die Konfiguration des Connectors gespeichert wird,
- der Gruppenname, unter dem der Connector läuft (im Falle der Verwendung des verteilten Modus).
Das offizielle Docker-Image des Projekts unterstützt die Konfiguration über Umgebungsvariablen - und das werden wir auch nutzen. Also laden wir das Image herunter:
docker pull debezium/connectDer minimale Satz an Umgebungsvariablen, der benötigt wird, um den Connector zu starten, sieht folgendermaßen aus:
-
BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092– die initiale Liste der Kafka-Cluster-Server, um die vollständige Liste der Cluster-Mitglieder zu erhalten; -
OFFSET_STORAGE_TOPIC=connector-offsets– das Thema zur Speicherung der Positionen, an denen sich der Connector momentan befindet; -
CONNECT_STATUS_STORAGE_TOPIC=connector-status— Thema zur Speicherung des Status des Connectors und seiner Aufgaben; -
CONFIG_STORAGE_TOPIC=connector-config— Thema zur Speicherung der Konfigurationsdaten des Connectors und seiner Aufgaben; -
GROUP_ID=1— Identifikator der Worker-Gruppe, auf der die Aufgabe des Connectors ausgeführt werden kann; erforderlich bei Verwendung des verteilten (distributed) Betriebs.
Wir starten den Container mit diesen Variablen:
docker run
-e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092'
-e GROUP_ID=1
-e CONFIG_STORAGE_TOPIC=my_connect_configs
-e OFFSET_STORAGE_TOPIC=my_connect_offsets
-e STATUS_STORAGE_TOPIC=my_connect_statuses debezium/connect:1.2Hinweis zu Avro
Standardmäßig schreibt Debezium Daten im JSON-Format, was für Sandbox-Umgebungen und kleine Datenmengen akzeptabel ist, jedoch in stark belasteten Datenbanken problematisch werden kann. Eine Alternative zum JSON-Konverter ist die Serialisierung von Nachrichten mithilfe von in ein binäres Format, was die Belastung des I/O-Subsystems in Apache Kafka verringert.
Für die Nutzung von Avro ist die Bereitstellung eines separaten (zur Speicherung von Schemata) erforderlich. Die Variablen für den Konverter sehen wie folgt aus:
name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER
value: io.confluent.connect.avro.AvroConverterDie Details zur Verwendung von Avro und dessen Registrierung gehen über den Rahmen dieses Artikels hinaus — zur Veranschaulichung werden wir weiter unten JSON verwenden.
2. Konfiguration des Connectors
Jetzt können wir direkt zur Konfiguration des Connectors übergehen, der Daten aus der Quelle lesen wird.
Lassen Sie uns am Beispiel von zwei Datenbanksystemen: PostgreSQL und MongoDB betrachten — für die ich Erfahrung habe und bei denen es Unterschiede gibt (auch wenn diese geringfügig sind, sind sie in manchen Fällen doch erheblich!).
Die Konfiguration wird in JSON-Notation beschrieben und über einen POST-Request in Kafka Connect geladen.
2.1. PostgreSQL
Beispiel für die Konfiguration des Connectors für PostgreSQL:
{
"name": "pg-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"plugin.name": "pgoutput",
"database.hostname": "127.0.0.1",
"database.port": "5432",
"database.user": "debezium",
"database.password": "definitelynotpassword",
"database.dbname" : "dbname",
"database.server.name": "pg-dev",
"table.include.list": "public.(.*)",
"heartbeat.interval.ms": "5000",
"slot.name": "dbname_debezium",
"publication.name": "dbname_publication",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": "pg-dev.public.(.*)",
"transforms.AddPrefix.replacement": "data.cdc.dbname"
}
}Die Funktionsweise des Connectors nach dieser Konfiguration ist recht einfach:
- Bei der ersten Ausführung verbindet sich der Connector mit der in der Konfiguration angegebenen Datenbank und startet im initialen Snapshot, dabei sendet er einen anfänglichen Datensatz in Kafka, der durch die Bedingung
SELECT * FROM table_name. - erhalten wurde. Nachdem die Initialisierung abgeschlossen ist, wechselt der Connector in den Modus, in dem Änderungen aus den WAL-Dateien von PostgreSQL gelesen werden.
Zu den verwendeten Optionen:
-
name— der Name des Connectors, für den die unten beschriebene Konfiguration verwendet wird; dieser Name wird später verwendet, um mit dem Connector zu arbeiten (d.h. Status zu überprüfen/neu zu starten/die Konfiguration über die REST API von Kafka Connect zu aktualisieren); -
connector.class— Connector-Klasse für das DBMS, die vom konfigurierbaren Connector verwendet wird; -
plugin.name— Name des Plugins zur logischen Dekodierung von Daten aus WAL-Dateien. Folgende Optionen stehen zur Auswahl:wal2json,decoderbuffsundpgoutput. Die ersten beiden erfordern die Installation der entsprechenden Erweiterungen im DBMS, währendpgoutputfür PostgreSQL Version 10 und höher keine zusätzlichen Maßnahmen erforderlich sind; -
database.*— Optionen zur Verbindung mit der DB, wobeidatabase.server.name— Name der PostgreSQL-Instanz, die zur Bildung des Thementitels im Kafka-Cluster verwendet wird; -
table.include.list— Liste der Tabellen, deren Änderungen wir verfolgen möchten; formatiert alsschema.table_name; kann nicht zusammen mittable.exclude.list; -
heartbeat.interval.ms— Intervall (in Millisekunden), in dem der Connector Heartbeat-Nachrichten an ein spezielles Thema sendet; -
heartbeat.action.query— Abfrage, die bei jedem Versand einer Heartbeat-Nachricht ausgeführt wird (Option eingeführt ab Version 1.1); -
slot.name— Name des Replikationsslots, der vom Connector verwendet wird; publication.name— Name in PostgreSQL, die vom Connector verwendet wird. Falls sie nicht existiert, wird Debezium versuchen, sie zu erstellen. Wenn der Benutzer, unter dem die Verbindung erfolgt, dafür nicht über ausreichende Rechte verfügt, wird der Connector mit einem Fehler abbrechen;-
sinksbestimmt, wie der Name des Zielthemen geändert werden soll:-
transforms.AddPrefix.typegibt an, dass wir reguläre Ausdrücke verwenden werden; -
transforms.AddPrefix.regex— das Muster, nach dem der Name des Zielthemen überschrieben wird; -
transforms.AddPrefix.replacement— das, worauf wir umbenennen.
-
Weitere Informationen zu heartbeat und transforms
Standardmäßig sendet der Connector Daten an Kafka bei jeder bestätigten Transaktion und protokolliert deren LSN (Log Sequence Number) in einem Systemthema offset. Aber was passiert, wenn der Connector so konfiguriert ist, dass er nicht die gesamte Datenbank, sondern nur Teile ihrer Tabellen liest (in denen die Datenaktualisierung nicht häufig stattfindet)?
- Der Connector wird WAL-Dateien lesen und keine bestätigten Transaktionen in den Tabellen finden, die er überwacht.
- Daher wird er seine aktuelle Position weder im Thema noch im Replikationsslot aktualisieren.
- Dies führt wiederum dazu, dass WAL-Dateien auf der Festplatte „festgehalten“ werden, was möglicherweise den gesamten Speicherplatz erschöpfen kann.
Hier kommen die Optionen ins Spiel, heartbeat.interval.ms und heartbeat.action.query. Die Verwendung dieser Optionen in Kombination ermöglicht es, bei jeder Sendung einer Heartbeat-Nachricht einen Änderungsantrag in einer separaten Tabelle durchzuführen. Damit wird das LSN, an dem sich der Connector (im Replikationsslot) gerade befindet, ständig aktualisiert. Dies ermöglicht es der Datenbank, WAL-Dateien zu löschen, die nicht mehr benötigt werden. Weitere Informationen zur Funktionsweise der Optionen finden Sie in .
Eine andere Option, die mehr Aufmerksamkeit verdient, ist sinks. Auch wenn es eher um Benutzerfreundlichkeit und Ästhetik geht…
Standardmäßig erstellt Debezium Themen gemäß folgender Namenspolitik: serverName.schemaName.tableName. Dies ist nicht immer praktisch. Mit den Optionen sinks können Sie mithilfe von regulären Ausdrücken die Liste der Tabellen definieren, deren Events in ein Thema mit einem bestimmten Namen geroutet werden sollen.
In unserer Konfiguration geschieht Folgendes: Abschließend zur Beschreibung der Connector-Konfiguration für PostgreSQL sollten einige Merkmale/Einschränkungen seiner Funktionsweise erwähnt werden: Lassen Sie uns also unsere Konfiguration in den Connector laden: Wir überprüfen, ob das Laden erfolgreich war und der Connector gestartet wurde: Ausgezeichnet: Er ist konfiguriert und bereit für den Einsatz. Lassen Sie uns jetzt als Consumer auftreten und eine Verbindung zu Kafka herstellen, um dann einen Datensatz in der Tabelle hinzuzufügen und zu ändern: In unserem Thema wird dies wie folgt dargestellt: Ein sehr langes JSON mit unseren Änderungen In beiden Fällen bestehen die Aufzeichnungen aus dem Schlüssel (PK) des Datensatzes, der geändert wurde, und dem Inhalt der Änderungen: Wie der Datensatz vorher war und wie er danach aussieht. Dieser Connector verwendet den standardmäßigen Replikationsmechanismus von MongoDB, indem er Informationen aus dem oplog des primären DB-Knotens liest. Ähnlich wie beim bereits beschriebenen Connector für PgSQL wird hier beim ersten Start auch ein primärer Snapshot der Daten erstellt, danach wechselt der Connector in den Lesemodus des oplogs. Beispielkonfiguration: Wie bereits erwähnt, gibt es hier im Vergleich zum vorherigen Beispiel keine neuen Optionen, aber die Anzahl der Optionen, die sich auf die Verbindung zur Datenbank beziehen, und deren Präfixe wurde reduziert. Einstellungen Die Frage der Fehlertoleranz und Hochverfügbarkeit ist heutzutage so relevant wie nie – insbesondere wenn wir von Daten und Transaktionen sprechen, und das Tracking von Datenänderungen ist hierbei nicht unwichtig. Lassen Sie uns betrachten, was prinzipiell schiefgehen kann und was in jedem dieser Fälle mit Debezium passieren wird. Es gibt drei Ausfallvarianten: Es gibt jedoch Ausnahmen. Wenn der Connector längere Zeit offline war (oder nicht mit einer MongoDB-Instanz kommunizieren konnte) und der oLog während dieser Zeit rotiert wurde, wird der Connector beim Wiederherstellen der Verbindung ruhig von der ersten verfügbaren Position aus weiter lesen, wodurch Teile der Daten in Kafka nicht gelangen. Debezium ist meine erste Erfahrung mit CDC-Systemen und insgesamt ziemlich positiv. Das Projekt überzeugt mit der Unterstützung der gängigen DBMS, der einfachen Konfiguration, der Clusterunterstützung und einer aktiven Community. Allen, die sich für die Praxis interessieren, empfehle ich, sich die Anleitungen für und . Im Gegensatz zum JDBC-Connector für Kafka Connect bietet Debezium den Vorteil, dass Änderungen aus den Datenbankprotokollen gelesen werden, was eine Datenübertragung mit minimaler Verzögerung ermöglicht. Der JDBC-Connector (aus der Kafka Connect-Bibliothek) führt Anfragen an die überwachte Tabelle in festen Intervallen aus und generiert (aus diesem Grund) keine Nachrichten bei der Löschung von Daten (wie kann man Daten anfordern, die nicht vorhanden sind?). Für vergleichbare Aufgaben können folgende Lösungen in Betracht gezogen werden (neben Debezium): Lesen Sie auch in unserem Blog: Quelle: habr.comsinks Alle CDC-Ereignisse aus der überwachten Datenbank werden in das Thema mit dem Namen data.cdc.dbname. Andernfalls (ohne diese Einstellungen) würde Debezium standardmäßig für jedes Tabellenobjekt ein Topic erstellen: pg-dev.public..
Einschränkungen des Connectors
Anwendung der Konfiguration
curl -i -X POST -H "Accept:application/json"
-H "Content-Type:application/json" http://localhost:8083/connectors/
-d @pg-con.json$ curl -i http://localhost:8083/connectors/pg-connector/status
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)
{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}$ kafka/bin/kafka-console-consumer.sh
--bootstrap-server kafka:9092
--from-beginning
--property print.key=true
--topic data.cdc.dbname
postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', 'foo@bar.com');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":null,
"after":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"foo@bar.com"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600374991648,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":602,
"lsn":34088472,
"xmin":null
},
"op":"c",
"ts_ms":1600374991762,
"transaction":null
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"foo@bar.com"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"foo@bar.com"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600375609365,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":603,
"lsn":34089688,
"xmin":null
},
"op":"u",
"ts_ms":1600375609778,
"transaction":null
}
}
INSERT: Der Wert vor (before) ist gleich null, und nachher ist es die eingefügte Zeichenfolge. UPDATE: In payload.before wird der vorherige Zustand der Zeichenfolge angezeigt, und in payload.after – der neue mit dem Inhalt der Änderungen.2.2 MongoDB
{
"name": "mp-k8s-mongo-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max": "1",
"mongodb.hosts": "MainRepSet/mongo:27017",
"mongodb.name": "mongo",
"mongodb.user": "debezium",
"mongodb.password": "dbname",
"database.whitelist": "db_1,db_2",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
"transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
}
}sinks In diesem Fall wird Folgendes gemacht: Der Name des Zielthemas wird aus dem Schema umgewandelt. .. in data.cdc.mongo_.Ausfallsicherheit
Fazit
P.S.
