ProHoster > Blog > Administrácia > Úvod do Debezium — CDC pre Apache Kafka

Úvod do Debezium — CDC pre Apache Kafka

Úvod do Debezium — CDC pre Apache Kafka

Vo svojej práci sa často stretávam s novými technickými riešeniami\/softvérovými produktmi, o ktorých je v slovenskom internete veľmi málo informácií. Touto článkom sa pokúsim zaplniť jednu takúto medzeru príkladom z mojej nedávnej praxe, keď bolo potrebné nastaviť odosielanie CDC udalostí z dvoch populárnych SDB (PostgreSQL a MongoDB) do klastera Kafka pomocou Debezium. Dúfam, že tento prehľadový článok, vzniknutý na základe vykonanej práce, bude užitočný aj ostatným.

Čo je Debezium a vôbec CDC?

Debezium — zástupca kategórie softvérového CDC (Capture Data Change), a presnejšie povedané, je to súbor konektorov pre rôzne SDB, kompatibilných s frameworkom Apache Kafka Connect.

Toto Projekt s otvoreným zdrojom, ktorý využíva licenciu Apache License v2.0 a je sponzorovaný spoločnosťou Red Hat. Vývoj prebieha od roku 2016 a v súčasnosti podporuje nasledujúce SDB: MySQL, PostgreSQL, MongoDB, SQL Server. Existujú aj konektory pre Cassandra a Oracle, ale momentálne sú v štádiu "skorého prístupu", pričom nové vydania nezaručujú spätnú kompatibilitu.

Ak porovnáme CDC s tradičným prístupom (keď aplikácia číta údaje priamo zo SDB), jeho hlavné výhody zahŕňajú realizáciu streamovania zmien údajov na úrovni riadkov s nízkou latenciou, vysokou spoľahlivosťou a dostupnosťou. Posledné dva body sa dosahujú vďaka použitiu klastera Kafka ako úložiska CDC udalostí.

Medzi výhody patrí aj fakt, že na uchovávanie udalostí sa používa jednotný model, a preto sa konečná aplikácia nemusí obávať nuáns prevádzky rôznych SDB.

Nakoniec, vďaka použitiu správy správ sa otvára priestor na horizontálne škálovanie aplikácií, ktoré sledovať zmeny v údajoch. Pri tom sa vplyv na zdroj údajov minimalizuje, keďže získavanie údajov prebieha nie priamo zo SDB, ale z klastera Kafka.

O architektúre Debezium

Používanie Debezium sa zjednodušuje na takúto jednoduchú schému:

SDB (ako zdroj údajov) → konektor v Kafka Connect → Apache Kafka → konzument

Ako ilustráciu uvádzam schému zo stránky projektu:

Úvod do Debezium — CDC pre Apache Kafka

Avšak táto schéma sa mi veľmi nepáči, pretože vytvára dojem, že je možné len použitie sink konektora.

V skutočnosti však situácia vyzerá inak: naplnenie vášho Data Lake (posledná časť na schéme vyššie) — to nie je jediný spôsob využitia Debezia. Udalosti odoslané do Apache Kafka môžu byť využívané vašimi aplikáciami na riešenie rôznych situácií. Napríklad:

  • odstránenie neaktuálnych dát z cache;
  • odosielanie upozornení;
  • aktualizácie vyhľadávacích indexov;
  • akýsi typ auditných logov;
  • …

V prípade, že máte aplikáciu na Jave a nie je potrebné/možné používať klaster Kafka, existuje aj možnosť práce cez embedded-konnectory. Zrejmou výhodou je, že sa môžete vzdať dodatočnej infraštruktúry (vo forme konektora a Kafky). Avšak toto riešenie je považované za zastarané (deprecated) od verzie 1.1 a už sa neodporúča používať (v budúcich vydaniach môže byť jeho podpora odstránená).

V tomto článku sa bude diskutovať o odporúčanej architektúre od vývojárov, ktorá zabezpečuje odolnosť voči chybám a možnosť škálovania.

Konfigurácia konektora

