Tutvustame Debeziumi — CDC Apache Kafka jaoks

Oma töös kohtan sageli uusi tehnilisi lahendusi/softtooteid, mille kohta on venekeelses internetis üsna vähe teavet. Selle artikliga püüan täita üht sellist puudujääki, tuues näite oma hiljutisest praktikast, kus tuli seadistada CDC-sündmuste saatmine kahest populaarset andmebaasist (PostgreSQL ja MongoDB) Kafka klastrisse Debeziumi abil. Loodan, et see ülevaade, mis on kirjutatud tehtud töö tulemuste põhjal, osutub kasulikuks ka teistele.

Mis on Debezium ja üldiselt CDC?

Debezium on CDC (Capture Data Change) tarkvara kategooria esindaja, täpsemalt on see konnektorite kogum erinevatele andmebaasidele, mis on ühilduvad Apache Kafka Connect raamistikuga.

See on Avaallika projekt, millel on Apache License v2.0 litsents ja mille sponsoreerib Red Hat. Arendustööd on alustatud 2016. aastal ja praeguseks on selles ametlik tugi järgmistele andmebaasidele: MySQL, PostgreSQL, MongoDB, SQL Server. Samuti on olemas konnektorid Cassandra ja Oracle jaoks, kuid praegu on need "varajase juurdepääsu" staatuses ning uued väljaanded ei taga tagasipöörduvat ühilduvust.

CDC-d võrreldes traditsioonilise lähenemisega (kui rakendus loeb andmeid otse andmebaasist) on selle peamisteks eelisteks andmete muutuste streamimise realiseerimine ridade tasemel koos madala latentsuse, kõrge usaldusväärsuse ja kättesaadavusega. Viimane kaks punkti saavutatakse, kasutades CDC-sündmuste hoidmiseks Kafka klastrit.

Sarnaselt on eelisteks ka see, et sündmuste salvestamiseks kasutatakse ühtset mudelit, mistõttu lõpp-rakendusel ei pea olema mures erinevate andmebaaside kasutamise nüansside üle.

L finalmente, tänu sõnumite vahendaja kasutamisele avardub rakenduste horisontaalne skaleerimine, mis jälgivad andmete muutusi. Samuti on andmeallika mõju minimaalselt piiratud, kuna andmeid saadakse mitte otse andmebaasist, vaid Kafka klastrist.

Debeziumi arhitektuurist

Debeziumi kasutamine seisneb sellises lihtsas skeemis:

Andmebaas (kui andmeallikas) → konnektor Kafka Connectis → Apache Kafka → tarbija

Illustreerimiseks toome välja skeemi projekti veebilehelt:

Kuid see skeem mulle ei meeldi, kuna mul on tunne, et on võimalik ainult sink-konektori kasutamine.

Kuid tegelikult on olukord teistsugune: teie Data Lake'i täitmine (viimane element ülaltoodud skeemis) — see ei ole ainus viis Debeziumi rakendamiseks. Apache Kafkasse edastatud sündmusi saavad teie rakendused kasutada erinevate olukordade lahendamiseks. Näiteks:

• väheharjutatud andmete kustutamine vahemälust;
• teavituste saatmine;
• otsingumootorite indeksite värskendamine;
• mingi sarnane auditi logi;
• …

Kui teil on Java-rakendus ega ole vaja/võimalust kasutada Kafka klastrit, on samuti võimalik kasutada embedded-konektorit. Ilmselge pluss on see, et selle abil saab vältida lisainfrastruktuuri (nagu konektor ja Kafka) kasutamist. Kuid see lahendus on versioonist 1.1 deprekeeritud ja rohkem ei soovitata. Tulevastes väljaannetes võib selle toe eemaldada.

Selles artiklis käsitletakse soovitatavat arhitektuuri, mille on välja töötanud arendajad, et tagada tõrkeoleku kindlus ja skaleeritavus.

Konektori konfiguratsioon

Andmete, millel on kõige suurem väärtus, muudatuste jälgimiseks, on meil vajalik:

1. andmeallikas, milleks võivad olla MySQL versioonist 5.7 alates, PostgreSQL 9.6+, MongoDB 3.2+ (täielik nimekiri);
2. Apache Kafka klaster;
3. Kafka Connect instants (versioonid 1.x, 2.x);
4. konfigureeritud Debeziumi konektor.

Esimese kahe punkti, st andmebaasi ja Apache Kafkade installimise protsess, ületab artikli piirid. Kuid neile, kes soovivad kõike liivakastis käivitada, on ametlikus näidiste hoidlas valmis docker-compose.yaml.

Keskendume aga kahele viimasele punktile.

0. Kafka Connect

Siin ja edaspidi artikkel kõik konfiguratsiooni näited käsitletakse Debeziumi arendajate levitatud Docker-pildis. See sisaldab kõiki vajalikke pluginifailisid (konektorid) ning eeldab Kafka Connect'i seadistamist keskkonnamuutujate kaudu.

Kui kavatsete kasutada Confluenti Kafka Connect'i, peate ise vajalikud konektorite pluginad lisama katalooge, nagu on näidatud plugin.path või seadistada keskkonnamuutujaga CLASSPATH. Kafka Connecti ja konnektorite seaded määratakse konfiguratsioonifailide kaudu, mis edastatakse argumendina töötaja käivitamise käsule. Lisateavet vt dokumentatsioon.

Kogu Debeziumi seadistamise protsess konnektori variandiga toimub kahes etapis. Vaatame igaüht neist:

1. Kafka Connecti raamistikku seadistamine

Andmete voogedastamiseks Apache Kafka klastrisse seadistatakse Kafka Connecti raamistikus spetsiifilised parameetrid, nagu:

• ühenduse parameetrid klastri juurde,
• teemade nimed, kus konnektori seadistused endid hoitakse,
• grupi nimi, milles konnektor on käivitatud (jaotatud režiimi kasutamisel).

Projekti ametlik Docker-pilt toetab seadistamist keskkonna muutujate abil — seda me ka kasutame. Nii et, tõmmake pilt:

docker pull debezium/connect

Minimaalne muutujate komplekt, mis on vajalik konnektori käivitamiseks, näeb välja järgmine:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 — klastrisse Kafka ühendusserversite esialgne loend, et saada täielik loend klastriliikmetest;
• OFFSET_STORAGE_TOPIC=connector-offsets — teema, kus hoitakse positsioone, kus konnektor hetkel asub;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status — teema, kus hoitakse konnektori ja selle ülesannete staatust;
• CONFIG_STORAGE_TOPIC=connector-config — teema, kus hoitakse konnektori ja selle ülesannete seadistuste andmeid;
• GROUP_ID=1 — töötajate grupi ID, millel konnektori ülesanne võib töötada; vajalik jaotatud (distributed) režiimi kasutamisel.

Käivita konteiner nende muutujatega:

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

Märkus Avro kohta

Vaikimisi kirjutab Debezium andmed JSON-formaadis, mis sobib liivakastide ja väikeste andmemahtude jaoks, kuid võib muutuda probleemiks kõrgelt koormatud andmebaasides. Alternatiiv JSON-konverterile on sõnumite serialiseerimine Avro binaarformaati, mis võimaldab vähendada I/O alamsüsteemi koormust Apache Kafka-s.

Avro kasutamiseks on vajalik eraldi schema-registry (skeemide hoidmiseks). Muutujad konverteri jaoks näevad välja järgmised:

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 kasutamise ja registri seadistamise üksikasjad jäävad artikli raames väljapoole — edaspidi kasutame selguse huvides JSON-i.

2. Konnektori seadistamine

Nüüd saame liikuda otse konnektori konfiguratsiooni juurde, mis loeb andmeid allikast.

Vaatame kahe andmebaasi konnektori näitel: PostgreSQL ja MongoDB — mul on nende kohta kogemus ning nendes on erinevusi (kuigi väikseid, võivad need mõnel juhul olla olulised!).

