ແນະນໍາ Debezium - CDC ສໍາລັບ Apache Kafka

ໃນການເຮັດວຽກຂອງຂ້ອຍ, ຂ້ອຍມັກຈະພົບວິທີແກ້ໄຂດ້ານວິຊາການໃຫມ່ / ຜະລິດຕະພັນຊອບແວ, ຂໍ້ມູນກ່ຽວກັບສິ່ງທີ່ບໍ່ຄ່ອຍມີຢູ່ໃນອິນເຕີເນັດທີ່ເວົ້າພາສາລັດເຊຍ. ດ້ວຍບົດຄວາມນີ້, ຂ້ອຍຈະພະຍາຍາມຕື່ມຂໍ້ມູນໃສ່ຊ່ອງຫວ່າງດັ່ງກ່າວດ້ວຍຕົວຢ່າງຈາກການປະຕິບັດທີ່ຜ່ານມາຂອງຂ້ອຍ, ເມື່ອຂ້ອຍຕ້ອງຕັ້ງຄ່າການສົ່ງເຫດການ CDC ຈາກສອງ DBMSs ທີ່ນິຍົມ (PostgreSQL ແລະ MongoDB) ໄປຫາກຸ່ມ Kafka ໂດຍໃຊ້ Debezium. ຂ້າພະເຈົ້າຫວັງວ່າບົດຄວາມທົບທວນນີ້, ເຊິ່ງປະກົດວ່າເປັນຜົນມາຈາກການເຮັດວຽກທີ່ເຮັດ, ຈະເປັນປະໂຫຍດຕໍ່ຄົນອື່ນ.

Debezium ແລະ CDC ໂດຍທົ່ວໄປແມ່ນຫຍັງ?

ເດເບຊຽມ - ຜູ້ຕາງຫນ້າຂອງປະເພດຊອບແວ CDC (ບັນທຶກການປ່ຽນແປງຂໍ້ມູນ), ຫຼືຫຼາຍກວ່າທີ່ຊັດເຈນ, ມັນແມ່ນຊຸດຂອງຕົວເຊື່ອມຕໍ່ສໍາລັບ DBMSs ຕ່າງໆທີ່ເຫມາະສົມກັບກອບ Apache Kafka Connect.

ນີ້ ໂຄງ​ການ​ແຫຼ່ງ​ເປີດ​, ໄດ້ຮັບອະນຸຍາດພາຍໃຕ້ Apache License v2.0 ແລະໄດ້ຮັບການສະໜັບສະໜູນໂດຍ Red Hat. ການພັດທະນາໄດ້ດໍາເນີນມາຕັ້ງແຕ່ປີ 2016 ແລະໃນປັດຈຸບັນມັນສະຫນອງການສະຫນັບສະຫນູນຢ່າງເປັນທາງການສໍາລັບ DBMS ຕໍ່ໄປນີ້: MySQL, PostgreSQL, MongoDB, SQL Server. ຍັງມີຕົວເຊື່ອມຕໍ່ສໍາລັບ Cassandra ແລະ Oracle, ແຕ່ພວກເຂົາຢູ່ໃນສະຖານະການ "ເຂົ້າເຖິງໄວ", ແລະການອອກໃຫມ່ບໍ່ໄດ້ຮັບປະກັນຄວາມເຂົ້າກັນໄດ້ໃນດ້ານຫລັງ.

ຖ້າພວກເຮົາປຽບທຽບ CDC ກັບວິທີການແບບດັ້ງເດີມ (ເມື່ອແອັບພລິເຄຊັນອ່ານຂໍ້ມູນຈາກ DBMS ໂດຍກົງ), ປະໂຫຍດຕົ້ນຕໍຂອງມັນປະກອບມີການປະຕິບັດການຖ່າຍທອດຂໍ້ມູນໃນລະດັບແຖວທີ່ມີ latency ຕ່ໍາ, ຄວາມຫນ້າເຊື່ອຖືສູງແລະຄວາມພ້ອມ. ສອງຈຸດສຸດທ້າຍແມ່ນບັນລຸໄດ້ໂດຍການໃຊ້ກຸ່ມ Kafka ເປັນບ່ອນເກັບມ້ຽນຂອງເຫດການ CDC.

ນອກຈາກນີ້, ຂໍ້ດີລວມເຖິງຄວາມຈິງທີ່ວ່າຮູບແບບດຽວຖືກນໍາໃຊ້ເພື່ອເກັບຮັກສາເຫດການ, ດັ່ງນັ້ນຄໍາຮ້ອງສະຫມັກສຸດທ້າຍບໍ່ຈໍາເປັນຕ້ອງກັງວົນກ່ຽວກັບ nuances ຂອງການດໍາເນີນງານທີ່ແຕກຕ່າງກັນ DBMS.

ສຸດທ້າຍ, ການນໍາໃຊ້ນາຍຫນ້າຂໍ້ຄວາມເປີດຂອບເຂດສໍາລັບການຂະຫຍາຍອອກຕາມລວງນອນຂອງຄໍາຮ້ອງສະຫມັກທີ່ຕິດຕາມການປ່ຽນແປງຂໍ້ມູນ. ໃນຂະນະດຽວກັນ, ຜົນກະທົບຕໍ່ແຫຼ່ງຂໍ້ມູນແມ່ນຫນ້ອຍທີ່ສຸດ, ເນື່ອງຈາກວ່າຂໍ້ມູນບໍ່ໄດ້ຮັບໂດຍກົງຈາກ DBMS, ແຕ່ຈາກກຸ່ມ Kafka.

ກ່ຽວກັບສະຖາປັດຕະຍະກໍາ Debezium

ການນໍາໃຊ້ Debezium ມາລົງໃນໂຄງການງ່າຍດາຍນີ້:

DBMS (ເປັນແຫຼ່ງຂໍ້ມູນ) → ຕົວເຊື່ອມຕໍ່ໃນ Kafka Connect → Apache Kafka → ຜູ້ບໍລິໂພກ

ໃນຖານະເປັນຕົວຢ່າງ, ຂ້າພະເຈົ້າຈະໃຫ້ແຜນວາດຈາກເວັບໄຊທ໌ຂອງໂຄງການ:

ຢ່າງໃດກໍ່ຕາມ, ຂ້ອຍບໍ່ມັກໂຄງການນີ້, ເພາະວ່າມັນເບິ່ງຄືວ່າມີພຽງແຕ່ຕົວເຊື່ອມຕໍ່ບ່ອນຫລົ້ມຈົມເທົ່ານັ້ນ.