Aby sme mohli sledovať zmeny najdôležitejšej hodnoty — dát — budeme potrebovať:

  1. zdroj dát, ktorým môže byť MySQL od verzie 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (úplný zoznam);
  2. klaster Apache Kafka;
  3. inštancia Kafka Connect (verzie 1.x, 2.x);
  4. konfigurovaný konektor Debezium.

Práca na prvých dvoch bodoch, t.j. proces inštalácie DBMS a Apache Kafka, presahuje rámec článku. Avšak pre tých, ktorí chcú všetko rozbehnúť v pieskovisku, v oficiálnom repozitári s príkladmi je pripravený docker-compose.yaml.

Zastavíme sa podrobnejšie na dvoch posledných bodoch.

0. Kafka Connect

Tu a ďalej v článku sú všetky príklady konfigurácie zvažované v kontexte Docker obrazu, ktorý poskytujú vývojári Debezium. Obsahuje všetky potrebné súbory pluginov (konektorov) a predpokladá konfiguráciu Kafka Connect pomocou premenných prostredia.

V prípade, že sa predpokladá použitie Kafka Connect od Confluent, bude potrebné samostatne pridať pluginy potrebných konektorov do adresára, ktorý je určený v plugin.path alebo stanovený prostredníctvom premennej prostredia CLASSPATH. Nastavenia pracovníka Kafka Connect a konektorov sú definované prostredníctvom konfiguračných súborov, ktoré sú predávané ako argumenty príkazu na spustenie pracovníka. Viac informácií nájdete v dokumenácie.

Celý proces nastavenia Debezium s konektorom sa vykonáva v dvoch krokoch. Pozrime sa na každý z nich:

1. Nastavenie rámca Kafka Connect

Pre streamovanie údajov do klastru Apache Kafka v rámci Kafka Connect sa nastavujú špecifické parametre, ako sú:

  • parametre pripojenia ku klastru,
  • názvy tém, v ktorých bude uchovávaná samotná konfigurácia konektora,
  • meno skupiny, v ktorej je konektor spustený (v prípade použitia distribuovaného režimu).

Oficiálny Docker obraz projektu podporuje konfiguráciu pomocou premenných prostredia – a to využijeme. Takže, sťahujeme obraz:

docker pull debezium/connect

Minimálny súbor premenných prostredia, potrebný na spustenie konektora, vyzerá nasledovne:

  • BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 – počiatočný zoznam serverov klastru Kafka na získanie kompletného zoznamu členov klastru;
  • OFFSET_STORAGE_TOPIC=connector-offsets – téma pre uchovávanie pozícií, na ktorých sa v súčasnosti konektor nachádza;
  • CONNECT_STATUS_STORAGE_TOPIC=connector-status – téma pre uchovávanie stavu konektora a jeho úloh;
  • CONFIG_STORAGE_TOPIC=connector-config – téma pre uchovávanie údajov konfigurácie konektora a jeho úloh;
  • GROUP_ID=1 – identifikátor skupiny pracovníkov, na ktorých môže byť úloha konektora vykonávaná; nutnou podmienkou pri použití distribuovaného (distributed) režimu.

Spúšťame kontajner s týmito premennými:

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

Poznámka o Avro

Predvolene Debezium zapisuje údaje vo formáte JSON, čo je prijateľné pre pieskoviská a malé objemy údajov, ale môže sa to stať problémom vo vysoko zaťažených databázach. Alternatívou k JSON prevodníku je serializácia správ pomocou Avro do binárneho formátu, čo umožňuje znížiť záťaž na subsystém I/O v Apache Kafka.

Na použitie Avro je potrebné rozvinúť samostatný schema-registry (pre uchovávanie schém). Premenné pre prevodník budú vyzerať nasledovne:

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

Podrobnosti o použití Avro a nastavení registry presahujú rámec článku – na ďalšie ilustrácie budeme používať JSON.

2. Nastavenie samotného konektora

Teraz môžeme prejsť priamo na konfiguráciu samotného konektora, ktorý bude čítať údaje zo zdroja.