Konfiguratsioon on kirja pandud JSON-notatsioonis ja laaditakse Kafka Connecti POST-päringu kaudu.

2.1. PostgreSQL

PostgreSQL konnektori konfiguratsiooni näide:

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

Konnektori tööpõhimõte pärast sellist seadistamist on üsna lihtne:

• Esimese käivitamise korral ühendub see konfiguratsioonis märgitud andmebaasi ja käivitub režiimis initial snapshot, saates Kafka-sse algse andmekogumi, mis on saadud tingimusliku SELECT * FROM table_name.
• Pärast initsialiseerimise lõppemist läheb konnektor üle muutuste lugemise režiimile PostgreSQL WAL-failidest.

Kasutatavatest valikutest:

• nimi — konnektori nimi, mille jaoks kasutatakse allpool kirjeldatud konfiguratsiooni; hiljem kasutatakse seda nime konnektoriga töötamiseks (st staatuse vaatamiseks/käivitamiseks/konfiguratsiooni värskendamiseks) läbi Kafka Connect REST API;
• connector.class — andmebaasi konnektori klass, mida konfigureeritav konnektor kasutab;
• plugin.name — pluginanimi, mis on vajalik andmete loogiliseks dekodeerimiseks WAL-failidest. Saadaval on valikud wal2json, decoderbuffs ja pgoutput. Esimesed kaks nõuavad vastavate laienduste installimist andmebaasi, kuid pgoutput PostgreSQL versioon 10 ja uuem ei vaja täiendavaid toiminguid;
• database.* — valikud andmebaasi ühendamiseks, kus database.server.name — PostgreSQL instantsi nimi, mida kasutatakse teema nime loomiseks Kafka klastris;
• table.include.list — nimekiri tabelitest, kus soovime muudatusi jälgida; määratakse formaadis schema.table_name; ei tohi kasutada koos table.exclude.list;
• heartbeat.interval.ms — intervall (millisekundites), mil konnektor saadab heartbeat-sõnumeid spetsiaalsesse teema;
• heartbeat.action.query — päring, mis täidetakse iga heartbeat-sõnumi saatmise ajal (valik ilmus versioonis 1.1);
• slot.name — replikatsioonislot'i nimi, mida konnektor kasutab;
• publication.name — nimi väljaandmise PostgreSQL-is, mida konnektor kasutab. Kui seda ei eksisteeri, proovib Debezium selle luua. Kui kasutajal, kelle alt ühendus toimub, ei ole selle toimingu jaoks piisavalt õigusi, lõpetab konnektor töö veaga;
• transforms määrab, kuidas täpselt muuta sihteema nime:
  • transforms.AddPrefix.type näitab, et kasutame regulaaravaldusi;
  • transforms.AddPrefix.regex — mask, mille alusel sihteema nimi määratakse uuesti;
  • transforms.AddPrefix.replacement — see, millele me selle ümber määrame.

Rohkem infot heartbeat'i ja transforms'i kohta

Vaikimisi saadab konnektor andmeid Kafka-sse igas komitendis tehingus ning selle LSN (Log Sequence Number) salvestatakse teenindusteemasse offset. Kuid mis juhtub, kui konnektor on seadistatud lugema mitte kogu andmebaasi, vaid ainult osa selle tabelitest (kus andmete värskendamine ei toimu sageli)?

• Konnektor loeb WAL-faile ja ei leia neist tehingute komiteid nendesse tabelitesse, millele ta tähelepanu pöörab.
• Seetõttu ei uuenda ta oma praegust positsiooni ei teemadesse ega replikatsioonislotis.
• See omakorda toob kaasa WAL-failide "hoidmise" kettal ja võimaliku kogu kettaruumi ammendumise.

