წარმოგიდგენთ Debezium - CDC Apache Kafka-სთვის

ჩემს საქმიანობაში ხშირად ვხვდები ახალ ტექნიკურ გადაწყვეტილებებს/პროგრამულ პროდუქტებს, რომელთა შესახებ ინფორმაცია საკმაოდ მწირია რუსულენოვან ინტერნეტში. ამ სტატიით შევეცდები შევავსო ერთი ასეთი ხარვეზი ჩემი ბოლოდროინდელი პრაქტიკის მაგალითით, როდესაც მჭირდებოდა CDC მოვლენების გაგზავნის კონფიგურაცია ორი პოპულარული DBMS-დან (PostgreSQL და MongoDB) კაფკას კლასტერში Debezium-ის გამოყენებით. ვიმედოვნებ, რომ ეს მიმოხილვის სტატია, რომელიც გაწეული სამუშაოს შედეგად ჩანს, სასარგებლო იქნება სხვებისთვის.

რა არის Debezium და CDC ზოგადად?

დებეზიუმი - CDC პროგრამული უზრუნველყოფის კატეგორიის წარმომადგენელი (გადაიღეთ მონაცემთა ცვლილება), უფრო ზუსტად, ეს არის კონექტორების ნაკრები სხვადასხვა DBMS-ისთვის, რომელიც თავსებადია Apache Kafka Connect-ის ჩარჩოსთან.

იგი ღია კოდის პროექტი, ლიცენზირებულია Apache ლიცენზიით v2.0 და დაფინანსებულია Red Hat-ის მიერ. განვითარება მიმდინარეობს 2016 წლიდან და ამჟამად ის უზრუნველყოფს ოფიციალურ მხარდაჭერას შემდეგი DBMS-ებისთვის: MySQL, PostgreSQL, MongoDB, SQL Server. ასევე არის კონექტორები Cassandra-სა და Oracle-სთვის, მაგრამ ამ დროისთვის ისინი არიან „ადრეული წვდომის“ სტატუსში და ახალი გამოშვებები არ იძლევა გარანტიას ჩამორჩენილი თავსებადობის შესახებ.

თუ CDC-ს შევადარებთ ტრადიციულ მიდგომას (როდესაც აპლიკაცია პირდაპირ კითხულობს მონაცემებს DBMS-დან), მისი მთავარი უპირატესობები მოიცავს მონაცემთა ცვლილების ნაკადის განხორციელებას რიგის დონეზე დაბალი შეყოვნებით, მაღალი საიმედოობით და ხელმისაწვდომობით. ბოლო ორი წერტილი მიიღწევა კაფკას კლასტერის, როგორც CDC მოვლენების საცავში გამოყენებით.

კიდევ ერთი უპირატესობა არის ის ფაქტი, რომ ერთი მოდელი გამოიყენება მოვლენების შესანახად, ამიტომ საბოლოო აპლიკაცია არ უნდა ინერვიულოს სხვადასხვა DBMS-ების მუშაობის ნიუანსებზე.

და ბოლოს, შეტყობინებების ბროკერის გამოყენება საშუალებას აძლევს აპლიკაციებს, რომლებიც აკონტროლებენ მონაცემთა ცვლილებებს, ჰორიზონტალურად გაფართოვდეს. ამავდროულად, ზემოქმედება მონაცემთა წყაროზე მინიმუმამდეა დაყვანილი, ვინაიდან მონაცემები მიიღება არა უშუალოდ DBMS-დან, არამედ კაფკას კლასტერიდან.

დებეზიუმის არქიტექტურის შესახებ

Debezium-ის გამოყენება ამ მარტივ სქემამდე მოდის:

DBMS (როგორც მონაცემთა წყარო) → კონექტორი Kafka Connect-ში → Apache Kafka → მომხმარებელი

საილუსტრაციოდ, აქ მოცემულია დიაგრამა პროექტის ვებსაიტიდან:

