Fa'ailoa Debezium - CDC mo Apache Kafka

I laʻu galuega, e masani ona ou maua ni fofo faʻatekinisi fou / oloa faʻapipiʻi, faʻamatalaga e tau le maua i luga ole Initaneti ole gagana Rusia. Faatasi ai ma lenei tusiga, o le a ou taumafai e faʻatumu se tasi o ia avanoa i se faʻataʻitaʻiga mai laʻu faʻataʻitaʻiga talu ai nei, pe a ou manaʻomia le faʻatulagaina o le auina atu o mea CDC mai DBMS taʻutaʻua e lua (PostgreSQL ma MongoDB) i le Kafka cluster e faʻaaoga ai Debezium. Ou te faʻamoemoe o lenei iloiloga tusiga, lea na aliali mai o se taunuuga o le galuega na faia, o le a aoga i isi.

O le a le Debesium ma le CDC lautele?

Debesium - Sui ole vaega ole polokalame CDC (Pu'e suiga o fa'amaumauga), pe sili atu le saʻo, o se seti o fesoʻotaʻiga mo DBMS eseese e fetaui ma le Apache Kafka Connect framework.

lea poloketi tatala, laiseneina i lalo ole Apache License v2.0 ma lagolagoina e Red Hat. O loʻo faʻaauau le atinaʻe talu mai le 2016 ma i le taimi nei o loʻo tuʻuina atu ai le lagolago aloaia mo DBMS nei: MySQL, PostgreSQL, MongoDB, SQL Server. O loʻo iai foʻi fesoʻotaʻiga mo Cassandra ma Oracle, ae o loʻo i ai nei i le "vave avanoa", ma o faʻasalalauga fou e le faʻamaonia ai le fetaui i tua.

Afai tatou te faʻatusatusaina le CDC ma le auala masani (pe a faitau saʻo e le tusi talosaga faʻamatalaga mai le DBMS), o lona tulaga sili ona lelei e aofia ai le faʻatinoina o suiga o faʻamatalaga e tafe i le laina laina ma le maualalo o le latency, maualuga le faʻatuatuaina ma le avanoa. O mea mulimuli e lua e ausia e ala i le faʻaaogaina o le Kafka cluster e fai ma faʻamaumauga mo mea CDC.

E le gata i lea, o mea lelei e aofia ai le mea moni o se faʻataʻitaʻiga e tasi e faʻaaogaina e teu ai mea tutupu, o lea o le talosaga mulimuli e le tatau ona popole e uiga i nuances o le faʻaogaina o DBMS eseese.

Ma le mea mulimuli, o le faʻaaogaina o se fefaʻatauaʻiga feʻau e matala ai le avanoa mo le faʻalauteleina o faʻasalalauga o talosaga e siaki ai suiga i faʻamaumauga. I le taimi lava e tasi, o le aʻafiaga i luga o faʻamatalaga faʻamatalaga e faʻaitiitia, talu ai o faʻamaumauga e le maua saʻo mai le DBMS, ae mai le kulupu Kafka.

E uiga i le fausaga o le Debesium

O le faʻaaogaina o Debezium e sau i lalo i lenei faiga faigofie:

DBMS (e pei o faʻamaumauga) → fesoʻotaʻiga i Kafka Connect → Apache Kafka → tagata faʻatau

Mo se faʻataʻitaʻiga, o le a ou tuʻuina atu se ata mai le upega tafaʻilagi o le poloketi:

Ae ui i lea, ou te le fiafia tele i lenei polokalame, aua e foliga mai e naʻo se soʻo faʻapipiʻi e mafai.