ໃນຄວາມເປັນຈິງ, ສະຖານະການແມ່ນແຕກຕ່າງກັນ: ການຕື່ມຂໍ້ມູນ Lake ຂອງທ່ານ (ລິ້ງສຸດທ້າຍໃນແຜນວາດຂ້າງເທິງ) ບໍ່ແມ່ນວິທີດຽວທີ່ຈະໃຊ້ Debezium. ເຫດການທີ່ຖືກສົ່ງໄປຫາ Apache Kafka ສາມາດຖືກນໍາໃຊ້ໂດຍຄໍາຮ້ອງສະຫມັກຂອງທ່ານເພື່ອຈັດການກັບສະຖານະການຕ່າງໆ. ຍົກ​ຕົວ​ຢ່າງ:

• ການກໍາຈັດຂໍ້ມູນທີ່ບໍ່ກ່ຽວຂ້ອງອອກຈາກ cache;
• ການ​ສົ່ງ​ແຈ້ງ​ການ​;
• ການປັບປຸງດັດສະນີຄົ້ນຫາ;
• ບາງປະເພດຂອງບັນທຶກການກວດສອບ;
• ...

ໃນ​ກໍ​ລະ​ນີ​ທີ່​ທ່ານ​ມີ​ຄໍາ​ຮ້ອງ​ສະ​ຫມັກ Java ແລະ​ບໍ່​ມີ​ຄວາມ​ຕ້ອງ​ການ / ຄວາມ​ເປັນ​ໄປ​ໄດ້​ທີ່​ຈະ​ນໍາ​ໃຊ້​ກຸ່ມ Kafka​, ມັນ​ຍັງ​ມີ​ຄວາມ​ເປັນ​ໄປ​ໄດ້​ທີ່​ຈະ​ເຮັດ​ວຽກ​ໂດຍ​ຜ່ານ​ການ​. ຕົວເຊື່ອມຕໍ່ຝັງ. ບວກທີ່ຊັດເຈນແມ່ນວ່າກັບມັນທ່ານສາມາດປະຕິເສດໂຄງສ້າງພື້ນຖານເພີ່ມເຕີມ (ໃນຮູບແບບຂອງຕົວເຊື່ອມຕໍ່ແລະ Kafka). ແນວໃດກໍ່ຕາມ, ການແກ້ໄຂນີ້ໄດ້ຖືກຍົກເລີກຕັ້ງແຕ່ເວີຊັນ 1.1 ແລະບໍ່ຖືກແນະນຳໃຫ້ໃຊ້ອີກຕໍ່ໄປ (ມັນອາດຈະຖືກລຶບອອກໃນລຸ້ນໃນອະນາຄົດ).

ບົດ​ຄວາມ​ນີ້​ຈະ​ປຶກ​ສາ​ຫາ​ລື​ສະ​ຖາ​ປັດ​ຕະ​ທີ່​ແນະ​ນໍາ​ໂດຍ​ຜູ້​ພັດ​ທະ​ນາ​, ທີ່​ໃຫ້​ຄວາມ​ທົນ​ທານ​ຄວາມ​ຜິດ​ພາດ​ແລະ​ການ​ຂະ​ຫຍາຍ​ຕົວ​ໄດ້​.

ການຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່

ເພື່ອເລີ່ມຕົ້ນການຕິດຕາມການປ່ຽນແປງໃນມູນຄ່າທີ່ສໍາຄັນທີ່ສຸດ - ຂໍ້ມູນ - ພວກເຮົາຕ້ອງການ:

1. ແຫຼ່ງຂໍ້ມູນ, ເຊິ່ງສາມາດເປັນ MySQL ເລີ່ມຈາກເວີຊັນ 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (ບັນຊີລາຍຊື່ຄົບຖ້ວນ);
2. ກຸ່ມ Apache Kafka
3. Kafka Connect instance (ເວີຊັ່ນ 1.x, 2.x);
4. ຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່ Debezium.

ເຮັດວຽກສອງຈຸດທໍາອິດ, i.e. ຂະບວນການຕິດຕັ້ງ DBMS ແລະ Apache Kafka ແມ່ນເກີນຂອບເຂດຂອງບົດຄວາມ. ຢ່າງໃດກໍຕາມ, ສໍາລັບຜູ້ທີ່ຕ້ອງການທີ່ຈະນໍາໃຊ້ທຸກສິ່ງທຸກຢ່າງໃນ sandbox, ມີຫນຶ່ງທີ່ກຽມພ້ອມໃນ repository ຢ່າງເປັນທາງການທີ່ມີຕົວຢ່າງ. docker-compose.yaml.

ພວກເຮົາຈະສຸມໃສ່ສອງຈຸດສຸດທ້າຍໃນລາຍລະອຽດເພີ່ມເຕີມ.

0. Kafka Connect

ທີ່ນີ້ແລະຕໍ່ມາໃນບົດຄວາມ, ຕົວຢ່າງການຕັ້ງຄ່າທັງຫມົດຖືກພິຈາລະນາໃນສະພາບການຂອງຮູບພາບ Docker ທີ່ແຈກຢາຍໂດຍນັກພັດທະນາ Debezium. ມັນປະກອບດ້ວຍໄຟລ໌ plugin ທີ່ຈໍາເປັນທັງຫມົດ (ຕົວເຊື່ອມຕໍ່) ແລະສະຫນອງການຕັ້ງຄ່າ Kafka Connect ໂດຍໃຊ້ຕົວແປສະພາບແວດລ້ອມ.

ຖ້າທ່ານຕັ້ງໃຈຈະໃຊ້ Kafka Connect ຈາກ Confluent, ທ່ານຈະຕ້ອງເພີ່ມ plugins ຂອງຕົວເຊື່ອມຕໍ່ທີ່ຈໍາເປັນດ້ວຍຕົວທ່ານເອງກັບໄດເລກະທໍລີທີ່ລະບຸໄວ້ໃນ. plugin.path ຫຼືຕັ້ງຜ່ານຕົວແປສະພາບແວດລ້ອມ CLASSPATH. ການຕັ້ງຄ່າສໍາລັບ Kafka Connect worker ແລະ connectors ແມ່ນຖືກກໍານົດໂດຍຜ່ານໄຟລ໌ການຕັ້ງຄ່າທີ່ຖືກສົ່ງຜ່ານເປັນການໂຕ້ຖຽງກັບຄໍາສັ່ງເລີ່ມຕົ້ນຂອງພະນັກງານ. ສໍາລັບລາຍລະອຽດເບິ່ງ ເອກະສານ.