თუმცა, მე ნამდვილად არ მომწონს ეს სქემა, რადგან, როგორც ჩანს, მხოლოდ ნიჟარის კონექტორის გამოყენებაა შესაძლებელი.

სინამდვილეში, სიტუაცია განსხვავებულია: თქვენი მონაცემთა ტბის შევსება (ბოლო ბმული ზემოთ მოცემულ დიაგრამაში) ეს არ არის დებეზიუმის გამოყენების ერთადერთი გზა. Apache Kafka-ზე გაგზავნილი ღონისძიებები შეიძლება გამოყენებულ იქნას თქვენი აპლიკაციების მიერ სხვადასხვა სიტუაციების მოსაგვარებლად. Მაგალითად:

• არარელევანტური მონაცემების ამოღება ქეშიდან;
• შეტყობინებების გაგზავნა;
• საძიებო ინდექსის განახლებები;
• გარკვეული სახის აუდიტის ჟურნალი;
• ...

იმ შემთხვევაში, თუ თქვენ გაქვთ Java აპლიკაცია და არ არის საჭირო/შესაძლებლობა გამოიყენოს კაფკას კლასტერი, ასევე არსებობს მუშაობის შესაძლებლობა. ჩაშენებული კონექტორი. აშკარა უპირატესობა ის არის, რომ ის გამორიცხავს დამატებითი ინფრასტრუქტურის საჭიროებას (კონექტორისა და კაფკას სახით). თუმცა, ეს გადაწყვეტა მოძველებულია 1.1 ვერსიიდან და აღარ არის რეკომენდებული გამოსაყენებლად (მისი მხარდაჭერა შეიძლება მოიხსნას მომავალ გამოშვებებში).

ეს სტატია განიხილავს დეველოპერების მიერ რეკომენდებულ არქიტექტურას, რომელიც უზრუნველყოფს შეცდომების შემწყნარებლობას და მასშტაბურობას.

კონექტორის კონფიგურაცია

იმისათვის, რომ დავიწყოთ ცვლილებების თვალყურის დევნება ყველაზე მნიშვნელოვან მნიშვნელობაში - მონაცემებში - გვჭირდება:

1. მონაცემთა წყარო, რომელიც შეიძლება იყოს MySQL 5.7 ვერსიიდან დაწყებული, PostgreSQL 9.6+, MongoDB 3.2+ (სრული სია);
2. Apache Kafka კასეტური;
3. Kafka Connect ინსტანცია (ვერსიები 1.x, 2.x);
4. კონფიგურირებული Debezium კონექტორი.

იმუშავეთ პირველ ორ პუნქტზე, ე.ი. DBMS-ისა და Apache Kafka-ს ინსტალაციის პროცესი სცილდება სტატიის ფარგლებს. თუმცა, მათთვის, ვისაც სურს ყველაფრის განლაგება ქვიშის ყუთში, ოფიციალურ საცავს მაგალითებით აქვს მზა დოკერ-შედგენა.yaml.

ბოლო ორ პუნქტზე უფრო დეტალურად ვისაუბრებთ.

0. კაფკა კონექტი

აქ და შემდგომ სტატიაში, ყველა კონფიგურაციის მაგალითი განიხილება Debezium-ის დეველოპერების მიერ გავრცელებულ Docker გამოსახულების კონტექსტში. ის შეიცავს ყველა საჭირო დანამატის ფაილს (კონექტორებს) და უზრუნველყოფს Kafka Connect-ის კონფიგურაციას გარემოს ცვლადების გამოყენებით.

თუ თქვენ აპირებთ Kafka Connect-ის გამოყენებას Confluent-დან, თქვენ მოგიწევთ დამოუკიდებლად დაამატოთ საჭირო კონექტორების დანამატები მითითებულ დირექტორიაში. plugin.path ან დააყენეთ გარემოს ცვლადის მეშვეობით CLASSPATH. Kafka Connect მუშაკისა და კონექტორების პარამეტრები განისაზღვრება კონფიგურაციის ფაილებით, რომლებიც არგუმენტებად გადაეცემა მუშაკის გაშვების ბრძანებას. დაწვრილებით იხ დოკუმენტაცია.