O le mea moni, e ese le tulaga: faʻatumu lau Vaituloto o Data (so'oga mulimuli i le ata i luga) e le na o le pau lea o le auala e faʻaaoga ai Debezium. O mea na tutupu na lafoina i Apache Kafka e mafai ona faʻaogaina e au talosaga e faʻafetaui ai tulaga eseese. Faataitaiga:

• aveeseina o faʻamatalaga le taua mai le faʻaoga;
• auina atu o fa'amatalaga;
• su'esu'e fa'ailoga fa'afouga;
• nisi ituaiga o ogalaau suetusi;
• ...

I le tulaga o loʻo i ai sau talosaga Java ma e leai se manaʻoga / avanoa e faʻaaoga ai se fuifui Kafka, o loʻo i ai foi le avanoa e galue ai. so'oga fa'apipi'i. O le faʻaopoopoga manino e faʻapea e mafai ona e teena isi atinaʻe (i le tulaga o se fesoʻotaʻiga ma Kafka). Ae ui i lea, o lenei fofo ua le toe faʻaaogaina talu mai le version 1.1 ma ua le toe fautuaina mo le faʻaaogaina (e mafai ona aveese i faʻasalalauga i le lumanaʻi).

O lenei tusiga o le a talanoaina le fausaga o loʻo fautuaina e le au atinaʻe, lea e maua ai le faʻapalepale masei ma le faʻalauteleina.

Feso'ota'iga feso'ota'iga

Ina ia amata siaki suiga i le taua sili ona taua - faʻamaumauga - matou te manaʻomia:

1. punaoa faʻamatalaga, lea e mafai ona avea ma MySQL amata mai le version 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (lisi atoa);
2. Apache Kafka fuifui
3. Fa'ata'ita'iga a Kafka Connect (versions 1.x, 2.x);
4. configured fesoʻotaʻiga Debesium.

Galue i vaega muamua e lua, i.e. o le faagasologa o le faʻapipiʻiina o se DBMS ma Apache Kafka e sili atu i le lautele o le tusiga. Ae ui i lea, mo i latou e manaʻo e faʻapipiʻi mea uma i totonu o le sandbox, o loʻo i ai se mea ua saunia i totonu o le fale teu oloa faʻatasi ma faʻataʻitaʻiga. docker-compose.yaml.

O le a tatou taulaʻi atili i manatu mulimuli e lua.

0. Kafka Fesootai

O iinei ma mulimuli ane i le tusiga, o faʻataʻitaʻiga faʻataʻitaʻiga uma o loʻo mafaufauina i totonu o le faʻataʻitaʻiga o le ata Docker tufatufaina e le au atinaʻe Debezium. O loʻo iai uma faila faʻapipiʻi manaʻomia (fesoʻotaʻiga) ma tuʻuina atu le faʻaogaina o le Kafka Connect e faʻaaoga ai fesuiaiga o le siosiomaga.

Afai e te manaʻo e faʻaoga Kafka Connect mai Confluent, e te manaʻomia le faʻaopoopoina o faʻapipiʻi o fesoʻotaʻiga talafeagai oe lava ia i le lisi o loʻo faʻamaonia i totonu. plugin.path pe seti e ala i se fesuiaiga o le siosiomaga CLASSPATH. O faʻatulagaga mo le Kafka Connect tagata faigaluega ma fesoʻotaʻiga e faʻamalamalamaina e ala i faila faʻatulagaina e pasia e fai ma finauga i le tagata faigaluega amata poloaiga. Mo fa'amatalaga tagai fa'amaumauga.

O le faagasologa atoa o le faʻatulagaina o Debeizum i le fesoʻotaʻiga o loʻo faia i ni vaega se lua. Seʻi o tatou iloiloina i latou taʻitasi:

1. Faʻatulagaina le faʻavae o le Kafka Connect

Ina ia faʻasalalau faʻamatalaga i se Apache Kafka fuifui, faʻamaufaʻailoga faʻapitoa e seti i le Kafka Connect framework, e pei o:

• fa'apipi'i feso'ota'iga,
• igoa o autu o le a teuina ai le faatulagaga o le fesoʻotaʻiga lava ia,
• le igoa o le kulupu o loʻo faʻaogaina ai le fesoʻotaʻiga (i le tulaga o le faʻaaogaina o le tuʻufaʻatasia).

O le ata Docker aloaia o le poloketi e lagolagoina le faʻaogaina e faʻaaoga ai suiga o le siosiomaga - o le mea lea o le a matou faʻaaogaina. Se'i o tatou sii maia le ata:

docker pull debezium/connect

Ole seti pito maualalo ole si'osi'omaga e mana'omia e fa'agaioi ai le feso'ota'iga e fa'apea:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 - lisi muamua o 'au'aunaga fuifui a Kafka e maua ai se lisi atoa o sui auai;
• OFFSET_STORAGE_TOPIC=connector-offsets - se autu mo le teuina o tulaga o loʻo i ai nei le soʻo;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status - se autu mo le teuina o le tulaga o le fesoʻotaʻiga ma ana galuega;
• CONFIG_STORAGE_TOPIC=connector-config - se autu mo le teuina o faʻamaumauga o fesoʻotaʻiga ma ana galuega;
• GROUP_ID=1 - fa'ailoaina o le vaega o tagata faigaluega e mafai ona fa'atino ai le galuega feso'ota'i; mana'omia pe a fa'aaoga tufatufa (tufatufa) pulega.

Matou te amataina le koneteina i nei fesuiaiga:

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

Manatua e uiga ia Avro

Ona o le le mafai, e tusia e Debezium faʻamaumauga i le JSON format, lea e talia mo sandboxes ma le tele o faʻamaumauga, ae mafai ona avea ma faʻafitauli i faʻamaumauga mamafa. O se isi mea i le JSON converter o le fa'asologa o fe'au fa'aoga Avro i se faatulagaga binary, lea e faʻaitiitia ai le uta i luga o le I / O subsystem i Apache Kafka.

Ina ia faʻaaoga Avro, e tatau ona e faʻapipiʻi se eseʻese schema-resitala (mo le teuina o fuafuaga). O fesuiaiga mo le tagata liliu mai o le a pei o lenei:

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

O faʻamatalaga i le faʻaaogaina o Avro ma le faʻatulagaina o se resitala mo ia e sili atu i le lautele o le tusiga - sili atu, mo le manino, o le a matou faʻaogaina le JSON.

2. Fa'atulaga le so'oga lava ia

O lenei e mafai ona e alu saʻo i le faʻatulagaina o le fesoʻotaʻiga lava ia, lea o le a faitau ai faʻamatalaga mai le punavai.

Seʻi o tatou vaʻavaʻai i le faʻataʻitaʻiga o fesoʻotaʻiga mo DBMS e lua: PostgreSQL ma MongoDB, lea ou te maua ai le poto masani ma o loʻo i ai eseesega (e ui ina laʻititi, ae i nisi tulaga taua!).

O le faʻatulagaga o loʻo faʻamatalaina i le JSON notation ma tuʻuina atu i le Kafka Connect e faʻaaoga ai se talosaga POST.

2.1. PostgreSQL

Fa'ata'ita'iga feso'ota'iga feso'ota'iga mo 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"
  }
}

