Представяме ви Debezium - CDC за Apache Kafka

В работата си често се натъквам на нови технически решения / софтуерни продукти, информацията за които е доста оскъдна в рускоезичния интернет. С тази статия ще се опитам да запълня една такава празнина с пример от моята скорошна практика, когато трябваше да настроя изпращане на CDC събития от две популярни СУБД (PostgreSQL и MongoDB) към клъстер на Kafka, използвайки Debezium. Надявам се, че тази обзорна статия, която се появи в резултат на свършената работа, ще бъде полезна за други.

Какво е Debezium и CDC като цяло?

Дебезиум - Представител на категорията CDC софтуер (Заснемане на промяна на данните), или по-точно, това е набор от конектори за различни СУБД, които са съвместими с рамката Apache Kafka Connect.

То проект с отворен код, лицензиран под Apache License v2.0 и спонсориран от Red Hat. Разработката е в ход от 2016 г. и към момента предоставя официална поддръжка за следните СУБД: MySQL, PostgreSQL, MongoDB, SQL Server. Има също конектори за Cassandra и Oracle, но те в момента са в статус "ранен достъп" и новите версии не гарантират обратна съвместимост.

Ако сравним CDC с традиционния подход (когато приложението чете данни от СУБД директно), тогава основните му предимства включват внедряването на поточно предаване на промяна на данни на ниво ред с ниска латентност, висока надеждност и наличност. Последните две точки се постигат чрез използване на клъстер Kafka като хранилище за CDC събития.

Също така, предимствата включват факта, че един модел се използва за съхраняване на събития, така че крайното приложение не трябва да се тревожи за нюансите на работа с различни СУБД.

И накрая, използването на брокер на съобщения отваря поле за хоризонтално мащабиране на приложения, които проследяват промените в данните. В същото време въздействието върху източника на данни е сведено до минимум, тъй като данните се получават не директно от СУБД, а от клъстера Kafka.

Относно архитектурата Debezium

Използването на Debezium се свежда до тази проста схема:

СУБД (като източник на данни) → конектор в Kafka Connect → Apache Kafka → потребител

Като илюстрация ще дам диаграма от сайта на проекта:

Тази схема обаче не ми харесва много, защото изглежда, че е възможен само конектор за мивка.

В действителност ситуацията е различна: запълване на вашето Data Lake (последната връзка в диаграмата по-горе) не е единственият начин да използвате Debezium. Събитията, изпратени до Apache Kafka, могат да се използват от вашите приложения за справяне с различни ситуации. Например:

• премахване на неподходящи данни от кеша;
• изпращане на известия;
• актуализации на индекса за търсене;
• някакъв вид одитни регистрационни файлове;
• ...

В случай, че имате Java приложение и няма нужда/възможност да използвате Kafka клъстер, има и възможност за работа през вграден конектор. Очевидният плюс е, че с него можете да откажете допълнителна инфраструктура (под формата на конектор и Kafka). Това решение обаче е отхвърлено от версия 1.1 и вече не се препоръчва за употреба (може да бъде премахнато в бъдещи версии).

Тази статия ще обсъди препоръчаната от разработчиците архитектура, която осигурява толерантност към грешки и мащабируемост.

Конфигурация на конектора

За да започнем да проследяваме промените в най-важната стойност - данните - имаме нужда от:

1. източник на данни, който може да бъде MySQL от версия 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (пълен списък);
2. Клъстер Apache Kafka
3. Екземпляр на Kafka Connect (версии 1.x, 2.x);
4. конфигуриран Debezium конектор.

Работете по първите две точки, т.е. процесът на инсталиране на СУБД и Apache Kafka са извън обхвата на статията. Въпреки това, за тези, които искат да разположат всичко в пясъчник, има готов такъв в официалното хранилище с примери docker-compose.yaml.

Ще се спрем по-подробно на последните две точки.

0. Kafka Connect

Тук и по-нататък в статията всички примери за конфигурация се разглеждат в контекста на изображението на Docker, разпространено от разработчиците на Debezium. Той съдържа всички необходими плъгин файлове (конектори) и предоставя конфигурация на Kafka Connect с помощта на променливи на средата.