Debeizum-ის კონექტორის ვერსიაში დაყენების მთელი პროცესი ორ ეტაპად მიმდინარეობს. მოდით შევხედოთ თითოეულ მათგანს:

1. Kafka Connect-ის ჩარჩოს დაყენება

Apache Kafka კლასტერში მონაცემების გადასატანად, კაფკა კონექტის ჩარჩოში დაყენებულია კონკრეტული პარამეტრები, როგორიცაა:

• კლასტერთან დაკავშირების პარამეტრები,
• თემების სახელები, რომლებშიც უშუალოდ შეინახება თავად კონექტორის კონფიგურაცია,
• ჯგუფის სახელი, რომელშიც მუშაობს კონექტორი (თუ გამოიყენება განაწილებული რეჟიმი).

პროექტის ოფიციალური Docker სურათი მხარს უჭერს კონფიგურაციას გარემოს ცვლადების გამოყენებით - ეს არის ის, რასაც ჩვენ გამოვიყენებთ. ასე რომ, გადმოწერეთ სურათი:

docker pull debezium/connect

კონექტორის გასაშვებად საჭირო გარემო ცვლადების მინიმალური ნაკრები შემდეგია:

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 — კაფკას კლასტერული სერვერების საწყისი სია კლასტერის წევრების სრული სიის მისაღებად;
• OFFSET_STORAGE_TOPIC=connector-offsets — თემა იმ პოზიციების შესანახად, სადაც ამჟამად მდებარეობს კონექტორი;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status — კონექტორის სტატუსისა და მისი ამოცანების შესანახი თემა;
• CONFIG_STORAGE_TOPIC=connector-config — კონექტორის კონფიგურაციის მონაცემებისა და მისი ამოცანების შესანახი თემა;
• GROUP_ID=1 — მუშათა ჯგუფის იდენტიფიკატორი, რომელზედაც შესაძლებელია შემაერთებელი დავალების შესრულება; აუცილებელია განაწილებული გამოყენებისას (განაწილებული) რეჟიმი.

ჩვენ ვხსნით კონტეინერს ამ ცვლადებით:

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

შენიშვნა ავროს შესახებ

ნაგულისხმევად, Debezium წერს მონაცემებს JSON ფორმატში, რაც მისაღებია ქვიშის ყუთებისთვის და მცირე რაოდენობის მონაცემებისთვის, მაგრამ შეიძლება გახდეს პრობლემა ძალიან დატვირთულ მონაცემთა ბაზებში. JSON კონვერტორის ალტერნატივა არის შეტყობინებების სერიული გამოყენება Avro ორობით ფორმატში, რომელიც ამცირებს დატვირთვას I/O ქვესისტემაზე Apache Kafka-ში.

Avro-ს გამოსაყენებლად, თქვენ უნდა განათავსოთ ცალკე სქემა-რეესტრი (დიაგრამების შესანახად). კონვერტორის ცვლადები ასე გამოიყურება:

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 ნოტაციაში და აიტვირთება 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"
  }
}

კონექტორის მუშაობის პრინციპი ამ დაყენების შემდეგ საკმაოდ მარტივია:

• პირველად გაშვებისას, ის უერთდება კონფიგურაციაში მითითებულ მონაცემთა ბაზას და იწყება რეჟიმში საწყისი ფოტოსურათი, კაფკას უგზავნის პირობითის გამოყენებით მიღებული მონაცემების საწყისი ნაკრების SELECT * FROM table_name.
• ინიციალიზაციის დასრულების შემდეგ, კონექტორი გადადის რეჟიმში, რათა წაიკითხოს ცვლილებები PostgreSQL WAL ფაილებიდან.

გამოყენებული ვარიანტების შესახებ:

• name - კონექტორის სახელი, რომლისთვისაც გამოიყენება ქვემოთ აღწერილი კონფიგურაცია; მომავალში, ეს სახელი გამოიყენება კონექტორთან მუშაობისთვის (ანუ სტატუსის ნახვა/გადატვირთვა/კონფიგურაციის განახლება) Kafka Connect REST API-ის მეშვეობით;
• connector.class — DBMS კონექტორის კლასი, რომელსაც გამოიყენებს კონფიგურირებული კონექტორი;
• plugin.name — მოდულის სახელი WAL ფაილებიდან მონაცემების ლოგიკური გაშიფვრისთვის. ხელმისაწვდომია ასარჩევად wal2json, decoderbuffs и pgoutput. პირველი ორი მოითხოვს DBMS-ში შესაბამისი გაფართოებების დაყენებას და pgoutput PostgreSQL 10 და უფრო მაღალი ვერსიისთვის არ საჭიროებს დამატებით მანიპულაციებს;
• database.* — მონაცემთა ბაზასთან დაკავშირების ვარიანტები, სადაც database.server.name — PostgreSQL ინსტანციის სახელი, რომელიც გამოიყენება კაფკას კლასტერში თემის სახელის შესაქმნელად;
• table.include.list — ცხრილების სია, რომლებშიც გვინდა თვალყური ადევნოთ ცვლილებებს; ფორმატში მითითებული schema.table_name; არ შეიძლება გამოყენებულ იქნას ერთად table.exclude.list;
• heartbeat.interval.ms — ინტერვალი (მილიწამებში), რომლითაც კონექტორი აგზავნის გულისცემის შეტყობინებებს სპეციალურ თემაზე;
• heartbeat.action.query — მოთხოვნა, რომელიც შესრულდება ყოველი გულისცემის შეტყობინების გაგზავნისას (ოფცია გამოჩნდა 1.1 ვერსიაში);
• slot.name — რეპლიკაციის სლოტის სახელი, რომელსაც გამოიყენებს კონექტორი;
• publication.name - სახელი განთავსების PostgreSQL-ში, რომელსაც კონექტორი იყენებს. თუ ის არ არსებობს, Debezium შეეცდება მის შექმნას. თუ მომხმარებელს, რომლის ქვეშაც ხდება კავშირი, არ აქვს საკმარისი უფლებები ამ მოქმედებისთვის, კონექტორი შეცდომით შეწყდება;
• transforms ზუსტად განსაზღვრავს, თუ როგორ უნდა შეიცვალოს სამიზნე თემის სახელი:
  • transforms.AddPrefix.type მიუთითებს, რომ ჩვენ გამოვიყენებთ რეგულარულ გამონათქვამებს;
  • transforms.AddPrefix.regex — ნიღაბი, რომელიც ხელახლა განსაზღვრავს სამიზნე თემის სახელს;
  • transforms.AddPrefix.replacement - პირდაპირ რას განვსაზღვრავთ.

მეტი გულისცემისა და გარდაქმნების შესახებ

ნაგულისხმევად, კონექტორი აგზავნის მონაცემებს კაფკას თითოეული ჩადენილი ტრანზაქციისთვის და მისი LSN (Log Sequence Number) ჩაიწერება სერვისის თემაში. offset. მაგრამ რა მოხდება, თუ კონექტორი კონფიგურირებულია ისე, რომ წაიკითხოს არა მთელი მონაცემთა ბაზა, არამედ მისი ცხრილების მხოლოდ ნაწილი (რომელშიც მონაცემთა განახლებები ხშირად არ ხდება)?

• კონექტორი წაიკითხავს WAL ფაილებს და არ აღმოაჩენს ტრანზაქციის ვალდებულებებს იმ ცხრილებთან, რომლებსაც აკონტროლებს.
• ამიტომ, ის არ განაახლებს თავის ამჟამინდელ პოზიციას არც თემაში და არც რეპლიკაციის სლოტში.
• ეს, თავის მხრივ, გამოიწვევს WAL ფაილების შენარჩუნებას დისკზე და სავარაუდოდ ამოიწურება დისკზე.