O le mataupu faavae o le faʻaogaina o le fesoʻotaʻiga pe a maeʻa lenei faʻatulagaga e faigofie lava:

• I le amataga muamua, e faʻafesoʻotaʻi i le faʻamaumauga o loʻo faʻamaonia i le faʻatulagaina ma amata i le faiga ata muamua, auina atu i Kafka le seti muamua o faʻamaumauga na maua ma le tuutuuga SELECT * FROM table_name.
• A maeʻa le amataga, e ulufale le fesoʻotaʻiga i le faiga o le faitau suiga mai faila PostgreSQL WAL.

E uiga i filifiliga na fa'aaogaina:

• name - le igoa o le fesoʻotaʻiga lea e faʻaogaina ai le faʻatulagaga o loʻo faʻamatalaina i lalo; i le lumanaʻi, o lenei igoa e faʻaaogaina e galulue ai ma le fesoʻotaʻiga (e pei o le vaʻai i le tulaga / toe amata / faʻafouina le faʻatulagaina) e ala i le Kafka Connect REST API;
• connector.class - le vasega faʻafesoʻotaʻi DBMS o le a faʻaaogaina e le fesoʻotaʻiga faʻapipiʻi;
• plugin.name o le igoa o le plugin mo le fa'avasegaina o fa'amaumauga mai faila WAL. Avanoa e filifili mai ai wal2json, decoderbuffs и pgoutput. O le lua muamua e manaʻomia le faʻapipiʻiina o faʻaopoopoga talafeagai i le DBMS, ma pgoutput mo PostgreSQL version 10 ma le maualuga e le manaʻomia ni faʻaoga faaopoopo;
• database.* - filifiliga mo le faʻafesoʻotaʻi i le database, i fea database.server.name - o le igoa o le PostgreSQL faʻataʻitaʻiga na faʻaaogaina e fausia ai le igoa o le autu i le vaega o Kafka;
• table.include.list - se lisi o laulau tatou te mananaʻo e siaki suiga; tuuina atu i le faatulagaga schema.table_name; e le mafai ona faʻaaogaina faʻatasi ma table.exclude.list;
• heartbeat.interval.ms - vaeluaga (i milliseconds) lea e auina atu ai e le soʻo se feʻau tata o fatu i se autu faʻapitoa;
• heartbeat.action.query - o se talosaga o le a faʻataunuʻuina pe a lafoina feʻau taʻitasi taʻitasi (o le filifiliga na faʻaalia talu mai le version 1.1);
• slot.name - le igoa o le faʻasologa o le slot o le a faʻaaogaina e le fesoʻotaʻiga;
• publication.name - Igoa lomiga i PostgreSQL o loʻo faʻaogaina e le fesoʻotaʻiga. I le tulaga e le o iai, o le a taumafai Debezium e fatuina. Afai o le tagata faʻaoga lea e faia ai le fesoʻotaʻiga e le lava le aia mo lenei gaioiga, o le a alu ese le fesoʻotaʻiga ma se mea sese;
• transforms fuafua pe fa'afefea tonu ona sui le igoa ole autu fa'atatau:
  • transforms.AddPrefix.type ua faailoa mai ai o le a tatou faaaogaina faaupuga masani;
  • transforms.AddPrefix.regex — ufimata lea e toe faauigaina ai le igoa o le autu;
  • transforms.AddPrefix.replacement - tuusa'o mea tatou te toe faauigaina.