ຂະບວນການທັງຫມົດຂອງການຕັ້ງຄ່າ Debeizum ໃນສະບັບເຊື່ອມຕໍ່ແມ່ນດໍາເນີນໃນສອງຂັ້ນຕອນ. ຂໍ​ໃຫ້​ພິຈາລະນາ​ພວກ​ເຂົາ​ແຕ່​ລະ​ຄົນ:

1. ການຕັ້ງຄ່າກອບ Kafka Connect

ເພື່ອຖ່າຍທອດຂໍ້ມູນໄປຫາກຸ່ມ Apache Kafka, ຕົວກໍານົດການສະເພາະແມ່ນຖືກກໍານົດໄວ້ໃນກອບ Kafka Connect, ເຊັ່ນ:

• ການ​ຕັ້ງ​ຄ່າ​ການ​ເຊື່ອມ​ຕໍ່​ກຸ່ມ​,
• ຊື່ຂອງຫົວຂໍ້ທີ່ການຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່ຂອງມັນເອງຈະຖືກເກັບໄວ້,
• ຊື່ຂອງກຸ່ມທີ່ຕົວເຊື່ອມຕໍ່ກໍາລັງແລ່ນ (ໃນກໍລະນີຂອງການນໍາໃຊ້ຮູບແບບການແຈກຢາຍ).

ຮູບພາບ Docker ຢ່າງເປັນທາງການຂອງໂຄງການສະຫນັບສະຫນູນການຕັ້ງຄ່າໂດຍໃຊ້ຕົວແປສະພາບແວດລ້ອມ - ນີ້ແມ່ນສິ່ງທີ່ພວກເຮົາຈະໃຊ້. ດັ່ງ​ນັ້ນ​ໃຫ້​ດາວ​ນ​໌​ໂຫລດ​ຮູບ​ພາບ​:

docker pull debezium/connect

ຊຸດຕົວແປສະພາບແວດລ້ອມຂັ້ນຕ່ຳທີ່ຕ້ອງການເພື່ອແລ່ນຕົວເຊື່ອມຕໍ່ມີດັ່ງນີ້:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 - ບັນຊີລາຍຊື່ເບື້ອງຕົ້ນຂອງເຄື່ອງແມ່ຂ່າຍຂອງກຸ່ມ Kafka ເພື່ອໃຫ້ໄດ້ຮັບບັນຊີລາຍຊື່ຄົບຖ້ວນຂອງສະມາຊິກກຸ່ມ;
• OFFSET_STORAGE_TOPIC=connector-offsets — ຫົວຂໍ້ສໍາລັບການເກັບຮັກສາຕໍາແຫນ່ງທີ່ຕົວເຊື່ອມຕໍ່ຕັ້ງຢູ່ໃນປະຈຸບັນ;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status - ຫົວຂໍ້ສໍາລັບການເກັບຮັກສາສະຖານະພາບຂອງຕົວເຊື່ອມຕໍ່ແລະວຽກງານຂອງມັນ;
• CONFIG_STORAGE_TOPIC=connector-config - ຫົວຂໍ້ສໍາລັບການເກັບຮັກສາຂໍ້ມູນການຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່ແລະວຽກງານຂອງມັນ;
• GROUP_ID=1 — ຕົວລະບຸກຸ່ມຄົນງານທີ່ສາມາດປະຕິບັດໜ້າທີ່ເຊື່ອມຕໍ່; ຕ້ອງການໃນເວລາທີ່ການນໍາໃຊ້ແຈກຢາຍ (ແຈກຢາຍ) ລະບອບ.

ພວກເຮົາເລີ່ມຕົ້ນ container ດ້ວຍຕົວແປເຫຼົ່ານີ້:

docker run 
  -e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092' 
  -e GROUP_ID=1 
  -e CONFIG_STORAGE_TOPIC=my_connect_configs 
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets 
  -e STATUS_STORAGE_TOPIC=my_connect_statuses  debezium/connect:1.2

ຫມາຍເຫດກ່ຽວກັບ Avro

ໂດຍຄ່າເລີ່ມຕົ້ນ, Debezium ຂຽນຂໍ້ມູນໃນຮູບແບບ JSON, ເຊິ່ງຍອມຮັບໄດ້ສໍາລັບ sandboxes ແລະຂໍ້ມູນຈໍານວນນ້ອຍໆ, ແຕ່ສາມາດເປັນບັນຫາໃນຖານຂໍ້ມູນທີ່ໂຫລດຫຼາຍ. ທາງເລືອກຂອງຕົວແປງ JSON ແມ່ນເພື່ອ serialize ຂໍ້ຄວາມໂດຍໃຊ້ Avro ກັບຮູບແບບຖານສອງ, ເຊິ່ງຫຼຸດຜ່ອນການໂຫຼດໃນລະບົບຍ່ອຍ I / O ໃນ Apache Kafka.

ເພື່ອໃຊ້ Avro, ທ່ານຈໍາເປັນຕ້ອງໃຊ້ແຍກຕ່າງຫາກ schema-registry (ສໍາລັບການເກັບຮັກສາ schemas). ຕົວແປຂອງຕົວແປງສັນຍານຈະມີລັກສະນະນີ້:

name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER   
value: io.confluent.connect.avro.AvroConverter

ລາຍລະອຽດກ່ຽວກັບການນໍາໃຊ້ Avro ແລະການຕັ້ງຄ່າການລົງທະບຽນສໍາລັບມັນເກີນຂອບເຂດຂອງບົດຄວາມ - ຕື່ມອີກ, ເພື່ອຄວາມຊັດເຈນ, ພວກເຮົາຈະໃຊ້ JSON.

2. ການຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່ຕົວມັນເອງ

ໃນປັດຈຸບັນທ່ານສາມາດໄປໂດຍກົງກັບການຕັ້ງຄ່າຂອງຕົວເຊື່ອມຕໍ່ຕົວມັນເອງ, ເຊິ່ງຈະອ່ານຂໍ້ມູນຈາກແຫຼ່ງ.