Pozrime sa na príklad konektorov pre dve DBMS: PostgreSQL a MongoDB, — pri ktorých mám skúsenosti a existujú medzi nimi rozdiely (nech sú aj malé, ale v niektorých prípadoch — podstatné!).

Konfigurácia je popísaná v notácii JSON a načítava sa do Kafka Connect pomocou POST požiadavky.

2.1. PostgreSQL

Príklad konfigurácie konektora pre 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"
  }
}

Princíp fungovania konektora po takejto konfigurácii je dosť jednoduchý:

  • Pri prvom spustení sa pripojí k databáze uvedenej v konfigurácii a spustí sa v režime initial snapshot, odosielajúc do Kafka počiatočný súbor údajov získaných pomocou podmienečného SELECT * FROM table_name.
  • Akonáhle je inicializácia dokončená, konektor prechádza do režimu čítania zmien z WAL súborov PostgreSQL.

O použitých možnostiach:

  • name — názov konektora, pre ktorý sa používa konfigurácia opísaná nižšie; nielenže sa toto meno používa na prácu s konektorom (t.j. kontrolu stavu/reštartovanie/aktualizáciu konfigurácie) prostredníctvom REST API Kafka Connect;
  • connector.class — trieda konektora DBMS, ktorá sa bude používať konfigurovaným konektorom;
  • plugin.name — názov pluginu na logické dekódovanie údajov z WAL súborov. Na výber sú dostupné wal2json, decoderbuffs a pgoutput. Prvé dve vyžadujú inštaláciu zodpovedajúcich rozšírení v DBMS, zatiaľ čo pgoutput pre PostgreSQL verzie 10 a vyššie nevyžaduje ďalšie manipulácie;
  • database.* — možnosti pripojenia k DB, kde database.server.name — názov inštancie PostgreSQL, použitý na formovanie názvu témy v klastri Kafka;
  • table.include.list — zoznam tabuliek, v ktorých chceme sledovať zmeny; zadané vo formáte schema.table_name; nemožno používať spolu s table.exclude.list;
  • heartbeat.interval.ms — interval (v milisekundách), s ktorým konektor posiela heartbeat správy do špeciálnej témy;
  • heartbeat.action.query — dopyt, ktorý sa vykoná pri odosielaní každej heartbeat správy (možnosť bola pridaná od verzie 1.1);
  • slot.name — názov replikovacieho slotu, ktorý bude používať konektor;
  • publication.name — názov publikácii v PostgreSQL, ktorý používa konektor. V prípade, že neexistuje, pokúsi sa jej vytvoriť. Ak nemá používateľ, pod ktorým sa pripája, dostatočné práva na tento úkon, konektor skončí s chybou;
  • transformácie určuje, ako presne sa má zmeniť názov cieľovej témy:
    • transforms.AddPrefix.type uvádza, že budeme používať regulárne výrazy;
    • transforms.AddPrefix.regex — maska, podľa ktorej sa prepisuje názov cieľovej témy;
    • transforms.AddPrefix.replacement — priamo to, na čo sa prepisujeme.

Viac informácií o heartbeat a transforms

Predvolene konektor odosiela údaje do Kafka pri každej potvrdenej transakcii a jej LSN (Log Sequence Number) zapíše do systémovej témy offset. Ale čo sa stane, ak je konektor nastavený na čítanie nie celej databázy, ale iba časti jej tabuliek (v ktorých sa aktualizácia údajov nedeje často)?

  • Konektor bude čítať WAL súbory a nebude v nich zisťovať potvrdenia transakcií v tých tabuľkách, ktoré sleduje.
  • Preto nebude aktualizovať svoju aktuálnu pozíciu ani v téme, ani v replikatornom slote.
  • To zas povedie k "uchovávaniu" WAL súborov na disku a pravdepodobnému vyčerpaniu celého diskového priestoru.