და სწორედ აქ მოდის ვარიანტები სამაშველოში. heartbeat.interval.ms и heartbeat.action.query. ამ ოფციების წყვილებში გამოყენება შესაძლებელს ხდის გულისცემის შეტყობინების გაგზავნის ყოველ ჯერზე მონაცემების შეცვლის მოთხოვნის შესრულებას ცალკე ცხრილში. ამრიგად, LSN, რომელზეც ამჟამად მდებარეობს კონექტორი (რეპლიკაციის სლოტში) მუდმივად განახლდება. ეს საშუალებას აძლევს DBMS-ს წაშალოს WAL ფაილები, რომლებიც აღარ არის საჭირო. თქვენ შეგიძლიათ გაიგოთ მეტი იმის შესახებ, თუ როგორ მუშაობს პარამეტრები დოკუმენტაცია.

კიდევ ერთი ვარიანტი, რომელიც ყურადღების ღირსია transforms. თუმცა ეს უფრო მოხერხებულობასა და სილამაზეს ეხება...

ნაგულისხმევად, Debezium ქმნის თემებს შემდეგი დასახელების პოლიტიკის გამოყენებით: serverName.schemaName.tableName. ეს შეიძლება ყოველთვის არ იყოს მოსახერხებელი. Პარამეტრები transforms თქვენ შეგიძლიათ გამოიყენოთ რეგულარული გამონათქვამები ცხრილების სიის დასადგენად, მოვლენები, რომლიდანაც საჭიროა კონკრეტული სახელის თემაში გადატანა.

ჩვენს კონფიგურაციაში მადლობა transforms ხდება შემდეგი: ყველა CDC მოვლენა მონიტორინგის ქვეშ მყოფი მონაცემთა ბაზიდან გადავა თემაში სახელწოდებით data.cdc.dbname. წინააღმდეგ შემთხვევაში (ამ პარამეტრების გარეშე), Debezium ნაგულისხმევად შექმნის თემას თითოეული ცხრილისთვის, როგორიცაა: pg-dev.public.<table_name>.

კონექტორის შეზღუდვები

PostgreSQL-ის კონექტორის კონფიგურაციის აღწერილობის დასასრულებლად, ღირს საუბარი მისი მუშაობის შემდეგ მახასიათებლებზე/შეზღუდვებზე:

1. კონექტორის ფუნქციონირება PostgreSQL-სთვის ეყრდნობა ლოგიკური დეკოდირების კონცეფციას. ამიტომ ის არ აკონტროლებს მოთხოვნებს მონაცემთა ბაზის სტრუქტურის შესაცვლელად (DDL) - შესაბამისად, ეს მონაცემები არ იქნება თემებში.
2. იმის გამო, რომ რეპლიკაციის სლოტები გამოიყენება, კონექტორის დაკავშირება შესაძლებელია მხოლოდ წამყვანი DBMS ინსტანციისთვის.
3. თუ მომხმარებელს, რომელსაც კონექტორი უკავშირდება მონაცემთა ბაზას, აქვს მხოლოდ წაკითხვის უფლებები, მაშინ პირველ გაშვებამდე მოგიწევთ ხელით შექმნათ რეპლიკაციის სლოტი და გამოაქვეყნოთ მონაცემთა ბაზაში.

კონფიგურაციის გამოყენება

მაშ ასე, ჩავტვირთოთ ჩვენი კონფიგურაცია კონექტორში:

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

ჩვენ ვამოწმებთ, რომ ჩამოტვირთვა წარმატებით დასრულდა და კონექტორი დაიწყო:

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

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