ຂໍໃຫ້ເບິ່ງຕົວຢ່າງຂອງຕົວເຊື່ອມຕໍ່ສໍາລັບສອງ DBMS: PostgreSQL ແລະ MongoDB, ທີ່ຂ້ອຍມີປະສົບການແລະມີຄວາມແຕກຕ່າງ (ເຖິງແມ່ນວ່າຈະນ້ອຍ, ແຕ່ໃນບາງກໍລະນີທີ່ສໍາຄັນ!).

ການຕັ້ງຄ່າໄດ້ຖືກອະທິບາຍໄວ້ໃນ JSON notation ແລະອັບໂຫລດໄປຍັງ Kafka Connect ໂດຍໃຊ້ຄໍາຮ້ອງຂໍ POST.

2.1. PostgreSQL

ຕົວຢ່າງການຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່ສໍາລັບ PostgreSQL:

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "definitelynotpassword",
    "database.dbname" : "dbname",
    "database.server.name": "pg-dev",
    "table.include.list": "public.(.*)",
    "heartbeat.interval.ms": "5000",
    "slot.name": "dbname_debezium",
    "publication.name": "dbname_publication",
    "transforms": "AddPrefix",
    "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex": "pg-dev.public.(.*)",
    "transforms.AddPrefix.replacement": "data.cdc.dbname"
  }
}

ຫຼັກການຂອງການດໍາເນີນງານຂອງຕົວເຊື່ອມຕໍ່ຫຼັງຈາກການຕັ້ງຄ່ານີ້ແມ່ນຂ້ອນຂ້າງງ່າຍດາຍ:

• ໃນຕອນເລີ່ມຕົ້ນທໍາອິດ, ມັນເຊື່ອມຕໍ່ກັບຖານຂໍ້ມູນທີ່ລະບຸໄວ້ໃນການຕັ້ງຄ່າແລະເລີ່ມຕົ້ນໃນໂຫມດ ພາບຖ່າຍເບື້ອງຕົ້ນ, ສົ່ງກັບ Kafka ຊຸດຂໍ້ມູນເບື້ອງຕົ້ນທີ່ໄດ້ຮັບດ້ວຍເງື່ອນໄຂ SELECT * FROM table_name.
• ຫຼັງ​ຈາກ​ການ​ເລີ່ມ​ຕົ້ນ​ແມ່ນ​ສໍາ​ເລັດ​, ຕົວ​ເຊື່ອມ​ຕໍ່​ເຂົ້າ​ໄປ​ໃນ​ຮູບ​ແບບ​ການ​ອ່ານ​ການ​ປ່ຽນ​ແປງ​ຈາກ​ໄຟລ​໌ PostgreSQL WAL​.

ກ່ຽວ​ກັບ​ທາງ​ເລືອກ​ທີ່​ນໍາ​ໃຊ້​:

• name — ຊື່​ຂອງ​ຕົວ​ເຊື່ອມ​ຕໍ່​ທີ່​ການ​ຕັ້ງ​ຄ່າ​ອະ​ທິ​ບາຍ​ຂ້າງ​ລຸ່ມ​ນີ້​ໄດ້​ຖືກ​ນໍາ​ໃຊ້​; ໃນອະນາຄົດ, ຊື່ນີ້ຖືກນໍາໃຊ້ເພື່ອເຮັດວຽກກັບຕົວເຊື່ອມຕໍ່ (i. e. ເບິ່ງສະຖານະ / restart / ປັບປຸງການຕັ້ງຄ່າ) ຜ່ານ Kafka Connect REST API;
• connector.class — ຫ້ອງຮຽນຕົວເຊື່ອມຕໍ່ DBMS ທີ່ຈະໃຊ້ໂດຍຕົວເຊື່ອມຕໍ່ທີ່ກຳນົດຄ່າ;
• plugin.name ແມ່ນຊື່ຂອງ plugin ສໍາລັບການຖອດລະຫັດຢ່າງມີເຫດຜົນຂອງຂໍ້ມູນຈາກໄຟລ໌ WAL. ມີໃຫ້ເລືອກ wal2json, decoderbuffs и pgoutput. ສອງອັນທໍາອິດຮຽກຮ້ອງໃຫ້ມີການຕິດຕັ້ງສ່ວນຂະຫຍາຍທີ່ເຫມາະສົມໃນ DBMS, ແລະ pgoutput ສໍາລັບ PostgreSQL ຮຸ່ນ 10 ແລະສູງກວ່າບໍ່ຮຽກຮ້ອງໃຫ້ມີການຫມູນໃຊ້ເພີ່ມເຕີມ;
• database.* — ທາງ​ເລືອກ​ສໍາ​ລັບ​ການ​ເຊື່ອມ​ຕໍ່​ກັບ​ຖານ​ຂໍ້​ມູນ​, ບ່ອນ​ທີ່​ database.server.name - ຊື່ຂອງຕົວຢ່າງ PostgreSQL ທີ່ໃຊ້ເພື່ອສ້າງຊື່ຂອງຫົວຂໍ້ໃນກຸ່ມ Kafka;
• table.include.list - ບັນຊີລາຍຊື່ຂອງຕາຕະລາງທີ່ພວກເຮົາຕ້ອງການທີ່ຈະຕິດຕາມການປ່ຽນແປງ; ມອບໃຫ້ໃນຮູບແບບ schema.table_name; ບໍ່ສາມາດໃຊ້ຮ່ວມກັນກັບ table.exclude.list;
• heartbeat.interval.ms — ໄລຍະຫ່າງ (ເປັນ milliseconds) ທີ່ຕົວເຊື່ອມຕໍ່ສົ່ງຂໍ້ຄວາມຫົວໃຈເຕັ້ນໄປຫາຫົວຂໍ້ພິເສດ;
• heartbeat.action.query - ຄໍາ​ຮ້ອງ​ສະ​ຫມັກ​ທີ່​ຈະ​ໄດ້​ຮັບ​ການ​ປະ​ຕິ​ບັດ​ໃນ​ເວ​ລາ​ທີ່​ສົ່ງ​ຂໍ້​ຄວາມ​ເຕັ້ນ​ຂອງ​ຫົວ​ໃຈ​ແຕ່​ລະ​ຄົນ (ທາງ​ເລືອກ​ທີ່​ໄດ້​ປາ​ກົດ​ຕັ້ງ​ແຕ່​ສະ​ບັບ 1.1​)​;
• slot.name — ຊື່ຂອງຊ່ອງ replication ທີ່ຈະຖືກນໍາໃຊ້ໂດຍຕົວເຊື່ອມຕໍ່;
• publication.name - ຊື່ publications ໃນ PostgreSQL ທີ່ຕົວເຊື່ອມຕໍ່ໃຊ້. ໃນກໍລະນີທີ່ມັນບໍ່ມີ, Debezium ຈະພະຍາຍາມສ້າງມັນ. ຖ້າຜູ້ໃຊ້ທີ່ເຊື່ອມຕໍ່ນັ້ນບໍ່ມີສິດພຽງພໍສໍາລັບການດໍາເນີນການນີ້, ຕົວເຊື່ອມຕໍ່ຈະອອກດ້ວຍຄວາມຜິດພາດ;
• transforms ກໍານົດວິທີການປ່ຽນຊື່ຂອງຫົວຂໍ້ເປົ້າຫມາຍທີ່ແນ່ນອນ:
  • transforms.AddPrefix.type ຊີ້ບອກວ່າພວກເຮົາຈະໃຊ້ສຳນວນປົກກະຕິ;
  • transforms.AddPrefix.regex — ໜ້າກາກທີ່ຊື່ຂອງຫົວຂໍ້ເປົ້າໝາຍຖືກກຳນົດຄືນໃໝ່;
  • transforms.AddPrefix.replacement - ກົງກັບສິ່ງທີ່ພວກເຮົາກໍານົດຄືນໃຫມ່.