Ако възнамерявате да използвате Kafka Connect от Confluent, ще трябва сами да добавите плъгините на необходимите конектори към директорията, посочена в plugin.path или зададен чрез променлива на средата CLASSPATH. Настройките за Kafka Connect worker и конектори се дефинират чрез конфигурационни файлове, които се предават като аргументи на командата worker start. За подробности вж документация.

Целият процес на настройка на Debeizum във версията на конектора се извършва на два етапа. Нека разгледаме всеки от тях:

1. Настройване на рамката Kafka Connect

За поточно предаване на данни към Apache Kafka клъстер, специфични параметри се задават в рамката на Kafka Connect, като например:

• настройки за свързване на клъстера,
• имена на теми, в които ще се съхранява конфигурацията на самия конектор,
• името на групата, в която работи конекторът (в случай на използване на разпределен режим).

Официалният Docker образ на проекта поддържа конфигурация с помощта на променливи на средата - това е, което ще използваме. Така че нека изтеглим изображението:

docker pull debezium/connect

Минималният набор от променливи на средата, необходими за изпълнение на конектора, е както следва:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 - първоначален списък на Kafka клъстерни сървъри, за да получите пълен списък на членовете на клъстера;
• OFFSET_STORAGE_TOPIC=connector-offsets — тема за съхраняване на позициите, където в момента се намира конекторът;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status - тема за съхраняване на състоянието на конектора и неговите задачи;
• CONFIG_STORAGE_TOPIC=connector-config - тема за съхраняване на данни за конфигурацията на конектора и неговите задачи;
• GROUP_ID=1 — идентификатор на групата работници, върху които може да се изпълни задачата за свързване; изисква се при използване на разпределени (разпространен) режим.

Стартираме контейнера с тези променливи:

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.2

Бележка за Avro

По подразбиране Debezium записва данни във формат JSON, който е приемлив за пясъчни кутии и малки количества данни, но може да бъде проблем при силно натоварени бази данни. Алтернатива на конвертора на JSON е сериализирането на съобщения с помощта на Avro в двоичен формат, което намалява натоварването на I/O подсистемата в Apache Kafka.

За да използвате Avro, трябва да разположите отделен схема-регистър (за съхраняване на схеми). Променливите за конвертора ще изглеждат така:

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.AvroConverter

Подробности за използването на Avro и настройката на регистър за него са извън обхвата на статията - освен това, за яснота, ще използваме JSON.

2. Настройка на самия конектор

Сега можете да преминете директно към конфигурацията на самия конектор, който ще чете данни от източника.

Нека разгледаме примера на конектори за две СУБД: PostgreSQL и MongoDB, за които имам опит и за които има разлики (макар и малки, но в някои случаи значителни!).

Конфигурацията е описана в JSON нотация и е качена в Kafka Connect с помощта на POST заявка.

2.1. PostgreSQL

Примерна конфигурация на конектор за 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"
  }
}

Принципът на работа на конектора след тази конфигурация е доста прост:

• При първото стартиране се свързва с базата данни, зададена в конфигурацията, и стартира в режим първоначална моментна снимка, изпращайки на Kafka първоначалния набор от данни, получени с условното SELECT * FROM table_name.
• След като инициализацията приключи, конекторът влиза в режим на четене на промените от PostgreSQL WAL файлове.

За използваните опции:

• name — името на съединителя, за който се използва описаната по-долу конфигурация; в бъдеще това име се използва за работа с конектора (т.е. преглед на състоянието / рестартиране / актуализиране на конфигурацията) чрез Kafka Connect REST API;
• connector.class — класът на СУБД конектор, който ще се използва от конфигурирания конектор;
• plugin.name е името на плъгина за логическо декодиране на данни от WAL файлове. Предлага се за избор wal2json, decoderbuffs и pgoutput. Първите две изискват инсталирането на съответните разширения в СУБД и pgoutput за PostgreSQL версия 10 и по-нова не изисква допълнителни манипулации;
• database.* — опции за свързване към базата данни, където database.server.name - името на екземпляра на PostgreSQL, използван за формиране на името на темата в клъстера Kafka;
• table.include.list - списък с таблици, в които искаме да проследяваме промените; дадени във формата schema.table_name; не може да се използва заедно с table.exclude.list;
• heartbeat.interval.ms — интервал (в милисекунди), с който конекторът изпраща сърдечни съобщения до специална тема;
• heartbeat.action.query - заявка, която ще се изпълнява при изпращане на всяко сърдечно съобщение (опцията се появи от версия 1.1);
• slot.name — името на слота за репликация, който ще се използва от конектора;
• publication.name - Име Публикуване в PostgreSQL, който конекторът използва. В случай, че не съществува, Debezium ще се опита да го създаде. Ако потребителят, под който е направена връзката, няма достатъчно права за това действие, конекторът ще излезе с грешка;
• transforms определя как точно да промените името на целевата тема:
  • transforms.AddPrefix.type показва, че ще използваме регулярни изрази;
  • transforms.AddPrefix.regex — маска, чрез която се предефинира името на целевата тема;
  • transforms.AddPrefix.replacement - директно това, което предефинираме.