E uiga i le tātā fatu ma suiga

Ona o le faaletonu, e auina atu e le so'o se fa'amatalaga i Kafka mo fefa'ataua'iga ta'itasi, ma tusi lana LSN (Log Sequence Numera) i le autu o auaunaga. offset. Ae o le a le mea e tupu pe a faʻapipiʻi le fesoʻotaʻiga e le faitau le database atoa, ae naʻo se vaega o ana laulau (lea e faʻafouina faʻamatalaga)?

• O le feso'ota'iga o le a faitau faila WAL ma e le iloa fefa'atauaiga o lo'o faia i totonu o latou laulau i luga o laulau na te mata'ituina.
• O le mea lea, o le a le faʻafouina lona tulaga o loʻo i ai nei pe i le autu poʻo le faʻasologa o faʻasologa.
• O le mea lea, o le a mafua ai ona "mau" faila WAL i luga o le tisiki ma e foliga mai o le a leai se avanoa.

Ma o iinei e sau ai filifiliga e laveai. heartbeat.interval.ms и heartbeat.action.query. O le fa'aogaina o nei filifiliga i paga e mafai ai ona fa'atino se talosaga e sui fa'amaumauga i se isi laulau i taimi uma e lafo ai se fe'au tata'o. O le mea lea, o le LSN o loʻo i ai nei le fesoʻotaʻiga (i totonu o le faʻasologa o faʻasologa) e faʻafouina pea. E mafai ai e le DBMS ona aveese faila WAL e le o toe mana'omia. Mo nisi fa'amatalaga ile auala e galue ai filifiliga, va'ai fa'amaumauga.

O le isi filifiliga e tatau ona vaʻaia lelei transforms. E ui lava e sili atu i le faigofie ma le matagofie ...

Ona o le le mafai, Debezium e fatuina autu e faʻaaoga ai le faiga faʻaigoaina nei: serverName.schemaName.tableName. Atonu e le faigofie i taimi uma. Filifiliga transforms fa'aaogaina fa'amatalaga masani, e mafai ona e fa'amatalaina se lisi o laulau o mea na tutupu e mana'omia ona fa'aulu i se autu ma se igoa patino.

I la matou faatulagaga faafetai i transforms mea nei e tupu: o mea uma a le CDC mai le fa'amaumauga fa'amaumauga o le a alu i le autu ma le igoa data.cdc.dbname. A leai (e aunoa ma nei tulaga), e le mafai e Debezium ona fatuina se autu mo laulau taʻitasi o le fomu: pg-dev.public.<table_name>.

Fa'agata feso'ota'iga

I le faaiuga o le faʻamatalaga o le fesoʻotaʻiga fesoʻotaʻiga mo PostgreSQL, e taua le talanoa e uiga i mea nei / tapulaʻa o lana galuega:

