In meiner Arbeit stoße ich häufig auf neue technische Lösungen/Softwareprodukte, über die im russischsprachigen Internet ziemlich wenig Informationen verfügbar sind. Mit diesem Artikel möchte ich eine solche Lücke schließen, indem ich ein Beispiel aus meiner jüngsten Praxis gebe, in dem es nötig war, die Übertragung von CDC-Ereignissen aus zwei beliebten Datenbanksystemen (PostgreSQL und MongoDB) in ein Kafka-Cluster mithilfe von Debezium einzurichten. Ich hoffe, dass dieser Überblicksartikel, der als Ergebnis der geleisteten Arbeit entstand, auch für andere nützlich ist.
Was ist Debezium und generell CDC?
ist ein Vertreter der Kategorie der CDC-Software (), genauer gesagt handelt es sich um eine Sammlung von Konnektoren für verschiedene Datenbanksysteme, die mit dem Rahmenwerk Apache Kafka Connect kompatibel sind.
Das ist das unter der Lizenz Apache License v2.0 steht und von der Firma Red Hat gesponsert wird. Die Entwicklung begann im Jahr 2016, und derzeit wird die offizielle Unterstützung für folgende Datenbanksysteme angeboten: MySQL, PostgreSQL, MongoDB, SQL Server. Es gibt auch Konnektoren für Cassandra und Oracle, die sich jedoch derzeit im Status „Früher Zugang“ befinden, und neue Releases garantieren keine Rückwärtskompatibilität.
Im Vergleich zu CDC, dem traditionellen Ansatz (bei dem die Anwendung Daten direkt aus der Datenbank liest), zählen die Hauptvorteile die Implementierung von Datenänderungs-Streaming auf Zeilenebene mit niedriger Latenz, hoher Zuverlässigkeit und Verfügbarkeit. Diese letzten beiden Punkte werden durch die Verwendung eines Kafka-Clusters als Speicher für CDC-Ereignisse erreicht.
Ein weiterer Vorteil ist, dass zur Speicherung von Ereignissen ein einheitliches Modell verwendet wird, sodass die Endanwendung sich nicht um die Feinheiten der Nutzung verschiedener Datenbanksysteme kümmern muss.
Schließlich ermöglicht die Verwendung eines Message Brokers horizontal skalierbare Anwendungen, die Änderungen in den Daten verfolgen. Der Einfluss auf die Datenquelle wird auf ein Minimum reduziert, da die Datenerfassung nicht direkt aus der Datenbank, sondern aus dem Kafka-Cluster erfolgt.
Zur Architektur von Debezium
Die Nutzung von Debezium lässt sich durch ein einfaches Schema darstellen:
Datenbank (als Datenquelle) → Konnektor in Kafka Connect → Apache Kafka → Consumer
Zur Illustration füge ich ein Schema von der Projektwebsite bei:
Allerdings gefällt mir dieses Schema nicht besonders, da der Eindruck entsteht, dass nur die Verwendung eines Sink-Konnektors möglich ist.
In Wirklichkeit sieht die Situation anders aus: die Befüllung Ihres Data Lake (letzte Glieder in der obigen Grafik) — ist nicht die einzige Möglichkeit, Debezium zu verwenden. Ereignisse, die an Apache Kafka gesendet werden, können von Ihren Anwendungen zur Lösung verschiedener Situationen genutzt werden. Zum Beispiel:
- Löschen von nicht mehr benötigten Daten aus dem Cache;
- Versand von Benachrichtigungen;
- Aktualisierungen von Suchindizes;
- eine Art von Audit-Logs;
- …
Falls Sie eine Java-Anwendung haben und keine Notwendigkeit/Möglichkeit besteht, ein Kafka-Cluster zu verwenden, gibt es auch die Möglichkeit, über zu arbeiten. Ein offensichtlicher Vorteil besteht darin, dass Sie auf zusätzliche Infrastruktur (in Form von Connector und Kafka) verzichten können. Dieses Lösung wurde jedoch mit Version 1.1 als veraltet (deprecated) erklärt und wird nicht mehr empfohlen (in zukünftigen Versionen könnte die Unterstützung eingestellt werden).
In diesem Artikel wird die von den Entwicklern empfohlene Architektur betrachtet, die Fehlertoleranz und Skalierbarkeit bietet.
Connector-Konfiguration
Um die wichtigste Ressource – die Daten – zu überwachen, benötigen wir:
- eine Datenquelle, welche MySQL ab Version 5.7, PostgreSQL 9.6+, MongoDB 3.2+ sein kann ();
- ein Apache Kafka-Cluster;
- eine Kafka Connect-Instanz (Versionen 1.x, 2.x);
- einen konfigurierten Debezium-Connector.
Die Arbeiten zu den ersten beiden Punkten, d.h. der Installationsprozess von DBMS und Apache Kafka, überschreiten den Rahmen dieses Artikels. Für diejenigen, die alles in einer Sandbox bereitstellen möchten, gibt es ein fertiges .
Wir werden uns jedoch ausführlicher mit den beiden letzten Punkten befassen.
0. Kafka Connect
Hier und im Folgenden werden alle Konfigurationsbeispiele im Kontext des von den Debezium-Entwicklern bereitgestellten Docker-Images betrachtet. Es enthält alle erforderlichen Plugin-Dateien (Connectoren) und sieht die Konfiguration von Kafka Connect über Umgebungsvariablen vor.
Falls die Verwendung von Kafka Connect von Confluent geplant ist, müssen die erforderlichen Connector-Plugins selbst in das Verzeichnis, das in der plugin.path angegeben wird, oder über die Umgebungsvariable CLASSPATHdefiniert werden. Die Einstellungen für den Kafka Connect-Worker und die Connectoren werden über Konfigurationsdateien festgelegt, die als Argumente an den Startbefehl des Workers übergeben werden. Weitere Einzelheiten finden Sie in .
Der gesamte Prozess der Einrichtung von Debeizium mit einem Connector erfolgt in zwei Phasen. Lassen Sie uns jede von ihnen betrachten:
1. Konfiguration des Kafka Connect Frameworks
Für das Streaming von Daten in ein Apache Kafka Cluster werden im Kafka Connect Framework spezifische Parameter festgelegt, wie beispielsweise:
- Verbindungsparameter zum Cluster,
- die Namen der Themen, in denen die Konfiguration des Connectors selbst gespeichert wird,
- der Name der Gruppe, in der der Connector ausgeführt wird (im Fall der Verwendung des verteilten Modus).
Das offizielle Docker-Image des Projekts unterstützt die Konfiguration über Umgebungsvariablen – genau das werden wir nutzen. Also laden wir das Image herunter:
docker pull debezium/connectDie minimalen Umgebungsvariablen, die für den Start des Connectors erforderlich sind, sehen wie folgt aus:
-
BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092– die Anfangsliste der Kafka Clusterserver zum Abrufen der vollständigen Liste der Clustermitglieder; -
OFFSET_STORAGE_TOPIC=connector-offsets– das Thema zur Speicherung der Positionen, an denen sich der Connector derzeit befindet; -
CONNECT_STATUS_STORAGE_TOPIC=connector-status– das Thema zur Speicherung des Status des Connectors und seiner Aufgaben; -
CONFIG_STORAGE_TOPIC=connector-config– das Thema zur Speicherung der Konfigurationsdaten des Connectors und seiner Aufgaben; -
GROUP_ID=1– die Gruppen-ID der Worker, auf denen die Aufgabe des Connectors ausgeführt werden kann; erforderlich im verteilten (distributed) Modus.
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 Debeizium Daten im JSON-Format, was für Sandboxen und kleine Datenmengen akzeptabel ist, aber in stark belasteten Datenbanken problematisch werden kann. Eine Alternative zum JSON-Converter ist die Serialisierung von Nachrichten durch in ein binäres Format, was die Belastung des I/O-Subsystems in Apache Kafka verringert.
Für die Verwendung von Avro muss ein separater (zur Speicherung von Schemata) bereitgestellt werden. Die Variablen für den Converter 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.AvroConverterDetails zur Verwendung von Avro und zur Konfiguration des Registry gehen über den Rahmen dieses Artikels hinaus – im Folgenden verwenden wir zu Illustrationszwecken JSON.
2. Einstellung des Connectors
Jetzt können wir direkt zur Konfiguration des Connectors übergehen, der die Daten aus der Quelle lesen wird.
Betrachten wir am Beispiel von Connectors für zwei Datenbanken: PostgreSQL und MongoDB, — für die ich Erfahrung habe und bei denen es Unterschiede gibt (auch wenn diese gering sind, sind sie in bestimmten Fällen erheblich!).
Die Konfiguration wird in JSON-Notation beschrieben und über einen POST-Request in Kafka Connect geladen.
2.1. PostgreSQL
Beispielkonfiguration 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"
}
}Das Prinzip der Funktionsweise des Connectors nach dieser Einstellung ist ziemlich einfach:
- Beim ersten Start verbindet er sich mit der in der Konfiguration angegebenen Datenbank und wird im Modus initial snapshot, der einen anfänglichen Datensatz, der mithilfe von
SELECT * FROM table_name. - erhalten wurde, nach Kafka sendet.
Nach Abschluss der Initialisierung wechselt der Connector in den Modus zum Lesen von Änderungen aus den WAL-Dateien von PostgreSQL.
-
nameZu den verwendeten Optionen: -
— der Name des Connectors, für den die unten beschriebene Konfiguration verwendet wird; dieser Name wird später zur Arbeit mit dem Connector verwendet (d.h. zum Überprüfen des Status / Neustart / Aktualisieren der Konfiguration) über die REST-API von Kafka Connect;connector.class -
— Klasse des DB-Connectors, die vom konfigurierbaren Connector verwendet wird;plugin.name— Name des Plugins zur logischen Dekodierung von Daten aus WAL-Dateien. Zur Auswahl stehen,wal2jsonunddecoderbuffspgoutputdecoderbuffs. Die ersten beiden erfordern die Installation der entsprechenden Erweiterungen in der DB, während -
für PostgreSQL Version 10 und höher keine zusätzlichen Manipulationen erforderlich sind;database.*— Optionen zur Verbindung mit der DB, wobeidatabase.server.name -
— der Name der PostgreSQL-Instanz ist, der zur Bildung des Topic-Namens im Kafka-Cluster verwendet wird;table.include.list— Liste der Tabellen, in denen wir Änderungen verfolgen möchten; wird im Formatschema.table_name; kann nicht zusammen mit; -
table.exclude.list— Intervall (in Millisekunden), mit dem der Connector Heartbeat-Nachrichten an ein spezielles Topic sendet; -
heartbeat.action.query— Abfrage, die bei jedem Versand einer Heartbeat-Nachricht ausgeführt wird (Option ist seit Version 1.1 verfügbar); -
slot.name— Name des Replikationsslots, der vom Connector verwendet wird; publication.name— Name in PostgreSQL, den der Connector verwendet. Falls dieser nicht vorhanden ist, versucht Debezium, ihn zu erstellen. Falls der Benutzer, unter dem die Verbindung hergestellt wird, nicht über ausreichende Rechte für diese Aktion verfügt, beendet der Connector die Arbeit mit einem Fehler;-
transformsbestimmt, wie der Name des Ziel-Topics geändert werden soll:-
transforms.AddPrefix.typegibt an, dass reguläre Ausdrücke verwendet werden; -
transforms.AddPrefix.regex— Maske, nach der der Name des Ziel-Topics überschrieben wird; -
transforms.AddPrefix.replacement— direkt das, auf was wir überschreiben.
-
Weitere Informationen zu Heartbeat und Transforms
Standardmäßig sendet der Connector Daten in Kafka bei jeder vermerkten Transaktion und protokolliert deren LSN (Log Sequence Number) in ein internes Topic offset. Aber was passiert, wenn der Connector so konfiguriert ist, dass er nicht die gesamte Datenbank, sondern nur einen Teil ihrer Tabellen liest (in denen nicht oft Datenaktualisierungen stattfinden)?
- Der Connector wird die WAL-Dateien lesen und keine Transaktionscommits in die Tabellen erkennen, die er überwacht.
- Daher wird er seine aktuelle Position weder im Topic noch im Replikationsslot aktualisieren.
- Dies wird wiederum dazu führen, dass die WAL-Dateien auf der Festplatte "festgehalten" werden und der gesamte Speicherplatz wahrscheinlich erschöpft wird.
Hier kommen Optionen ins Spiel table.exclude.list und heartbeat.action.query. Die Verwendung dieser Optionen in Kombination ermöglicht es, bei jedem Versand einer Heartbeat-Nachricht eine Anfrage zur Datenänderung in einer separaten Tabelle auszuführen. Dadurch wird LSN, auf dem sich der Connector derzeit befindet (im Replikationsslot), ständig aktualisiert. Dies erlaubt der DB, WAL-Dateien zu löschen, die nicht mehr benötigt werden. Weitere Informationen zur Funktionsweise der Optionen finden Sie in .
Eine andere Option, die besondere Aufmerksamkeit verdient, ist transforms. Auch wenn es eher um den Komfort und die Ästhetik geht...
Standardmäßig erstellt Debezium Topics anhand der folgenden Namenskonvention: serverName.schemaName.tableName. Dies ist nicht immer praktisch. Mit Optionen transforms Mit Hilfe von regulären Ausdrücken kann eine Liste von Tabellen bestimmt werden, deren Ereignisse in ein Thema mit einem bestimmten Namen geroutet werden sollen.
In unserer Konfiguration erfolgt dies durch Am Ende der Beschreibung der Konfiguration des Connectors für PostgreSQL sollte man auf folgende Besonderheiten/Einschränkungen seiner Arbeit hinweisen: Lassen Sie uns also unsere Konfiguration in den Connector laden: Überprüfen wir, ob das Laden erfolgreich war und der Connector gestartet wurde: Ausgezeichnet: Er ist konfiguriert und bereit zur Verwendung. Jetzt spielen wir den Verbraucher und verbinden uns mit Kafka, danach fügen wir einen Datensatz in die Tabelle ein und ändern ihn: In unserem Thema wird dies wie folgt angezeigt: Ein sehr langer JSON mit unseren Änderungen In beiden Fällen bestehen die Aufzeichnungen aus dem Schlüssel (PK) des geänderten Datensatzes und dem eigentlichen Inhalt der Änderungen: wie der Datensatz zuvor war und wie er danach aussieht. Dieser Connector verwendet den Standard-MongoDB-Replikationsmechanismus, indem er Informationen aus dem Oplog des primären Datenbankknotens liest. Ähnlich wie beim bereits beschriebenen PgSQL-Connector wird auch hier beim ersten Start ein primärer Datenbanksnapshot erstellt, wonach der Connector in den Oplog-Lesemodus wechselt. Beispielkonfiguration: Wie man sehen kann, gibt es hier keine neuen Optionen im Vergleich zum vorherigen Beispiel, sondern lediglich eine Verringerung der Anzahl der Optionen, die für die Verbindung zur Datenbank und ihre Präfixe verantwortlich sind. Einstellungen Die Frage der Fehlertoleranz und Hochverfügbarkeit steht heutzutage so dringlich wie nie — besonders wenn wir über Daten und Transaktionen sprechen, und die Nachverfolgung von Datenänderungen steht in diesem Zusammenhang nicht im Hintergrund. Lassen Sie uns betrachten, was grundsätzlich schiefgehen kann und was mit Debezium in jedem dieser Fälle passieren wird. Es gibt drei Ausfallvarianten: gelangen. nicht Debezium ist meine erste Erfahrung mit CDC-Systemen und insgesamt recht positiv. Das Projekt überzeugt durch Unterstützung für die gängigen Datenbanken, einfache Konfiguration, Unterstützung für Clustering und eine aktive Community. Interessierten empfehle ich, sich mit den Leitfäden für Kafka Connect und . Im Vergleich zum JDBC-Connector für Kafka Connect ist der Hauptvorteil von Debezium, dass Änderungen aus den Protokollen der Datenbankmanagementsysteme gelesen werden, was eine Datenübertragung mit minimaler Latenz ermöglicht. Der JDBC-Connector (aus der Lieferung von Kafka Connect) führt Abfragen an der überwachten Tabelle in festen Intervallen durch und generiert aus demselben Grund keine Nachrichten bei der Löschung von Daten (wie kann man Daten anfordern, die nicht mehr vorhanden sind?). Für die Lösung ähnlicher Aufgaben können die folgenden Lösungen (neben Debezium) in Betracht gezogen werden: Lesen Sie auch in unserem Blog: Quelle: habr.comtransforms : Alle CDC-Ereignisse aus der überwachten DB gelangen in das Thema mit dem Namen data.cdc.dbname. Andernfalls (ohne diese Einstellungen) würde Debezium standardmäßig ein Thema für jede Tabelle des Typs 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) entspricht null, und danach ist es die Zeichenfolge, die eingefügt wurde. UPDATE: in payload.before zeigt den vorherigen Zustand der Zeile an, während in payload.after — der neue Zustand 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"
}
}transforms macht man diesmal Folgendes: Wandelt den Namen des Zielthemas aus dem Schema .. in data.cdc.mongo_.Fehlertoleranz
Fazit
P.S.