Повече за сърдечния ритъм и трансформациите

По подразбиране конекторът изпраща данни на Kafka за всяка ангажирана транзакция и записва нейния LSN (пореден номер на журнал) в темата на услугата offset. Но какво се случва, ако конекторът е конфигуриран да чете не цялата база данни, а само част от нейните таблици (в които данните се актуализират рядко)?

• Конекторът ще чете WAL файлове и няма да открива транзакции в тях към таблиците, които наблюдава.
• Следователно няма да актуализира текущата си позиция нито в темата, нито в слота за репликация.
• Това, от своя страна, ще доведе до "залепване" на WAL файловете на диска и вероятно дисковото им пространство ще свърши.

И тук на помощ идват опциите. heartbeat.interval.ms и heartbeat.action.query. Използването на тези опции по двойки прави възможно изпълнението на заявка за промяна на данни в отделна таблица всеки път, когато се изпраща сърдечно съобщение. Така LSN, на който в момента се намира конекторът (в слота за репликация), се актуализира постоянно. Това позволява на СУБД да премахва WAL файлове, които вече не са необходими. За повече информация как работят опциите, вижте документация.

Друг вариант, който заслужава по-голямо внимание е transforms. Въпреки че става дума повече за удобство и красота...

По подразбиране Debezium създава теми, като използва следната политика за именуване: serverName.schemaName.tableName. Това не винаги може да е удобно. Настроики transforms като използвате регулярни изрази, можете да дефинирате списък от таблици, чиито събития трябва да бъдат насочени към тема с конкретно име.

В нашата конфигурация благодарение на transforms се случва следното: всички CDC събития от проследяваната база данни ще отидат в темата с името data.cdc.dbname. В противен случай (без тези настройки), Debezium по подразбиране ще създаде тема за всяка таблица на формуляра: pg-dev.public.<table_name>.

Ограничения на конектора

В края на описанието на конфигурацията на конектора за PostgreSQL си струва да говорим за следните характеристики / ограничения на неговата работа:

1. Функционалността на конектора за PostgreSQL разчита на концепцията за логическо декодиране. Следователно той не проследява заявки за промяна на структурата на базата данни (DDL) - съответно тези данни няма да ги има в темите.
2. Тъй като се използват слотове за репликация, свързването на конектора е възможно само към главния екземпляр на СУБД.
3. Ако потребителят, под който конекторът се свързва към базата данни, има права само за четене, тогава преди първото стартиране ще трябва ръчно да създадете слот за репликация и да публикувате в базата данни.

Прилагане на конфигурация

Така че нека заредим нашата конфигурация в конектора:

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, след което добавяме и променяме запис в таблицата:

$ 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', '[email protected]');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

В нашата тема това ще се покаже, както следва:

Много дълъг JSON с нашите промени

{
"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":"[email protected]"
},
"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":"[email protected]"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"[email protected]"
},
"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
}
}

И в двата случая записите се състоят от ключа (PK) на записа, който е променен, и самата същност на промените: какъв е бил записът преди и какво е станал след това.

• В случай на INSERT: стойност преди (before) равно на nullпоследвано от вмъкнатия низ.
• В случай на UPDATE: v payload.before показва се предишното състояние на реда и в payload.after - нов със същността на промяната.

2.2 MongoDB

Този конектор използва стандартния механизъм за репликация на MongoDB, като чете информация от oplog на основния възел на СУБД.