Ja siin tulevad appi valikud heartbeat.interval.ms ja heartbeat.action.query. Nende valikute kooskasutamine võimaldab igal ajal heartbeat-sõnumi saatmisel teostada päringu andmete muutmiseks eraldi tabelis. Sellega ajakohastatakse pidevalt LSN-i, kus konnektor praegu asub (replikatsioonislotis). See võimaldab andmebaasi halduril kustutada WAL-failid, mis enam ei ole vajalikud. Lisainfot nende valikute töötamisest saab dokumentatsioon.

Teine võimalus, mis väärib suuremat tähelepanu, on transforms. Kuigi see on pigem mugavuse ja ilu kohta…

Debezium loob vaikimisi teemasid, järgides järgmisi nimetamisreegleid: serverName.schemaName.tableName. See ei pruugi alati mugav olla. Võimalustega transforms saab kasutada regulaaravaldisi, et määrata tabelite loend, mille sündmusi peab suunama kindla nimega teema.

Meie konfiguratsioonis toimub tänu transforms järgnev: kõik CDC-sündmused jälgitavast andmebaasist jõuavad teema nimega data.cdc.dbname. Vastupidisel juhul (ilma nende seadistusteta) oleks Debezium vaikimisi loonud iga tabeli jaoks teema, mille formaat on: pg-dev.public.

Ühenduse piirangud

PostgreSQL konnektori konfiguratsiooni kokkuvõtteks tasub rääkida järgmistest omadustest/piirangutest:

1. PostgreSQL konnektori funktsionaalsus põhineb loogilise dekodeerimise kontseptsioonil. Seega ei jälgi ta andmebaasi struktuuri muutmise päringuid (DDL) — seetõttu ei ole nende andmete teemades.
2. Kuna kasutatakse replikatsiooniplokke, on konnektori ühendamine võimalik seda peamise andmebaasi instantsiga.
3. Kui andmebaasi kasutajal, kelle all konnektor andmebaasi ühendub, on ainult lugemisõigused, tuleb enne esimest käivitamist käsitsi luua replikatsiooniplokk ja väljaanne andmebaasis.

Konfiguratsiooni rakendamine

Seega laadime meie konfiguratsiooni konnektorisse:

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

Kontrollime, et laadimine läks hästi ja konnektor käivitati:

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

Suurepärane: see on seadistatud ja tööks valmis. Nüüd teeme nägu, et oleme tarbija ja ühendume Kafka'ga, seejärel lisame ja muudame kirjet tabelis:

$ 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

Meie teemas peegelduvad need järgmise kujul:

Väga pikk JSON meie muudatustega

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

Mõlemal juhul koosnevad salvestused kirje (PK) võtmetest, mis on muudetud, ja muudatuste olemusest: milline oli kirje enne ja milline pärast.

• Juhtumi puhul INSERT: väärtus enne (before) on võrdne null, ja pärast on sisestatud string.
• Juhtumi puhul UPDATE: payload.before kuvatakse rea eelmine olek, ja payload.after on uus koos muudatuste olemusega.

2.2 MongoDB

See konnektor kasutab MongoDB standardset replikatsiooni mehhanismi, lugedes teavet primaarse andmebaasi oplog’ist.

Sarnaselt eelnevalt kirjeldatud PgSQL konnektorile, teeb ka see esmakordsel käivitamisel esialgse andmete snapshotsi, pärast mida lülitub konnektor oplog'i lugemise režiimile.

Konfiguratsiooni näidis:

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

Nagu märgata, ei ole siin uusi võimalusi võrreldes eelmise näitega, kuid on vähenenud vaid andmebaasi ühendamise ja nende eeltehtud nimetuste arv.

Seaded transforms seekord teevad nad järgmist: muudavad sihttopiku nime mustrist .. ja data.cdc.mongo_.

Talitluskatkestustunne

Vastus vastupidavuse ja kõrge k disponibiliteedi küsimusele on tänapäeval äärmiselt oluline — eriti kui räägime andmetest ja tehingutest, ning andmete muutuste jälgimine ei jää selles küsimuses kõrvale. Vaatleme, mis võib tegelikult valesti minna ja mis toimub Debeziumiga igas juhtumis.

On kolm tõrkevarianti:

1. Kafka Connecti tõrge. Kui Connect on seadistatud töötama hajutatud režiimis, peavad mitmed töötajad omama sama group.id. Sel juhul taastatakse konnektor teisel töötajal ja jätkab lugemist viimase kinnitatud asukohast Kafka topikus.
2. Ühenduse kadumine Kafka klastriga. Koonnektor lihtsalt peatab lugemise asukohas, mida ei õnnestunud saata Kafka'sse, ja üritab perioodiliselt uuesti saata, kuni katse õnnestub.
3. Andmeallika kättesaamatust. Ühendus püüab uuesti ühendada allika külge vastavalt seadistusele. Vaikimisi on see 16 katset, kasutades eksponentsiaalset tagasihoidlikkust. Pärast 16. ebaõnnestunud katset märgitakse ülesanne ebaõnnestunuks ja selle taastamiseks on vajalik käsitsi käivitamine läbi Kafka Connect REST-API.
   • Juhtumi puhul PostgreSQL Andmed ei kao, kuna replikatsiooni sildid ei luba WAL-faile eemaldada, mida ühendus ei ole lugenud. Sellega kaasneb aga ka varjukülg: kui ühendus ühenduse ja andmebaasi vahel katkeb pikemaks ajaks, on tõenäolisem, et ketasruum saab otsa, mis võib viia andmebaasi täieliku tõrkeni.
   • Juhtumi puhul MySQL Binaarsete logifailide võib andmebaas ise enne taastumise ühendust pöörata. See toob kaasa, et ühendus läheb ebaõnnestumiseks ja normaalse töö taastamiseks on vaja uuesti käivitada algse jäägi režiimis, et jätkata lugemist binaarsetest logidest.
   • Küsimus MongoDB. Dokumentatsioon ütleb: ühenduse käitumine juhul, kui logifailid/oplog on kustutatud ja ühendus ei saa lugemist jätkata seal, kus see jäi, on kõigi andmebaaside jaoks sama. See tähendab, et ühendus läheb olekusse ebaõnnestunuks ja nõuab taaskäivitamist režiimis initial snapshot.

     Kuid on ka erandeid. Kui ühendus on pikka aega olnud välja lülitatud (või ei ole saanud MongoDB instantsile ligi), ja oplog on selle aja jooksul pööratud, siis ühenduse taastumisel jätkab ühendus vaikselt andmete lugemist esimesest kättesaadavast kohast, mille tõttu osa andmeid jõuab Kafka. ei kaotsi.

Kokkuvõte

Debezium on minu esimene kogemus CDC-süsteemidega ja üldiselt väga positiivne. Projekt paelus mind peamiste andmebaaside toega, lihtsa seadistamise, klastritoe ja aktiivse kogukonnaga. Huvi korral soovitan tutvuda juhenditega Kafka Connect ja Debezium.

Debeziumi peamine eelis Kafka Connect JDBC-ühendaja ees on see, et muudatused loetakse välja andmebaasi päevikutest, mis võimaldab andmeid saada minimaalse viivitusega. JDBC-ühendaja (mis on osa Kafka Connectist) pärib jälgitavast tabelist kindla aja jooksul ja (sama põhjuse tõttu) ei genereeri teateid andmete kustutamisel (kuidas saab küsida andmeid, mida pole?).

Sarnaste probleemide lahendamiseks tasub vaadata järgmisi lahendusi (välja arvatud Debezium):

• JDBC Connector Kafka Connect;
• Mõned lahendused ainult MySQL jaoks:
  • MySQL Streamer;
  • Maxwell;
  • SpinalTap;
• Oracle GoldenGate, kuid see on hoopis teistsugune "kaalukategooria."

P.S.

Lugege ka meie blogist:

• «Määrame sobiva suuruse Kafka klastrile Kuberneteses»;
• «Praktilised lood meie SRE-argielust. Osa 2»;
• «Lühike ülevaade PostgreSQL operaatoritest Kuberneteses, meie valik ja kogemus».

Allikas: habr.com