
U svom radu često se susrećem s novim tehničkim rješenjima/programskim proizvodima o kojima se na hrvatskom internetu može naći jako malo informacija. Ovim člankom pokušavam popuniti jedan takav prostor primjerom iz svoje nedavne prakse, kada je bilo potrebno postaviti slanje CDC događaja iz dva popularna DBMS-a (PostgreSQL i MongoDB) u Kafka klaster uz pomoć Debeziuma. Nadam se da će ovaj pregledni članak, koji je nastao na temelju obavljenog posla, biti koristan i drugima.
Što je Debezium i uopće CDC?
je predstavnik kategorije softvera CDC (), a preciznije, to je skup konektora za različite DBMS-ove, koji su kompatibilni s Apache Kafka Connect okvirom.
To koji koristi licencu Apache License v2.0 i sponzorira ga tvrtka Red Hat. Razvoj se odvija od 2016. godine, a trenutno se službeno podržavaju sljedeći DBMS-ovi: MySQL, PostgreSQL, MongoDB, SQL Server. Također postoje konektori za Cassandru i Oracle, ali trenutno su oni u statusu 'ranog pristupa', a nova izdanja ne jamče povratnu kompatibilnost.
Ako usporedimo CDC s tradicionalnim pristupom (kada aplikacija izravno čita podatke iz DBMS-a), njegova glavna prednost je implementacija strujanja promjene podataka na razini redaka s niskim kašnjenjem, visokom pouzdanošću i dostupnošću. Posljednja dva čimbenika postižu se zahvaljujući upotrebi Kafka klastera kao spremišta za CDC događaje.
Osim toga, za prednosti se može svrstati činjenica da se za pohranu događaja koristi jedinstven model, tako da krajnjoj aplikaciji ne mora brinuti o nijansama korištenja različitih DBMS-ova.
Na kraju, zahvaljujući upotrebi poslužitelja poruka otvara se mogućnost horizontalnog skaliranja aplikacija koje prate promjene u podacima. Pri tome je utjecaj na izvor podataka sveden na minimum, jer se podaci ne dobivaju izravno iz DBMS-a, već iz Kafka klastera.
O arhitekturi Debeziuma
Upotreba Debeziuma svodi se na jednostavnu shemu:
DBMS (kao izvor podataka) → konektor u Kafka Connect → Apache Kafka → konsument
Kao ilustraciju prikazat ću shemu s web stranice projekta:

Međutim, ova shema mi se ne sviđa previše, jer se stječe dojam da je moguće koristiti samo sink konektor.
U stvarnosti, situacija je drugačija: popunjavanje vašeg Data Lake-a (posljednji link na gornjoj shemi) — to nije jedini način korištenja Debeziuma. Događaji poslani u Apache Kafka mogu se koristiti u vašim aplikacijama za rješavanje različitih situacija. Na primjer:
- uklanjanje neaktualnih podataka iz cache-a;
- slanje obavijesti;
- ažuriranje pretraživačkih indeksa;
- neka vrsta audit logova;
- …
U slučaju da imate aplikaciju na Javi i nema potrebe/mogućnosti za korištenje Kafka klastera, također postoji mogućnost rada preko . Očita prednost je u tome što se može odustati od dodatne infrastrukture (poput konektora i Kafke). Ipak, ovo rješenje je proglašeno zastarjelim (deprecated) s verzijom 1.1 i više se ne preporučuje za upotrebu (u budućim izdanjima podrška za njega može biti uklonjena).
U ovom članku razmotrit će se preporučena arhitektura koju su razvili programeri, koja osigurava otpornost na kvarove i mogućnost skaliranja.
Konfiguracija konektora
Da bismo počeli pratiti promjene najvažnijih vrijednosti — podataka — potrebni su nam:
- izvor podataka, koji može biti MySQL od verzije 5.7, PostgreSQL 9.6+, MongoDB 3.2+ ();
- klaster Apache Kafka;
- instanca Kafka Connect (verzije 1.x, 2.x);
- konfigurirani konektor Debezium.
Radovi na prva dva stavka, tj. proces instalacije DBMS-a i Apache Kafke, izlaze iz okvira ovog članka. Međutim, za one koji žele pokrenuti sve u sandboxu, u službenom repozitoriju s primjerima postoji spreman .
Mi ćemo se detaljnije osvrnuti na dva posljednja stavka.
0. Kafka Connect
Ovdje i dalje u članku svi primjeri konfiguracije razmatraju se u kontekstu Docker slike koja se distribuira od strane Debezium programera. Ona sadrži sve potrebne plugin datoteke (konektore) i predviđa konfiguraciju Kafka Connect pomoću varijabli okoline.
U slučaju da se planira korištenje Kafka Connect-a od Confluent-a, bit će potrebno samostalno dodati plugine potrebnih konektora u direktoriju navedenu u plugin.path ili definiranu putem varijable okoline CLASSPATH. Postavke radnika Kafka Connect i konektora definiraju se putem konfiguracijskih datoteka koje se prosljeđuju kao argumenti pri pokretanju radnika. Više informacija možete pogledati u .
Cijeli proces postavljanja Debeziuma s optimizatorom odvija se u dva koraka. Razmotrimo svaki od njih:
1. Postavljanje okvira Kafka Connect
Za streamanje podataka u Apache Kafka klasteru, u okviru Kafka Connect postavljaju se specifični parametri, kao što su:
- parametri povezivanja s klasterom,
- imena tema u kojima će se pohranjivati sama konfiguracija optimizatora,
- ime grupe u kojoj je optimizator pokrenut (u slučaju korištenja distribuiranog režima).
Službeni Docker image projekta podržava konfiguraciju pomoću varijabli okruženja — to ćemo i iskoristiti. Dakle, preuzimamo image:
docker pull debezium/connectMinimalni skup varijabli okruženja potrebnih za pokretanje optimizatora izgleda ovako:
-
BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092— početni popis servera klastera Kafka za dobivanje cjelovitog popisa članova klastera; -
OFFSET_STORAGE_TOPIC=connector-offsets— tema za pohranu pozicija na kojima se trenutno nalazi optimizator; -
CONNECT_STATUS_STORAGE_TOPIC=connector-status— tema za pohranu statusa optimizatora i njegovih zadaća; -
CONFIG_STORAGE_TOPIC=connector-config— tema za pohranu podataka konfiguracije optimizatora i njegovih zadaća; -
GROUP_ID=1— identifikator grupe radnika na kojima se može izvršiti zadaća optimizatora; potreban pri korištenju distribuiranog (distributed) režima.
Pokrećemo kontejner s tim varijablama:
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.2Napomena o Avro
Po defaultu, Debezium zapisuje podatke u JSON formatu, što je prihvatljivo za testne okoline i male količine podataka, ali može postati problem u visoko opterećenim bazama. Alternativa JSON konverteru je serijalizacija poruka pomoću u binarni format, što smanjuje opterećenje na I/O podsustav u Apache Kafki.
Za korištenje Avra potrebno je pokrenuti odvojeni (za pohranu shema). Varijable za konverter će izgledati ovako:
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.AvroConverterDetalji o korištenju Avra i postavljanju registry izlaze iz okvira ovog članka — dalje ćemo, za ilustraciju, koristiti JSON.
2. Konfiguracija konektora
Sada možemo preći na konfiguraciju samog konektora, koji će čitati podatke iz izvora.
Razmotrit ćemo na primjeru konektora za dvije baze podataka: PostgreSQL i MongoDB, — za koje imam iskustva i postoje razlike (iako male, u nekim slučajevima — bitne!).
Konfiguracija se opisuje u JSON notaciji i učitava u Kafka Connect putem POST zahtjeva.
2.1. PostgreSQL
Primjer konfiguracije konektora za 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"
}
}Princip rada konektora nakon takve konfiguracije je prilično jednostavan:
- Pri prvom pokretanju povezuje se na bazu navedenu u konfiguraciji i pokreće se u modu initial snapshot, šaljući u Kafka početni skup podataka, dobivenih pomoću
SELECT * FROM table_name. - Nakon što je inicijalizacija završena, konektor prelazi u način čitanja promjena iz WAL datoteka PostgreSQL-a.
O korištenim opcijama:
-
ime— ime konektora za koji se koristi konfiguracija opisano u nastavku; ovo ime se kasnije koristi za rad s konektorom (tj. pregled statusa/ponovno pokretanje/ ažuriranje konfiguracije) putem REST API-ja Kafka Connect; -
connector.class— klasa konektora baze podataka koja će se koristiti konfiguriranim konektorom; -
plugin.name— naziv dodatka za logičko dekodiranje podataka iz WAL datoteka. Na raspolaganju suwal2json,decoderbuffsipgoutput. Prva dva zahtijevaju instalaciju odgovarajućih ekstenzija u bazi podataka, dokpgoutputza PostgreSQL verziju 10 i više ne zahtijeva dodatne manipulacije; -
database.*— opcije za povezivanje s bazom podataka, gdjedatabase.server.name— ime instance PostgreSQL-a koje se koristi za formiranje imena topika u Kafka klasteru; -
table.include.list— lista tablica u kojima želimo pratiti promjene; postavlja se u formatuschema.table_name; ne može se koristiti zajedno stable.exclude.list; -
heartbeat.interval.ms— interval (u milisekundama) s kojim konektor šalje heartbeat poruke na posebnu temu; -
heartbeat.action.query— upit koji će se izvršavati pri slanju svake heartbeat poruke (opcijski je dostupan od verzije 1.1); -
slot.name— ime replikacijskog slota koji će koristiti konektor; publication.name— ime u PostgreSQL-u, koju koristi konektor. U slučaju da ne postoji, Debezium će pokušati da je kreira. Ako korisnik, pod kojim se ostvaruje konekcija, nema dovoljno prava za ovu akciju, konektor će završiti rad s greškom;-
transformacijeodređuje kako tačno menjati naziv ciljne teme:-
transforms.AddPrefix.typeoznačava da ćemo koristiti regularne izraze; -
transforms.AddPrefix.regex— maska kojom se preimenuje naziv ciljne teme; -
transforms.AddPrefix.replacement— konkretno ono na što preimenujemo.
-
Više o heartbeat-u i transforms
Podrazumevano, konektor šalje podatke u Kafka-u pri svakoj potvrđenoj transakciji, a njen LSN (Log Sequence Number) zapisuje u službenu temu offset. Ali šta će se dogoditi ako je konektor podešen da čita ne celu bazu, već samo deo njenih tabela (u kojima se ažuriranje podataka ne dešava često)?
- Konektor će čitati WAL fajlove i neće prepoznati u njima potvrdu transakcija u onim tabelama koje prati.
- Zato neće ažurirati svoju trenutnu poziciju ni u temi, ni u replikacijskom slotu.
- To će, zauzvrat, dovesti do "zadržavanja" WAL fajlova na disku i verovatnog iscrpljivanja celog diskovnog prostora.
I ovde na pomoć dolaze opcije heartbeat.interval.ms i heartbeat.action.query. Korišćenje ovih opcija u paru omogućava da se pri svakom slanju heartbeat poruke izvrši upit za promenu podataka u odvojenoj tabeli. Na taj način se konstantno ažurira LSN na kojem se trenutno nalazi konektor (u replikacijskom slotu). To omogućava bazi podataka da izbriše WAL fajlove koji više nisu potrebni. Više o opcijama možete saznati u .
Druga opcija, koja zaslužuje više pažnje, je transformacije. Iako se više odnosi na udobnost i estetiku…
Podrazumevano, Debezium kreira teme, oslanjajući se na sledeću politiku imenovanja: serverName.schemaName.tableName. To ne mora uvek biti pogodno. Opcijama transformacije možete uz pomoć regularnih izraza odrediti popis tablica čiji se eventi trebaju usmjeriti u temu s određenim imenom.
U našoj konfiguraciji zahvaljujući Na kraju opisa konfiguracije konektora za PostgreSQL treba spomenuti sljedeće karakteristike/ograničenja u njegovom radu: Dakle, učitajmo našu konfiguraciju u konektor: Provjeravamo da je učitavanje prošlo uspješno i da se konektor pokrenuo: Odlično: postavljen je i spreman za rad. Sada ćemo se pretvarati da smo konzumator i povezati se s Kafka, nakon čega ćemo dodati i promijeniti zapis u tablici: U našoj temi to će se odraziti na sljedeći način: Vrlo dugi JSON s našim promjenama U oba slučaja zapisi se sastoje od ključa (PK) zapisa koji je promenjen i direktno od suštine promena: kako je zapis izgledao pre i kako izgleda nakon. Ovaj konektor koristi standardni mehanizam replikacije MongoDB, čitajući informacije iz oplog-a primarnog čvora DBMS-a. Slično prethodno opisano konektoru za PgSQL, i ovde se pri prvom pokretanju uzima primarni snapshot podataka, nakon čega se konektor prebacuje na režim čitanja oplog-a. Primjer konfiguracije: Kao što se može primetiti, ovde nema novih opcija u poređenju sa prethodnim primerom, ali je smanjen broj opcija koje se odnose na povezivanje s DB i njihove prefikse. Postavke Pitanje otpornosti na greške i visoke dostupnosti je danas važnije nego ikad — posebno kada govorimo o podacima i transakcijama, i praćenju promena podataka neizbežno je povezano sa tim pitanjem. Razmotrićemo šta može poći po zlu i šta će se desiti sa Debezium-om u svakom od slučajeva. Postoje tri opcije greške: Međutim, postoje iznimke. Ako je konnektor dugo vremena bio isključen (ili nije mogao doseći instancu MongoDB), a oplog se u tom vremenu rotirao, prilikom obnavljanja veze konnektor će smireno nastaviti čitati podatke s prve dostupne pozicije, zbog čega će dio podataka u Kafka ne biti izgubljen. Debezium — moje prvo iskustvo s CDC sustavima i općenito vrlo pozitivno. Projekt očarava podrškom za glavne DBMS-ove, jednostavnošću konfiguracije, podrškom za klasterizaciju i aktivnom zajednicom. Onima koji su zainteresirani za praksu preporučujem da se upoznaju s vodičima za i . U usporedbi s JDBC konnektorom za Kafka Connect, glavna prednost Debeziuma je ta što se promjene čitaju iz dnevnika DBMS-a, što omogućava dobivanje podataka s minimalnom latencijom. JDBC konnektor (iz isporuke Kafka Connect) šalje upite na praćenu tablicu u fiksnim intervalima i (iz istog razloga) ne generira poruke prilikom brisanja podataka (kako možete zatražiti podatke kojih nema?). Za rješavanje sličnih zadataka možete obratiti pozornost na sljedeća rješenja (osim Debeziuma): Pročitajte i na našem blogu: Izvor: habr.comtransformacije događa se sljedeće: svi CDC-eventi iz praćene Baze podataka bit će utemeljeni na temu s imenom data.cdc.dbname. U suprotnom (bez ovih postavki) Debezium bi prema zadanim postavkama kreirao po jednu temu za svaku tablicu u obliku: pg-dev.public..
Ograničenja konektora
Primjena konfiguracije
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
}
}
UBACI: vrednost pre (before) je jednaka null, a nakon — string koji je umetnut. AZURIRAJ: u payload.before prikazuje prethodno stanje reda, a u payload.after — novo sa suštinom promena.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"
}
}transformacije ovog puta čine sledeće: pretvaraju ime ciljnog topika iz šeme .. u data.cdc.mongo_.Otpornost na greške
Zaključak
P.S.