Подобно на вече описания конектор за PgSQL, и тук при първото стартиране се прави моментна снимка на първичните данни, след което конекторът преминава в режим на четене на oplog.

Пример за конфигурация:

{
"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 този път правят следното: обръщат името на целевата тема от схемата <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

отказоустойчивост

Въпросът за толерантността към грешки и високата достъпност в наше време е по-остър от всякога - особено когато говорим за данни и транзакции, а проследяването на промените в данните не е настрана по този въпрос. Нека да разгледаме какво може да се обърка по принцип и какво ще се случи с Debezium във всеки отделен случай.

Има три опции за отказ:

1. Kafka Connect грешка. Ако Connect е конфигуриран да работи в разпределен режим, това изисква няколко работници да зададат един и същ group.id. След това, ако един от тях се провали, конекторът ще бъде рестартиран на другия работник и ще продължи да чете от последната ангажирана позиция в темата в Kafka.
2. Загуба на връзка с клъстер Kafka. Конекторът просто ще спре да чете на позицията, която не е успял да изпрати на Kafka и периодично ще се опитва да го изпрати отново, докато опитът успее.
3. Източникът на данни е недостъпен. Конекторът ще се опита да се свърже отново към източника според конфигурацията. Използването по подразбиране е 16 опита експоненциално отстъпление. След 16-ия неуспешен опит задачата ще бъде маркирана като Неуспешно и ще трябва да се рестартира ръчно чрез интерфейса Kafka Connect REST.
   • В случай на PostgreSQL данните няма да бъдат загубени, т.к използването на слотове за репликация ще предотврати изтриването на WAL файлове, които не са прочетени от конектора. В този случай има недостатък: ако мрежовата свързаност между конектора и СУБД е прекъсната за дълго време, има вероятност дисковото пространство да свърши и това може да доведе до отказ на цялата СУБД.
   • В случай на MySQL binlog файловете могат да бъдат ротирани от самата СУБД, преди връзката да бъде възстановена. Това ще накара конектора да премине в неуспешно състояние и ще трябва да се рестартира в първоначален режим на моментна снимка, за да продължи да чете от binlogs, за да възстанови нормалната работа.
   • Около MongoDB. В документацията се казва: поведението на конектора в случай, че log/oplog файловете са изтрити и конекторът не може да продължи да чете от позицията, където е спрял, е еднакво за всички СУБД. Това се крие във факта, че конекторът ще премине в състояние Неуспешно и ще изисква рестартиране в режим първоначална моментна снимка.

     Има обаче и изключения. Ако конекторът е бил в прекъснато състояние за дълго време (или не е могъл да достигне екземпляра на MongoDB) и oplog е бил завъртян през това време, тогава, когато връзката бъде възстановена, конекторът ще продължи спокойно да чете данни от първата налична позиция , поради което някои от данните в Кафка не ще удари.

Заключение

Debezium е първият ми опит с CDC системи и като цяло беше много положителен. Проектът подкупи поддръжката на основната СУБД, лекота на конфигуриране, поддръжка за клъстериране и активна общност. За тези, които се интересуват от практиката, препоръчвам да прочетете ръководствата за Kafka Connect и Дебезиум.

В сравнение с JDBC конектора за Kafka Connect, основното предимство на Debezium е, че промените се четат от регистрационните файлове на СУБД, което позволява данните да се получават с минимално забавяне. JDBC Connector (предоставен от Kafka Connect) прави запитвания към проследяваната таблица на фиксиран интервал и (поради същата причина) не генерира съобщения, когато данните се изтриват (как можете да правите заявки за данни, които не са там?).

За да разрешите подобни проблеми, можете да обърнете внимание на следните решения (в допълнение към Debezium):

• JDBC Connector Kafka Connect
• Няколко решения само за MySQL:
  • MySQL Streamer;
  • Максуел;
  • Спинална пункция;
• Oracle Golden Gate, но това е съвсем различна "теглова категория".

PS

Прочетете също в нашия блог:

• «Определете подходящия размер за клъстер Kafka в Kubernetes»;
• «Практически истории от нашето SRE ежедневие. Част 2»;
• «Кратък преглед на изразите на PostgreSQL за Kubernetes, нашият избор и опит".

Източник: www.habr.com