A tu prichádzajú na pomoc možnosti heartbeat.interval.ms a heartbeat.action.query. Použitie týchto možností v páre umožňuje zakaždým pri odosielaní heartbeat správy vykonávať dopyt na zmenu údajov v samostatnej tabuľke. Odtiaľ sa neustále aktualizuje LSN, na ktorom sa teraz nachádza konektor (v replikatornom slote). To umožňuje DBMS odstrániť WAL súbory, ktoré už nie sú potrebné. Ďalšie informácie o fungovaní možností nájdete v dokumenácie.

Ďalšia možnosť, zasluhujúca si väčšiu pozornosť, je transformácie. Aj keď ide skôr o pohodlie a estetiku...

Predvolene Debezium vytvára témy v súlade s nasledujúcou politikou pomenovávania: serverName.schemaName.tableName. To nemusí byť vždy pohodlné. Možnosti transformácie pomocou regulárnych výrazov môžete určiť zoznam tabuliek, ktorých udalosti sa majú smerovať do témy s konkrétnym názvom.

V našej konfigurácii vďaka transformácie sa deje nasledujúce: všetky CDC-udalosti z monitorovanej databázy sa dostanú do témy s názvom data.cdc.dbname. V opačnom prípade (bez týchto nastavení) by Debezium predvolene vytváral tému pre každú tabuľku vo forme: pg-dev.public.

.

Obmedzenia konektora

Na záver popisu konfigurácie konektora pre PostgreSQL je potrebné spomenúť nasledujúce vlastnosti/obmedzenia jeho fungovania:

  1. Funkcionalita konektora pre PostgreSQL sa zakladá na koncepcii logického dekódovania. Preto nesledoval požiadavky na zmenu štruktúry databázy (DDL) — v témach týchto dát teda nebudú.
  2. Keďže sa používajú replikážne sloty, pripojenie konektora je možné iba k vedúcemu exempláru DBMS.
  3. Ak má používateľ, pod ktorým sa konektor pripája k databáze, iba práva na čítanie, je potrebné pred prvým spustením ručne vytvoriť replikážny slot a publikáciu v databáze.

Aplikácia konfigurácie

Tak sme načítali našu konfiguráciu do konektora:

curl -i -X POST -H "Accept:application/json" 
  -H  "Content-Type:application/json"  http://localhost:8083/connectors/ 
  -d @pg-con.json

Kontrolujeme, či načítanie prebehlo úspešne a konektor sa spustil:

$ 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"}

Výborne: je nastavený a pripravený na použitie. Teraz sa predstavíme ako konzument a pripojíme sa k Kafka, po čom pridáme a zmeníme záznam v tabuľke:

$ 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

V našej téme to bude vyzerať nasledovne:

Veľmi dlhý JSON s našimi zmenami

{
  "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
  }
}

V oboch prípadoch záznamy pozostávajú z kľúča (PK) záznamu, ktorý bol zmenený, a samotnej podstaty zmien: aký bol záznam predtým a aký sa stal po.

  • V prípade INSERT: hodnotenie pred (pred) je rovné null, a po - reťazec, ktorý bol vložený.
  • V prípade UPDATE: do payload.before zobrazuje predchádzajúci stav riadka, a v payload.after — nový s podstatou zmien.

2.2 MongoDB

Tento konektor používa štandardný mechanizmus replikácie MongoDB, čítajúci informácie z oplog-u primárneho uzla DBMS.

Podobne ako už popísaný konektor pre PgSQL, aj tu pri prvom spustení skenuje primárny snímok údajov, po čom sa konektor prepne do režimu čítania oplog-u.

Príklad konfigurácie:

{
  "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"
        }
  }

Ako si môžete všimnúť, tu nie sú nové možnosti v porovnaní s minulým príkladom, ale znížil sa len počet možností, ktoré zodpovedajú za pripojenie k DB a ich prefixy.

Nastavenia transformácie tentoraz robia nasledujúce: premieňajú názov cieľového topiku zo schémy .. v data.cdc.mongo_.

Odolnosť voči zlyhaniam

Otázka odolnosti a vysokej dostupnosti je v súčasnosti dôležitejšia ako kedykoľvek predtým – najmä keď hovoríme o údajoch a transakciách, a sledovanie zmien údajov v tomto poradi nie je výnimkou. Poďme sa pozrieť, čo všetko sa môže pokaziť a čo sa bude diať s Debezium v každom prípade.