შესანიშნავია: დაყენებულია და მზადაა წასასვლელად. ახლა მოდით ვიყოთ მომხმარებელი და დავუკავშირდეთ კაფკას, რის შემდეგაც დავამატებთ და შევცვლით ჩანაწერს ცხრილში:

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

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', '[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 ნაჩვენებია ხაზის წინა მდგომარეობა და in payload.after — ახალი ცვლილებების არსით.

2.2 MongoDB

ეს კონექტორი იყენებს სტანდარტულ MongoDB რეპლიკაციის მექანიზმს, კითხულობს ინფორმაციას პირველადი DBMS კვანძის ოპლოგიდან.

PgSQL-სთვის უკვე აღწერილი კონექტორის მსგავსად, აქაც, პირველი დაწყებისას, ხდება პირველადი მონაცემების სურათის გადაღება, რის შემდეგაც კონექტორი გადადის ოპლოგის კითხვის რეჟიმში.

კონფიგურაციის მაგალითი:

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

როგორც ხედავთ, წინა მაგალითთან შედარებით აქ ახალი ვარიანტები არ არის, მაგრამ მხოლოდ მონაცემთა ბაზასთან დაკავშირებაზე პასუხისმგებელი ვარიანტების რაოდენობა და მათი პრეფიქსები შემცირდა.

პარამეტრები transforms ამჯერად ისინი აკეთებენ შემდეგს: გარდაქმნიან სამიზნე თემის სახელს სქემიდან <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

შეცდომის ტოლერანტობა

შეცდომების ტოლერანტობის და მაღალი ხელმისაწვდომობის საკითხი ჩვენს დროში უფრო მწვავეა, ვიდრე ოდესმე - განსაკუთრებით მაშინ, როდესაც ვსაუბრობთ მონაცემებსა და ტრანზაქციებზე, და მონაცემთა ცვლილებების თვალყურის დევნება ამ საკითხში არ დგას განზე. მოდი ვნახოთ, რა შეიძლება მოხდეს პრინციპში არასწორად და რა მოუვა დებეზიუმს თითოეულ შემთხვევაში.

არჩევის სამი ვარიანტი არსებობს:

1. Kafka Connect-ის მარცხი. თუ Connect კონფიგურირებულია განაწილებულ რეჟიმში მუშაობისთვის, ეს მოითხოვს რამდენიმე მუშაკს, რათა დააყენონ იგივე group.id. შემდეგ, თუ ერთ-ერთი მათგანი ვერ მოხერხდა, კონექტორი ხელახლა ჩაირთვება სხვა მუშაკზე და გააგრძელებს კითხვას კაფკას თემის ბოლო ჩადებული პოზიციიდან.
2. კაფკას კლასტერთან კავშირის დაკარგვა. კონექტორი უბრალოდ შეწყვეტს კითხვას იმ პოზიციაზე, რომელიც ვერ გაუგზავნა კაფკას და პერიოდულად შეეცდება მის ხელახლა გაგზავნას, სანამ მცდელობა არ დასრულდება.
3. მონაცემთა წყარო მიუწვდომელია. კონექტორი შეეცდება ხელახლა დაუკავშირდეს წყაროს, როგორც კონფიგურირებულია. ნაგულისხმევი არის 16 მცდელობა გამოყენებით ექსპონენციური უკან დახევა. მე-16 წარუმატებელი მცდელობის შემდეგ დავალება მოინიშნება როგორც ჩაიშალა და თქვენ დაგჭირდებათ მისი ხელით გადატვირთვა Kafka Connect REST ინტერფეისის მეშვეობით.
   • იმ შემთხვევაში PostgreSQL მონაცემები არ დაიკარგება, რადგან რეპლიკაციის სლოტების გამოყენება თავიდან აგაცილებთ WAL ფაილების წაშლას, რომლებიც არ იკითხება კონექტორის მიერ. ამ შემთხვევაში, მონეტის უარყოფითი მხარეც არის: თუ კონექტორსა და DBMS-ს შორის ქსელის კავშირი დიდი ხნის განმავლობაში შეფერხებულია, არსებობს შესაძლებლობა, რომ დისკზე ადგილი ამოიწუროს და ამან შეიძლება გამოიწვიოს მარცხი. მთელი DBMS.
   • იმ შემთხვევაში MySQL binlog ფაილების როტაცია შესაძლებელია თავად DBMS-ით, სანამ კავშირი აღდგება. ეს გამოიწვევს კონექტორის წარუმატებელ მდგომარეობაში გადასვლას და ნორმალური მუშაობის აღსადგენად, თქვენ უნდა გადატვირთოთ საწყისი სნეპშოტის რეჟიმში, რათა გააგრძელოთ კითხვა ბილოგებიდან.
   • on MongoDB. დოკუმენტაციაში ნათქვამია: კონექტორის ქცევა იმ შემთხვევაში, თუ log/oplog ფაილები წაიშლება და კონექტორი ვერ აგრძელებს კითხვას იმ პოზიციიდან, სადაც შეჩერდა, იგივეა ყველა DBMS-ისთვის. ეს ნიშნავს, რომ კონექტორი გადავა მდგომარეობაში ჩაიშალა და საჭიროებს რეჟიმში გადატვირთვას საწყისი ფოტოსურათი.

     თუმცა არის გამონაკლისებიც. თუ კონექტორი დიდი ხნის განმავლობაში გათიშული იყო (ან ვერ მიაღწია MongoDB მაგალითს) და ამ დროის განმავლობაში ოპლოგი გადიოდა ბრუნვას, მაშინ როდესაც კავშირი აღდგება, კონექტორი მშვიდად გააგრძელებს მონაცემების წაკითხვას პირველი ხელმისაწვდომი პოზიციიდან. რის გამოც ზოგიერთი მონაცემი კაფკაში არარის დაარტყამს.

დასკვნა

Debezium არის ჩემი პირველი გამოცდილება CDC სისტემებთან და მთლიანობაში ძალიან დადებითი. პროექტმა გაიმარჯვა ძირითადი DBMS-ების მხარდაჭერით, კონფიგურაციის სიმარტივით, კლასტერული მხარდაჭერით და აქტიური საზოგადოებისთვის. მათთვის, ვინც დაინტერესებულია პრაქტიკით, გირჩევთ, წაიკითხოთ სახელმძღვანელო კაფკა კონექტი и დებეზიუმი.

Kafka Connect-ისთვის JDBC კონექტორთან შედარებით, Debezium-ის მთავარი უპირატესობა ის არის, რომ ცვლილებები იკითხება DBMS ჟურნალებიდან, რაც საშუალებას აძლევს მონაცემების მიღებას მინიმალური შეყოვნებით. JDBC Connector (კაფკა კონექტორიდან) სვამს მონიტორინგს ცხრილს ფიქსირებული ინტერვალით და (იმავე მიზეზის გამო) არ ქმნის შეტყობინებებს მონაცემების წაშლისას (როგორ შეგიძლიათ მოითხოვოთ მონაცემები, რომლებიც არ არსებობს?).

მსგავსი პრობლემების გადასაჭრელად შეგიძლიათ ყურადღება მიაქციოთ შემდეგ გადაწყვეტილებებს (გარდა Debezium):

• JDBC კონექტორი Kafka Connect;
• რამდენიმე გამოსავალი მხოლოდ MySQL-სთვის:
  • MySQL Streamer;
  • მაქსველი;
  • SpinalTap;
• Oracle GoldenGate, მაგრამ ეს არის სრულიად განსხვავებული "წონის კატეგორია".

PS

ასევე წაიკითხეთ ჩვენს ბლოგზე:

• «კაფკას კლასტერის შესაბამისი ზომის განსაზღვრა კუბერნეტში";
• «პრაქტიკული ისტორიები ჩვენი SRE ყოველდღიური ცხოვრებიდან. Მე -2 ნაწილი";
• «PostgreSQL განცხადებების მოკლე მიმოხილვა Kubernetes-ისთვის, ჩვენი არჩევანი და გამოცდილება".

წყარო: www.habr.com