1. Ole fa'aogaina ole feso'ota'iga mo le PostgreSQL e fa'alagolago ile manatu ole fa'avasegaina talafeagai. O lea na ia e le siaki talosaga e sui le fausaga o le database (DDL) - e tusa ai, o nei faʻamatalaga o le a le i totonu o autu.
2. Talu ai o lo'o fa'aogaina avanoa fa'aopoopo, e mafai ona feso'ota'i le feso'ota'iga ae i le master DBMS faʻataʻitaʻiga.
3. Afai o le tagata faʻaoga lea e fesoʻotaʻi ai le fesoʻotaʻiga i le database e iai naʻo le faitau, ona e manaʻomia lea aʻo leʻi amataina le faʻalauiloa muamua, e te manaʻomia le faia ma le lima o se kopi faʻapipiʻi ma faʻasalalau i le database.

Fa'aogaina o se fa'atulagaga

Se'i o tatou utaina la tatou faatulagaga i totonu o le so'oga:

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

Matou te siakiina na manuia le download ma amata le fesoʻotaʻiga:

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

Manaia: ua fa'atulaga ma sauni e alu. Ia tatou faafoliga o se tagata faʻatau ma faʻafesoʻotaʻi i Kafka, a maeʻa ona tatou faʻaopoopo ma suia se faʻamatalaga i le laulau:

$ 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

I la tatou autu, o le a faʻaalia lenei mea:

E umi tele JSON ma a tatou suiga

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

I tulaga uma e lua, o faʻamaumauga e aofia ai le ki (PK) o faʻamaumauga na suia, ma le ute tonu o suiga: o le a le faʻamaumauga muamua ma le mea na tupu mulimuli ane.

• I le tulaga o INSERT: taua muamua (before) tutusa nullsosoo ai ma le manoa na faaofiina.
• I le tulaga o UPDATE: i totonu payload.before o le tulaga muamua o le laina o loʻo faʻaalia, ma i totonu payload.after - fou ma le ute o suiga.

2.2 MongoDB

E fa'aogaina e lenei feso'ota'iga le faiga masani a MongoDB, faitau fa'amatalaga mai le oplog o le node muamua a le DBMS.

E tutusa ma le fesoʻotaʻiga ua uma ona faʻamatalaina mo PgSQL, iinei foi, i le amataga muamua, o le ata muamua o faʻamaumauga e ave, a maeʻa ona sui le fesoʻotaʻiga i le oplog reading mode.

Fa'ata'ita'iga fa'atusa:

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

E pei ona mafai ona e vaʻai, e leai ni filifiliga fou e faʻatusatusa i le faʻataʻitaʻiga muamua, ae naʻo le numera o filifiliga e nafa ma le faʻafesoʻotaʻi i le database ma o latou prefixes ua faʻaititia.

Faʻasologa transforms o le taimi lenei latou te faia mea nei: liliu le igoa o le autu autu mai le polokalame <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

faapalepale sese

O le mataupu o le faapalepale o sese ma le maualuga o avanoa i o tatou taimi e sili atu ona mataʻutia nai lo se isi lava taimi - aemaise lava pe a tatou talanoa e uiga i faʻamatalaga ma fefaʻatauaiga, ma le suʻesuʻeina o suiga o faʻamaumauga e leʻo i luga ole laiga i lenei mataupu. Seʻi o tatou vaʻavaʻai i mea e mafai ona sese i le mataupu faavae ma le mea o le a tupu ia Debezium i tulaga taʻitasi.

E tolu filifiliga e filifili ai e alu ese:

1. Kafka Connect toilalo. Afai e fa'apipi'i le Feso'ota'i e galue i le fa'asoa fa'asoa, e mana'omia ai le tele o tagata faigaluega e fa'atulaga le group.id tutusa. Ma, afai e le manuia se tasi, o le a toe amataina le fesoʻotaʻiga i luga o le isi tagata faigaluega ma faʻaauau le faitau mai le tulaga tuʻufaʻatasia mulimuli i le autu i Kafka.
2. Leai se feso'ota'iga ma le fuifui Kafka. O le feso'ota'iga o le a na ona taofi le faitau i le tulaga na le mafai ona lafo i Kafka ma taumafai i lea taimi ma lea taimi e toe auina atu seia manuia le taumafaiga.
3. Le maua mai fa'amaumauga. O le a taumafai le so'oga e toe fa'afeso'ota'i i le puna e tusa ai ma le fa'atulagaga. Ole faaletonu ole 16 taumafai e fa'aaoga fa'ato'a fa'ateteleina. A mae'a le taumafaiga lona 16 na le manuia, o le a fa'ailogaina le galuega mafai ma e manaʻomia le toe faʻafouina ma le lima e ala i le Kafka Connect REST interface.
   • I le tulaga o PostgreSQL o le a le leiloa faamatalaga, aua o le fa'aogaina o avanoa fa'aopoopo e taofia ai le tapeina o faila WAL e le'i faitauina e le so'o. I lenei tulaga, o loʻo i ai se faʻaletonu: afai o le fesoʻotaʻiga fesoʻotaʻiga i le va o le fesoʻotaʻiga ma le DBMS ua faʻalavelaveina mo se taimi umi, o loʻo i ai se avanoa e alu ese ai le vaʻa, ma e ono oʻo atu ai i le toilalo o le DBMS atoa.
   • I le tulaga o MySQL O faila binlog e mafai ona fesuia'i e le DBMS lava ia a'o le'i toe fa'aleleia le feso'ota'iga. O lenei mea o le a mafua ai ona alu le fesoʻotaʻiga i le tulaga le manuia, ma o le a manaʻomia le toe amata i le ata muamua ata e faʻaauau ai le faitau mai binlogs e toe faʻafoʻi ai galuega masani.
   • i MongoDB. Fai mai le faʻamaumauga: o le amio a le fesoʻotaʻiga pe a fai ua tape faila log / oplog ma e le mafai e le soʻo ona faʻaauau le faitau mai le tulaga na tuʻu ai e tutusa mo DBMS uma. O loʻo taoto i le mea moni o le fesoʻotaʻiga o le a alu i totonu o le setete mafai ma o le a manaʻomia se toe amata i le faiga ata muamua.

     Peitaʻi, e iai tuusaunoaga. Afai o le fesoʻotaʻiga sa i ai i se tulaga motusia mo se taimi umi (pe le mafai ona oʻo atu i le MongoDB instance), ma oplog na feliuliuaʻi i le taimi lea, ona toe faʻaleleia lea o le fesoʻotaʻiga, o le a faʻaauau pea ona faitau le fesoʻotaʻiga faʻamatalaga mai le avanoa muamua. , o le mafuaaga lea o nisi o faʻamatalaga i Kafka lē o le a taia.

iʻuga

O le Debezium o lo'u aafiaga muamua lea i faiga CDC ma ua matua lelei tele. O le poloketi na faʻatauaina le lagolago a le DBMS autu, faigofie o le faʻatulagaina, lagolago mo le faʻapipiʻiina ma se faʻalapotopotoga malosi. Mo i latou e fiafia i faʻataʻitaʻiga, ou te fautuaina oe e faitau le taʻiala mo Kafka Fesootai и Debesium.

Pe a faatusatusa i le JDBC connector mo Kafka Connect, o le aoga autu o Debezium o suiga e faitau mai le DBMS logs, lea e mafai ai ona maua faʻamatalaga ma sina faʻatuai. O le JDBC Connector (na saunia e Kafka Connect) e fesiligia le laulau siaki i se taimi tumau ma (mo le mafuaʻaga lava e tasi) e le faʻatupuina ni feʻau pe a tape faʻamaumauga (faʻafefea ona e fesili mo faʻamatalaga e le o iai?).

Ina ia foia faafitauli faapena, e mafai ona e gauai atu i fofo nei (faʻaopoopo i Debezium):

• Fa'afeso'ota'i JDBC Kafka Feso'ota'i
• O nai fofo MySQL-na'o:
  • MySQL Streamer;
  • Maxwell;
  • Tapula'i Tula'au;
• Faitotoa Auro Oracle, ae o se "vaega mamafa" e matua ese lava lenei.

SALA

Faitau foi i la matou blog:

• «Su'e le tele talafeagai mo se fuifui Kafka i Kubernetes";
• «Tala fa'ata'ita'i mai le tatou SRE o le olaga i aso faisoo. Vaega 2";
• «Ose Va'aiga Aotelega o Fa'amatalaga PostgreSQL mo Kubernetes, O Tatou Filifiliga ma le Aafiaga".

puna: www.habr.com