ເພີ່ມເຕີມກ່ຽວກັບການເຕັ້ນຂອງຫົວໃຈແລະການຫັນປ່ຽນ

ໂດຍຄ່າເລີ່ມຕົ້ນ, ຕົວເຊື່ອມຕໍ່ຈະສົ່ງຂໍ້ມູນໄປຫາ Kafka ສໍາລັບແຕ່ລະທຸລະກໍາທີ່ຫມັ້ນສັນຍາ, ແລະຂຽນ LSN ຂອງມັນ (Log Sequence Number) ໃສ່ຫົວຂໍ້ການບໍລິການ offset. ແຕ່ຈະເກີດຫຍັງຂຶ້ນຖ້າຕົວເຊື່ອມຕໍ່ຖືກຕັ້ງຄ່າໃຫ້ອ່ານບໍ່ແມ່ນຖານຂໍ້ມູນທັງໝົດ, ແຕ່ມີພຽງສ່ວນໜຶ່ງຂອງຕາຕະລາງຂອງມັນເທົ່ານັ້ນ (ຂໍ້ມູນໃດຖືກອັບເດດເລື້ອຍໆ)?

• ຕົວເຊື່ອມຕໍ່ຈະອ່ານໄຟລ໌ WAL ແລະບໍ່ກວດພົບການເຮັດທຸລະກໍາໃນພວກມັນກັບຕາຕະລາງທີ່ມັນຕິດຕາມ.
• ດັ່ງນັ້ນ, ມັນຈະບໍ່ປັບປຸງຕໍາແຫນ່ງປະຈຸບັນຂອງມັນບໍ່ວ່າຈະຢູ່ໃນຫົວຂໍ້ຫຼືໃນຊ່ອງ replication.
• ນີ້, ໃນທາງກັບກັນ, ຈະເຮັດໃຫ້ໄຟລ໌ WAL "ຕິດ" ໃນແຜ່ນແລະອາດຈະຫມົດພື້ນທີ່ດິດ.

ແລະໃນທີ່ນີ້ທາງເລືອກມາຫາກູ້ໄພ. heartbeat.interval.ms и heartbeat.action.query. ການນໍາໃຊ້ທາງເລືອກເຫຼົ່ານີ້ເປັນຄູ່ເຮັດໃຫ້ມັນເປັນໄປໄດ້ທີ່ຈະປະຕິບັດການຮ້ອງຂໍການປ່ຽນແປງຂໍ້ມູນໃນຕາຕະລາງແຍກຕ່າງຫາກໃນແຕ່ລະຄັ້ງທີ່ຂໍ້ຄວາມຫົວໃຈຖືກສົ່ງ. ດັ່ງນັ້ນ, LSN ທີ່ຕົວເຊື່ອມຕໍ່ຕັ້ງຢູ່ໃນປະຈຸບັນ (ຢູ່ໃນຊ່ອງ replication) ໄດ້ຖືກປັບປຸງຢ່າງຕໍ່ເນື່ອງ. ນີ້ອະນຸຍາດໃຫ້ DBMS ເອົາໄຟລ໌ WAL ທີ່ບໍ່ຈໍາເປັນອອກ. ສໍາລັບຂໍ້ມູນເພີ່ມເຕີມກ່ຽວກັບວິທີການທາງເລືອກເຮັດວຽກ, ເບິ່ງ ເອກະສານ.

ທາງເລືອກອື່ນທີ່ສົມຄວນໄດ້ຮັບຄວາມສົນໃຈຢ່າງໃກ້ຊິດແມ່ນ transforms. ເຖິງແມ່ນວ່າມັນມີຄວາມສະດວກສະບາຍແລະຄວາມງາມຫຼາຍກວ່າ ...

ໂດຍຄ່າເລີ່ມຕົ້ນ, Debezium ສ້າງຫົວຂໍ້ໂດຍໃຊ້ນະໂຍບາຍການຕັ້ງຊື່ຕໍ່ໄປນີ້: serverName.schemaName.tableName. ນີ້ອາດຈະບໍ່ສະດວກສະ ເໝີ ໄປ. ທາງເລືອກ transforms ການນໍາໃຊ້ການສະແດງອອກປົກກະຕິ, ທ່ານສາມາດກໍານົດບັນຊີລາຍຊື່ຂອງຕາຕະລາງທີ່ເຫດການຕ້ອງຖືກນໍາໄປຫາຫົວຂໍ້ທີ່ມີຊື່ສະເພາະ.