Existujú tri varianty zlyhania:

  1. Zlyhanie Kafka Connect. Ak je Connect nastavený na prácu v distribuovanom režime, je potrebné viacerým pracovníkom nastaviť rovnaké group.id. Potom pri zlyhaní jedného z nich bude konektor reštartovaný na inom pracovníkovi a pokračuje v čítaní z poslednej potvrdenej pozície v topiku v Kafka.
  2. Strata konektivity s Kafka klastrom. Konektor jednoducho zastaví čítanie na pozícii, ktorú sa nepodarilo odoslať do Kafka, a bude sa periodicky pokúšať znovu odoslať ju, kým sa pokus neskončí úspešne.
  3. Nedostupnosť zdroja údajov. Konektor sa pokúsi znova pripojiť k zdroju podľa konfigurácie. Predvolene je to 16 pokusov s použitím exponenciálneho backoffu. Po 16. neúspešnom pokuse bude úloha označená ako neúspešná a bude potrebné jej manuálne reštartovanie cez REST rozhranie Kafka Connect.
    • V prípade PostgreSQL údaje sa nestratia, pretože používanie replikovateľných slotov nedovolí odstrániť WAL súbory, ktoré neboli čítané konektorom. V tomto prípade je však aj druhá strana mince: ak bude dlhodobo prerušená sieťová konektivita medzi konektorom a DBMS, existuje pravdepodobnosť, že sa místa na disku zaplní, čo môže viesť k celkovému zlyhaniu DBMS.
    • V prípade MySQL súbory binlogov môžu byť otočené samotným DBMS skôr, než sa obnoví konektivita. To povedie k tomu, že konektor prejde do stavu neúspešná, a na obnovenie normálneho fungovania bude potrebné znovu spustiť v režime inicializácie snímky na pokračovanie čítania z binlogov.
    • O Kľúč-hodnota, grafový*. Dokumentácia uvádza: správanie konektora v prípadoch, keď sú súbory denníkov/oplogu odstránené a konektor nemôže pokračovať v čítaní z tej pozície, kde prestal, je rovnaké pre všetky DBMS. Ide o to, že konektor prejde do stavu neúspešná a bude vyžadovať opätovné spustenie v režime initial snapshot.

      Existujú však výnimky. Ak konektor bol dlhodobo v odpojenom stave (alebo nemohol dosiahnuť inštanciu MongoDB), a oplog za tento čas prešiel rotáciou, potom pri obnovení pripojenia konektor bez vzrušenia začne čítať dáta z prvej dostupnej pozície, čo môže spôsobiť, že niektoré dáta v Kafka nie sa dostanú.

Záver

Debezium — moja prvá skúsenosť s CDC systémami a celkovo veľmi pozitívna. Projekt zaujal podporou hlavných DBMS, jednoduchosťou konfigurácie, podporou klastrovania a aktívnou komunitou. Odporúčam tým, ktorí majú záujem o prax, aby si prečítali príručky pre Kafka Connect a Debezium.

V porovnaní s JDBC konektorom pre Kafka Connect je hlavnou výhodou Debezium to, že zmeny sa čítajú z denníkov DBMS, čo umožňuje získať dáta s minimálnym oneskorením. JDBC Connector (z dodávky Kafka Connect) robí dopyty na sledovanej tabuľke s pevnými intervalmi a (z rovnakého dôvodu) nevytvára správy pri odstraňovaní dát (ako by ste mohli dopytovať dáta, ktoré neexistujú?).

Na riešenie podobných úloh sa môžete zamerať na nasledujúce riešenia (okrem Debezium):

P.S.

Prečítajte si tiež na našom blogu:

Zdroj: habr.com

Kúpte si spoľahlivý hosting pre webové stránky s ochranou proti DDoS, VPS VDS servery 🔥 Kúpte si spoľahlivý hosting pre webové stránky s ochranou proti DDoS, VPS VDS servery | ProHoster