ໃນການຕັ້ງຄ່າຂອງພວກເຮົາຂໍຂອບໃຈກັບ transforms ຕໍ່ໄປນີ້ຈະເກີດຂຶ້ນ: ເຫດການ CDC ທັງຫມົດຈາກຖານຂໍ້ມູນທີ່ຖືກຕິດຕາມຈະໄປຫາຫົວຂໍ້ທີ່ມີຊື່ data.cdc.dbname. ຖ້າບໍ່ດັ່ງນັ້ນ (ໂດຍບໍ່ມີການຕັ້ງຄ່າເຫຼົ່ານີ້), Debezium ຈະສ້າງຫົວຂໍ້ສໍາລັບແຕ່ລະຕາຕະລາງຂອງແບບຟອມ: pg-dev.public.<table_name>.

ຂໍ້ຈໍາກັດຂອງຕົວເຊື່ອມຕໍ່

ໃນຕອນທ້າຍຂອງລາຍລະອຽດຂອງການຕັ້ງຄ່າຕົວເຊື່ອມຕໍ່ສໍາລັບ PostgreSQL, ມັນເປັນມູນຄ່າທີ່ຈະເວົ້າກ່ຽວກັບລັກສະນະຕໍ່ໄປນີ້ / ຂໍ້ຈໍາກັດຂອງການເຮັດວຽກຂອງມັນ:

1. ຫນ້າທີ່ເຊື່ອມຕໍ່ສໍາລັບ PostgreSQL ອີງໃສ່ແນວຄວາມຄິດຂອງການຖອດລະຫັດຢ່າງມີເຫດຜົນ. ເພາະສະນັ້ນ ລາວ ບໍ່ໄດ້ຕິດຕາມການຮ້ອງຂໍການປ່ຽນແປງໂຄງສ້າງຂອງຖານຂໍ້ມູນ (DDL) - ຕາມຄວາມເຫມາະສົມ, ຂໍ້ມູນນີ້ຈະບໍ່ຢູ່ໃນຫົວຂໍ້.
2. ນັບຕັ້ງແຕ່ສະລັອດຕິງ replication ຖືກນໍາໃຊ້, ການເຊື່ອມຕໍ່ຂອງຕົວເຊື່ອມຕໍ່ແມ່ນເປັນໄປໄດ້ ພຽງແຕ່ ກັບຕົວຢ່າງ DBMS ຕົ້ນສະບັບ.
3. ຖ້າຜູ້ໃຊ້ພາຍໃຕ້ຕົວເຊື່ອມຕໍ່ທີ່ເຊື່ອມຕໍ່ກັບຖານຂໍ້ມູນມີສິດອ່ານເທົ່ານັ້ນ, ຫຼັງຈາກນັ້ນ, ກ່ອນທີ່ຈະເປີດຕົວຄັ້ງທໍາອິດ, ທ່ານຈະຕ້ອງສ້າງຊ່ອງ replication ດ້ວຍຕົນເອງແລະເຜີຍແຜ່ໃນຖານຂໍ້ມູນ.

ການ​ນໍາ​ໃຊ້​ການ​ຕັ້ງ​ຄ່າ​

ສະນັ້ນໃຫ້ໂຫຼດການຕັ້ງຄ່າຂອງພວກເຮົາໃສ່ຕົວເຊື່ອມຕໍ່:

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

ພວກ​ເຮົາ​ກວດ​ສອບ​ວ່າ​ການ​ດາວ​ໂຫຼດ​ສໍາ​ເລັດ​ຜົນ​ແລະ​ຕົວ​ເຊື່ອມ​ຕໍ່​ໄດ້​ເລີ່ມ​ຕົ້ນ​:

$ curl -i http://localhost:8083/connectors/pg-connector/status 
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)

{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}

ດີເລີດ: ມັນຕັ້ງແລ້ວ ແລະພ້ອມທີ່ຈະໄປ. ຕອນນີ້ໃຫ້ສົມມຸດວ່າເປັນຜູ້ບໍລິໂພກແລະເຊື່ອມຕໍ່ກັບ Kafka, ຫຼັງຈາກນັ້ນພວກເຮົາເພີ່ມແລະປ່ຽນລາຍການໃນຕາຕະລາງ:

$ kafka/bin/kafka-console-consumer.sh 
  --bootstrap-server kafka:9092 
  --from-beginning 
  --property print.key=true 
  --topic data.cdc.dbname

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', '[email protected]');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

ໃນຫົວຂໍ້ຂອງພວກເຮົາ, ນີ້ຈະສະແດງດັ່ງຕໍ່ໄປນີ້:

JSON ຍາວຫຼາຍກັບການປ່ຽນແປງຂອງພວກເຮົາ

{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":null,
"after":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600374991648,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":602,
"lsn":34088472,
"xmin":null
},
"op":"c",
"ts_ms":1600374991762,
"transaction":null
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600375609365,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":603,
"lsn":34089688,
"xmin":null
},
"op":"u",
"ts_ms":1600375609778,
"transaction":null
}
}

ໃນທັງສອງກໍລະນີ, ບັນທຶກປະກອບດ້ວຍກຸນແຈ (PK) ຂອງບັນທຶກທີ່ມີການປ່ຽນແປງ, ແລະຄວາມສໍາຄັນຂອງການປ່ຽນແປງ: ສິ່ງທີ່ບັນທຶກກ່ອນແລະສິ່ງທີ່ມັນກາຍເປັນຫຼັງຈາກນັ້ນ.

• ໃນກໍລະນີຂອງ INSERT: ຄ່າ​ກ່ອນ (before) ເທົ່າກັບ nullຕິດຕາມດ້ວຍສາຍທີ່ໃສ່.
• ໃນກໍລະນີຂອງ UPDATE: ໃນ payload.before ສະຖານະທີ່ຜ່ານມາຂອງແຖວຖືກສະແດງ, ແລະໃນ payload.after - ໃຫມ່​ທີ່​ມີ​ເນື້ອ​ແທ້​ຂອງ​ການ​ປ່ຽນ​ແປງ​.

2.2 MongoDB

ຕົວເຊື່ອມຕໍ່ນີ້ໃຊ້ກົນໄກການຈໍາລອງແບບ MongoDB ມາດຕະຖານ, ອ່ານຂໍ້ມູນຈາກ oplog ຂອງ node ຫຼັກ DBMS.

ຄ້າຍຄືກັນກັບຕົວເຊື່ອມຕໍ່ທີ່ໄດ້ອະທິບາຍໄວ້ແລ້ວສໍາລັບ PgSQL, ທີ່ນີ້, ເຊັ່ນດຽວກັນ, ໃນຕອນເລີ່ມຕົ້ນທໍາອິດ, ພາບຖ່າຍຂໍ້ມູນຕົ້ນຕໍໄດ້ຖືກປະຕິບັດ, ຫຼັງຈາກນັ້ນຕົວເຊື່ອມຕໍ່ຈະປ່ຽນເປັນໂຫມດການອ່ານ oplog.

ຕົວຢ່າງການຕັ້ງຄ່າ:

{
"name": "mp-k8s-mongo-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max": "1",
"mongodb.hosts": "MainRepSet/mongo:27017",
"mongodb.name": "mongo",
"mongodb.user": "debezium",
"mongodb.password": "dbname",
"database.whitelist": "db_1,db_2",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
"transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
}
}

ດັ່ງທີ່ທ່ານສາມາດເຫັນໄດ້, ບໍ່ມີທາງເລືອກໃຫມ່ເມື່ອທຽບກັບຕົວຢ່າງທີ່ຜ່ານມາ, ແຕ່ວ່າພຽງແຕ່ຈໍານວນຂອງທາງເລືອກທີ່ຮັບຜິດຊອບໃນການເຊື່ອມຕໍ່ກັບຖານຂໍ້ມູນແລະຄໍານໍາຫນ້າຂອງເຂົາເຈົ້າໄດ້ຖືກຫຼຸດລົງ.

Settings transforms ເວລານີ້ພວກເຂົາເຮັດດັ່ງຕໍ່ໄປນີ້: ປ່ຽນຊື່ຂອງຫົວຂໍ້ເປົ້າຫມາຍຈາກໂຄງການ <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

ຄວາມ​ທົນ​ທານ​ຄວາມ​ຜິດ​ພາດ​

ບັນຫາຄວາມທົນທານຕໍ່ຄວາມຜິດແລະຄວາມພ້ອມສູງໃນເວລາຂອງພວກເຮົາແມ່ນມີຄວາມເຄັ່ງຕຶງຫຼາຍກ່ວາເກົ່າ - ໂດຍສະເພາະໃນເວລາທີ່ພວກເຮົາເວົ້າກ່ຽວກັບຂໍ້ມູນແລະການເຮັດທຸລະກໍາ, ແລະການຕິດຕາມການປ່ຽນແປງຂໍ້ມູນບໍ່ໄດ້ຢູ່ຂ້າງໃນເລື່ອງນີ້. ໃຫ້ພິຈາລະນາສິ່ງທີ່ສາມາດຜິດພາດໃນຫຼັກການແລະສິ່ງທີ່ຈະເກີດຂຶ້ນກັບ Debezium ໃນແຕ່ລະກໍລະນີ.

ມີສາມທາງເລືອກໃນການຖອນຕົວ:

1. Kafka Connect ຄວາມລົ້ມເຫຼວ. ຖ້າ Connect ຖືກຕັ້ງຄ່າໃຫ້ເຮັດວຽກຢູ່ໃນຮູບແບບການແຈກຢາຍ, ນີ້ຕ້ອງການພະນັກງານຫຼາຍຄົນເພື່ອຕັ້ງຄ່າ group.id ດຽວກັນ. ຫຼັງຈາກນັ້ນ, ຖ້າຫນຶ່ງຂອງພວກເຂົາລົ້ມເຫລວ, ຕົວເຊື່ອມຕໍ່ຈະຖືກເລີ່ມຕົ້ນໃຫມ່ໃນພະນັກງານອື່ນແລະສືບຕໍ່ອ່ານຈາກຕໍາແຫນ່ງທີ່ຫມັ້ນສັນຍາສຸດທ້າຍໃນຫົວຂໍ້ໃນ Kafka.
2. ການສູນເສຍການເຊື່ອມຕໍ່ກັບກຸ່ມ Kafka. ຕົວເຊື່ອມຕໍ່ພຽງແຕ່ຈະຢຸດການອ່ານຢູ່ໃນຕໍາແຫນ່ງທີ່ມັນລົ້ມເຫລວທີ່ຈະສົ່ງໄປຫາ Kafka ແລະພະຍາຍາມສົ່ງຄືນເປັນໄລຍະຈົນກ່ວາຄວາມພະຍາຍາມສໍາເລັດ.
3. ບໍ່ມີແຫຼ່ງຂໍ້ມູນ. ຕົວເຊື່ອມຕໍ່ຈະພະຍາຍາມເຊື່ອມຕໍ່ຄືນໃໝ່ກັບແຫຼ່ງຕາມການກຳນົດຄ່າ. ຄ່າເລີ່ມຕົ້ນແມ່ນ 16 ພະຍາຍາມໃຊ້ backoff exponential. ຫຼັງຈາກຄວາມພະຍາຍາມທີ່ລົ້ມເຫລວຄັ້ງທີ 16, ວຽກງານຈະຖືກຫມາຍເປັນ ລົ້ມເຫລວ ແລະມັນຈະຕ້ອງຖືກເລີ່ມຕົ້ນໃຫມ່ດ້ວຍຕົນເອງໂດຍຜ່ານການໂຕ້ຕອບ Kafka Connect REST.
   • ໃນກໍລະນີຂອງ PostgreSQL ຂໍ້ມູນຈະບໍ່ສູນເສຍ, ເພາະວ່າ ການ​ນໍາ​ໃຊ້​ສະ​ລັອດ​ຕິງ replication ຈະ​ປ້ອງ​ກັນ​ການ​ລຶບ​ໄຟລ​໌ WAL ທີ່​ບໍ່​ໄດ້​ອ່ານ​ໂດຍ​ຕົວ​ເຊື່ອມ​ຕໍ່​. ໃນກໍລະນີນີ້, ມີຂໍ້ເສຍຄື: ຖ້າການເຊື່ອມຕໍ່ເຄືອຂ່າຍລະຫວ່າງຕົວເຊື່ອມຕໍ່ແລະ DBMS ຖືກລົບກວນເປັນເວລາດົນນານ, ມີໂອກາດທີ່ພື້ນທີ່ດິດຈະຫມົດໄປ, ແລະນີ້ອາດຈະນໍາໄປສູ່ຄວາມລົ້ມເຫຼວຂອງ DBMS ທັງຫມົດ.
   • ໃນກໍລະນີຂອງ MySQL ໄຟລ໌ binlog ສາມາດຫມຸນໄດ້ໂດຍ DBMS ຕົວມັນເອງກ່ອນທີ່ການເຊື່ອມຕໍ່ຈະຖືກຟື້ນຟູ. ນີ້ຈະເຮັດໃຫ້ຕົວເຊື່ອມຕໍ່ເຂົ້າໄປໃນສະຖານະທີ່ລົ້ມເຫລວ, ແລະມັນຈະຕ້ອງປິດເປີດໃຫມ່ໃນໂຫມດ snapshot ເບື້ອງຕົ້ນເພື່ອສືບຕໍ່ການອ່ານຈາກ binlogs ເພື່ອຟື້ນຟູການເຮັດວຽກປົກກະຕິ.
   • ກ່ຽວກັບ MongoDB. ເອກະສານກ່າວວ່າ: ພຶດຕິກໍາຂອງຕົວເຊື່ອມຕໍ່ໃນກໍລະນີທີ່ໄຟລ໌ log / oplog ໄດ້ຖືກລຶບຖິ້ມແລະຕົວເຊື່ອມຕໍ່ບໍ່ສາມາດສືບຕໍ່ອ່ານຈາກຕໍາແຫນ່ງທີ່ມັນປະໄວ້ແມ່ນຄືກັນສໍາລັບ DBMS ທັງຫມົດ. ມັນຢູ່ໃນຄວາມຈິງທີ່ວ່າຕົວເຊື່ອມຕໍ່ຈະເຂົ້າໄປໃນລັດ ລົ້ມເຫລວ ແລະຈະຕ້ອງມີການປິດເປີດໃໝ່ໃນໂໝດ ພາບຖ່າຍເບື້ອງຕົ້ນ.

     ຢ່າງໃດກໍຕາມ, ມີຂໍ້ຍົກເວັ້ນ. ຖ້າຕົວເຊື່ອມຕໍ່ຢູ່ໃນສະຖານະຕັດການເຊື່ອມຕໍ່ເປັນເວລາດົນນານ (ຫຼືບໍ່ສາມາດເຂົ້າຫາຕົວຢ່າງ MongoDB), ແລະ oplog ໄດ້ຖືກຫມຸນໃນຊ່ວງເວລານີ້, ເມື່ອການເຊື່ອມຕໍ່ຖືກຟື້ນຟູຄືນໃຫມ່, ຕົວເຊື່ອມຕໍ່ຈະສືບຕໍ່ອ່ານຂໍ້ມູນຈາກຕໍາແຫນ່ງທໍາອິດທີ່ມີຢູ່. , ນັ້ນແມ່ນເຫດຜົນທີ່ບາງຂໍ້ມູນໃນ Kafka ບໍ່ ຈະຕີ.

ສະຫລຸບ

Debezium ແມ່ນປະສົບການທໍາອິດຂອງຂ້ອຍກັບລະບົບ CDC ແລະໄດ້ຮັບຜົນດີໂດຍລວມ. ໂຄງ​ການ​ດັ່ງ​ກ່າວ​ໄດ້​ໃຫ້​ສິນ​ບົນ​ການ​ສະ​ຫນັບ​ສະ​ຫນູນ​ຂອງ DBMS ຕົ້ນ​ຕໍ​, ຄວາມ​ງ່າຍ​ຂອງ​ການ​ຕັ້ງ​ຄ່າ​, ການ​ສະ​ຫນັບ​ສະ​ຫນູນ​ການ​ເປັນ​ກຸ່ມ​ແລະ​ການ​ເຄື່ອນ​ໄຫວ​ຂອງ​ຊຸມ​ຊົນ​. ສໍາລັບຜູ້ທີ່ສົນໃຈໃນການປະຕິບັດ, ຂ້າພະເຈົ້າແນະນໍາໃຫ້ທ່ານອ່ານຄູ່ມືສໍາລັບ Kafka ເຊື່ອມຕໍ່ и ເດເບຊຽມ.

ເມື່ອປຽບທຽບກັບຕົວເຊື່ອມຕໍ່ JDBC ສໍາລັບ Kafka Connect, ປະໂຫຍດຕົ້ນຕໍຂອງ Debezium ແມ່ນວ່າການປ່ຽນແປງໄດ້ຖືກອ່ານຈາກບັນທຶກ DBMS, ເຊິ່ງອະນຸຍາດໃຫ້ໄດ້ຮັບຂໍ້ມູນດ້ວຍຄວາມລ່າຊ້າຫນ້ອຍທີ່ສຸດ. JDBC Connector (ສະຫນອງໃຫ້ໂດຍ Kafka Connect) ສອບຖາມຕາຕະລາງຕິດຕາມໃນໄລຍະເວລາຄົງທີ່ແລະ (ສໍາລັບເຫດຜົນດຽວກັນ) ບໍ່ສ້າງຂໍ້ຄວາມເມື່ອຂໍ້ມູນຖືກລຶບ (ທ່ານສາມາດສອບຖາມຂໍ້ມູນທີ່ບໍ່ມີຢູ່ບ່ອນນັ້ນໄດ້ແນວໃດ?).

ເພື່ອແກ້ໄຂບັນຫາທີ່ຄ້າຍຄືກັນ, ທ່ານສາມາດເອົາໃຈໃສ່ກັບວິທີແກ້ໄຂຕໍ່ໄປນີ້ (ນອກເຫນືອຈາກ Debezium):

• JDBC Connector Kafka Connect
• ບາງວິທີແກ້ໄຂ MySQL ເທົ່ານັ້ນ:
  • MySQL Streamer;
  • Maxwell;
  • Spinal Tap;
• Oracle Golden Gate, ແຕ່ນີ້ແມ່ນ "ປະເພດນ້ໍາຫນັກ".

PS

ອ່ານຍັງຢູ່ໃນ blog ຂອງພວກເຮົາ:

• «ກໍານົດຂະຫນາດທີ່ເຫມາະສົມສໍາລັບກຸ່ມ Kafka ໃນ Kubernetes"
• «ເລື່ອງການປະຕິບັດຈາກຊີວິດປະຈໍາວັນຂອງພວກເຮົາ SRE. ພາກທີ 2"
• «ພາບລວມສັ້ນໆຂອງຄໍາຖະແຫຼງການ PostgreSQL ສໍາລັບ Kubernetes, ທາງເລືອກແລະປະສົບການຂອງພວກເຮົາ".

ແຫຼ່ງຂໍ້ມູນ: www.habr.com