🥇अपाचे काफ्कासाठी Debezium - CDC सादर करत आहे

माझ्या कामात, मला अनेकदा नवीन तांत्रिक उपाय/सॉफ्टवेअर उत्पादने आढळतात, ज्याची माहिती रशियन-भाषेच्या इंटरनेटवर फारच कमी आहे. या लेखाद्वारे, मी माझ्या अलीकडील सरावातील उदाहरणासह अशी एक पोकळी भरून काढण्याचा प्रयत्न करेन, जेव्हा मला दोन लोकप्रिय DBMSs (PostgreSQL आणि MongoDB) मधून Debezium वापरून काफ्का क्लस्टरवर CDC इव्हेंट पाठवणे सेट करायचे होते. मला आशा आहे की हा पुनरावलोकन लेख, केलेल्या कामाच्या परिणामी दिसला, इतरांसाठी उपयुक्त ठरेल.

सर्वसाधारणपणे डेबेझियम आणि सीडीसी म्हणजे काय?

डेबेझियम - सीडीसी सॉफ्टवेअर श्रेणीचे प्रतिनिधी (डेटा बदल कॅप्चर करा), किंवा अधिक स्पष्टपणे, हा Apache Kafka Connect फ्रेमवर्कशी सुसंगत विविध DBMS साठी कनेक्टरचा संच आहे.

या मुक्त स्रोत प्रकल्प, Apache License v2.0 अंतर्गत परवानाकृत आणि Red Hat द्वारे प्रायोजित. 2016 पासून विकास चालू आहे आणि सध्या ते खालील DBMS साठी अधिकृत समर्थन प्रदान करते: MySQL, PostgreSQL, MongoDB, SQL Server. Cassandra आणि Oracle साठी कनेक्टर देखील आहेत, परंतु याक्षणी ते "लवकर प्रवेश" स्थितीत आहेत आणि नवीन रिलीझ मागास अनुकूलतेची हमी देत नाहीत.

जर आपण CDC ची तुलना पारंपारिक दृष्टिकोनाशी केली (जेव्हा अनुप्रयोग थेट DBMS वरून डेटा वाचतो), तर त्याच्या मुख्य फायद्यांमध्ये कमी विलंबता, उच्च विश्वासार्हता आणि उपलब्धतेसह पंक्ती स्तरावर डेटा बदल प्रवाहाची अंमलबजावणी समाविष्ट आहे. सीडीसी इव्हेंट्ससाठी भांडार म्हणून काफ्का क्लस्टर वापरून शेवटचे दोन गुण प्राप्त केले जातात.

आणखी एक फायदा हा आहे की एकच मॉडेल इव्हेंट्स संचयित करण्यासाठी वापरला जातो, त्यामुळे शेवटच्या ऍप्लिकेशनला वेगवेगळ्या डीबीएमएस ऑपरेट करण्याच्या बारकावेबद्दल काळजी करण्याची गरज नाही.

शेवटी, मेसेज ब्रोकर वापरून डेटामधील बदलांचे परीक्षण करणार्‍या ऍप्लिकेशन्सना क्षैतिजरित्या स्केल आउट करण्याची अनुमती मिळते. त्याच वेळी, डेटा स्त्रोतावरील प्रभाव कमी केला जातो, कारण डेटा थेट DBMS कडून प्राप्त होत नाही, परंतु काफ्का क्लस्टरकडून प्राप्त होतो.

डेबेझियम आर्किटेक्चर बद्दल

डेबेझियम वापरणे या सोप्या योजनेत येते:

DBMS (डेटा स्रोत म्हणून) → काफ्का कनेक्ट → अपाचे काफ्का → ग्राहक मधील कनेक्टर

एक उदाहरण म्हणून, येथे प्रकल्प वेबसाइटवरील एक आकृती आहे:

तथापि, मला ही योजना खरोखर आवडत नाही, कारण असे दिसते की केवळ सिंक कनेक्टरचा वापर शक्य आहे.

प्रत्यक्षात, परिस्थिती वेगळी आहे: तुमचा डेटा तलाव भरणे (वरील चित्रातील शेवटची लिंक) डेबेझियम वापरण्याचा हा एकमेव मार्ग नाही. Apache Kafka ला पाठवलेले इव्हेंट तुमच्या ऍप्लिकेशन्सद्वारे विविध परिस्थिती हाताळण्यासाठी वापरले जाऊ शकतात. उदाहरणार्थ:

कॅशेमधून असंबद्ध डेटा काढून टाकणे;
सूचना पाठवणे;
अनुक्रमणिका अद्यतने शोधा;
काही प्रकारचे ऑडिट लॉग;
...

जर तुमच्याकडे जावा ऍप्लिकेशन असेल आणि काफ्का क्लस्टर वापरण्याची गरज/संभाव्यता नसेल, तर त्याद्वारे काम करण्याची देखील शक्यता आहे एम्बेडेड-कनेक्टर. स्पष्ट प्लस म्हणजे त्यासह आपण अतिरिक्त पायाभूत सुविधा (कनेक्टर आणि काफ्काच्या स्वरूपात) नाकारू शकता. तथापि, हे समाधान आवृत्ती 1.1 पासून नापसंत केले गेले आहे आणि यापुढे वापरण्यासाठी शिफारस केलेली नाही (ते भविष्यातील प्रकाशनांमध्ये काढले जाऊ शकते).

हा लेख विकासकांनी शिफारस केलेल्या आर्किटेक्चरची चर्चा करेल, जे दोष सहिष्णुता आणि स्केलेबिलिटी प्रदान करते.

कनेक्टर कॉन्फिगरेशन

सर्वात महत्वाच्या मूल्यातील बदलांचा मागोवा घेणे सुरू करण्यासाठी - डेटा - आम्हाला आवश्यक आहे:

डेटा स्रोत, जो आवृत्ती 5.7, PostgreSQL 9.6+, MongoDB 3.2+ पासून सुरू होणारा MySQL असू शकतो (संपूर्ण यादी);
अपाचे काफ्का क्लस्टर;
काफ्का कनेक्ट उदाहरण (आवृत्त्या 1.x, 2.x);
कॉन्फिगर केलेले डेबेझियम कनेक्टर.

पहिल्या दोन मुद्यांवर काम करा, म्हणजे. DBMS आणि Apache Kafka ची स्थापना प्रक्रिया लेखाच्या व्याप्तीच्या पलीकडे आहे. तथापि, ज्यांना सँडबॉक्समध्ये सर्व काही उपयोजित करायचे आहे, त्यांच्यासाठी उदाहरणांसह अधिकृत भांडार तयार आहे docker-compose.yaml.

आम्ही शेवटच्या दोन मुद्द्यांवर अधिक तपशीलवार राहू.

0. काफ्का कनेक्ट

येथे आणि नंतर लेखात, सर्व कॉन्फिगरेशन उदाहरणे डेबेझियम डेव्हलपर्सद्वारे वितरित डॉकर प्रतिमेच्या संदर्भात विचारात घेतली जातात. यामध्ये सर्व आवश्यक प्लगइन फाइल्स (कनेक्टर) आहेत आणि पर्यावरण व्हेरिएबल्स वापरून काफ्का कनेक्ट कॉन्फिगरेशन प्रदान करते.

जर तुम्‍हाला Confluent वरून Kafka Connect वापरण्‍याचा इरादा असेल, तर तुम्‍हाला आवश्‍यक कनेक्‍टर्सचे प्लगइन स्‍वत: म्‍हणून निर्देशीत निर्देशिकेत जोडावे लागतील. plugin.path किंवा पर्यावरण व्हेरिएबलद्वारे सेट करा CLASSPATH. काफ्का कनेक्ट वर्कर आणि कनेक्टर्ससाठी सेटिंग्ज कॉन्फिगरेशन फाइल्सद्वारे निर्धारित केल्या जातात ज्या वर्कर लॉन्च कमांडला वितर्क म्हणून पास केल्या जातात. अधिक तपशीलांसाठी, पहा दस्तऐवजीकरण.

कनेक्टर आवृत्तीमध्ये डेबीझम सेट करण्याची संपूर्ण प्रक्रिया दोन टप्प्यांत केली जाते. चला त्या प्रत्येकाकडे पाहूया:

1. काफ्का कनेक्ट फ्रेमवर्क सेट करणे

अपाचे काफ्का क्लस्टरमध्ये डेटा प्रवाहित करण्यासाठी, काफ्का कनेक्ट फ्रेमवर्कमध्ये विशिष्ट पॅरामीटर्स सेट केले जातात, जसे की:

क्लस्टरला जोडण्यासाठी पॅरामीटर्स,
विषयांची नावे ज्यात कनेक्टरचे कॉन्फिगरेशन स्वतःच थेट संग्रहित केले जाईल,
गटाचे नाव ज्यामध्ये कनेक्टर चालू आहे (जर वितरित मोड वापरला असेल).

प्रकल्पाची अधिकृत डॉकर प्रतिमा पर्यावरण व्हेरिएबल्स वापरून कॉन्फिगरेशनला समर्थन देते - हे आम्ही वापरणार आहोत. तर, प्रतिमा डाउनलोड करा:

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

Avro बद्दल टीप

डीफॉल्टनुसार, डेबेझियम JSON फॉरमॅटमध्ये डेटा लिहितो, जो सँडबॉक्सेस आणि थोड्या प्रमाणात डेटासाठी स्वीकार्य आहे, परंतु उच्च लोड केलेल्या डेटाबेसमध्ये समस्या होऊ शकते. JSON कन्व्हर्टरचा पर्याय म्हणजे मेसेज वापरून क्रमवारी लावणे AVRO बायनरी फॉरमॅटमध्ये, जे अपाचे काफ्कामधील I/O उपप्रणालीवरील भार कमी करते.

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 नोटेशनमध्ये केले आहे आणि POST विनंती वापरून Kafka Connect वर अपलोड केले आहे.

२.१. 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 — कनेक्टरचे नाव ज्यासाठी खाली वर्णन केलेले कॉन्फिगरेशन वापरले आहे; भविष्यात, हे नाव काफ्का कनेक्ट REST API द्वारे कनेक्टरसह कार्य करण्यासाठी (म्हणजे स्थिती पहा / रीस्टार्ट / कॉन्फिगरेशन अद्यतनित करण्यासाठी) वापरले जाईल;
connector.class — डीबीएमएस कनेक्टर क्लास जो कॉन्फिगर केलेल्या कनेक्टरद्वारे वापरला जाईल;
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 (लॉग अनुक्रम क्रमांक) सेवा विषयामध्ये नोंदवला जातो. 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 साठी कनेक्टर कॉन्फिगरेशनचे वर्णन पूर्ण करण्यासाठी, त्याच्या ऑपरेशनच्या खालील वैशिष्ट्यांबद्दल/मर्यादांबद्दल बोलणे योग्य आहे:

PostgreSQL साठी कनेक्टरची कार्यक्षमता लॉजिकल डीकोडिंगच्या संकल्पनेवर अवलंबून असते. त्यामुळे तो डेटाबेस संरचना बदलण्यासाठी विनंत्यांचा मागोवा घेत नाही (DDL) - त्यानुसार, हा डेटा विषयांमध्ये नसेल.
प्रतिकृती स्लॉट वापरल्या जात असल्याने, कनेक्टर कनेक्ट करणे शक्य आहे फक्त अग्रगण्य DBMS उदाहरणासाठी.
ज्या वापरकर्त्याच्या अंतर्गत कनेक्टर डेटाबेसशी कनेक्ट करतो त्याला केवळ-वाचण्याचे अधिकार असल्यास, प्रथम लॉन्च करण्यापूर्वी आपल्याला व्यक्तिचलितपणे प्रतिकृती स्लॉट तयार करणे आणि डेटाबेसवर प्रकाशित करणे आवश्यक आहे.

कॉन्फिगरेशन लागू करत आहे

तर, आमचे कॉन्फिगरेशन कनेक्टरमध्ये लोड करूया:

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

दोन्ही प्रकरणांमध्ये, रेकॉर्डमध्ये बदललेल्या रेकॉर्डची की (पीके) आणि बदलांचे सार असते: रेकॉर्ड आधी काय होते आणि नंतर काय झाले.

च्या बाबतीत INSERT: आधी मूल्य (before) बरोबरी null, आणि नंतर - घातलेली ओळ.
च्या बाबतीत UPDATE: येथे payload.before ओळीची मागील स्थिती प्रदर्शित केली जाते आणि मध्ये payload.after — बदलांच्या सारासह नवीन.

2.2 MongoDB

हा कनेक्टर मानक मोंगोडीबी प्रतिकृती यंत्रणा वापरतो, प्राथमिक DBMS नोडच्या oplog वरून माहिती वाचतो.

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

जसे तुम्ही बघू शकता, मागील उदाहरणाच्या तुलनेत येथे कोणतेही नवीन पर्याय नाहीत, परंतु डेटाबेसशी कनेक्ट करण्यासाठी जबाबदार असलेल्या पर्यायांची संख्या आणि त्यांचे उपसर्ग कमी केले आहेत.

सेटिंग्ज transforms यावेळी ते पुढील गोष्टी करतात: ते लक्ष्य विषयाचे नाव स्कीमातून बदलतात <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

चुकीची सहनशीलता

आमच्या काळातील दोष सहिष्णुता आणि उच्च उपलब्धता ही समस्या नेहमीपेक्षा अधिक तीव्र आहे - विशेषत: जेव्हा आम्ही डेटा आणि व्यवहारांबद्दल बोलतो आणि डेटा बदल ट्रॅकिंग या प्रकरणात बाजूला नाही. तत्त्वतः काय चूक होऊ शकते आणि प्रत्येक बाबतीत डेबेझियमचे काय होईल ते पाहूया.

तीन निवड रद्द करण्याचे पर्याय आहेत:

काफ्का कनेक्ट अयशस्वी. Connect वितरित मोडमध्ये कार्य करण्यासाठी कॉन्फिगर केले असल्यास, यासाठी एकाधिक कामगारांना समान group.id सेट करणे आवश्यक आहे. त्यानंतर, त्यापैकी एक अयशस्वी झाल्यास, कनेक्टर दुसर्या कार्यकर्त्यावर रीस्टार्ट केला जाईल आणि काफ्कामधील विषयातील शेवटच्या वचनबद्ध स्थितीपासून वाचन सुरू ठेवेल.
काफ्का क्लस्टरशी संपर्क तुटला. कनेक्टर फक्त काफ्काला पाठवण्यात अयशस्वी झालेल्या स्थितीवर वाचन थांबवेल आणि प्रयत्न यशस्वी होईपर्यंत वेळोवेळी ते पुन्हा पाठवण्याचा प्रयत्न करेल.
डेटा स्रोत अनुपलब्धता. कनेक्टर कॉन्फिगर केल्याप्रमाणे स्त्रोताशी पुन्हा कनेक्ट करण्याचा प्रयत्न करेल. डीफॉल्ट वापरून 16 प्रयत्न आहे घातांकीय बॅकऑफ. 16 व्या अयशस्वी प्रयत्नानंतर, कार्य म्हणून चिन्हांकित केले जाईल अयशस्वी आणि तुम्हाला ते काफ्का कनेक्ट REST इंटरफेसद्वारे व्यक्तिचलितपणे रीस्टार्ट करावे लागेल.
- च्या बाबतीत पोस्टग्रे एसक्यूएल डेटा गमावला जाणार नाही, कारण प्रतिकृती स्लॉट वापरल्याने कनेक्टरद्वारे न वाचलेल्या WAL फाइल्स हटवण्यापासून तुम्हाला प्रतिबंध होईल. या प्रकरणात, नाण्याला एक नकारात्मक बाजू देखील आहे: जर कनेक्टर आणि डीबीएमएस दरम्यान नेटवर्क कनेक्टिव्हिटी बर्याच काळासाठी व्यत्यय आणली असेल, तर डिस्कची जागा संपण्याची शक्यता आहे आणि यामुळे अयशस्वी होऊ शकते. संपूर्ण DBMS.
- च्या बाबतीत , MySQL कनेक्टिव्हिटी पुनर्संचयित होण्यापूर्वी बिनलॉग फाइल्स डीबीएमएसद्वारेच फिरवल्या जाऊ शकतात. यामुळे कनेक्टर अयशस्वी स्थितीत जाईल आणि सामान्य ऑपरेशन पुनर्संचयित करण्यासाठी, तुम्हाला बिनलॉग्समधून वाचन सुरू ठेवण्यासाठी प्रारंभिक स्नॅपशॉट मोडमध्ये रीस्टार्ट करणे आवश्यक आहे.
- वर MongoDB. दस्तऐवजात नमूद केले आहे: लॉग/ऑपलॉग फाइल्स हटविल्या गेल्यास कनेक्टरचे वर्तन आणि कनेक्टर ज्या स्थितीतून सोडले आहे तिथून वाचणे सुरू ठेवू शकत नाही सर्व DBMS साठी सारखेच आहे. याचा अर्थ कनेक्टर राज्यात जाईल अयशस्वी आणि मोडमध्ये रीस्टार्ट करणे आवश्यक आहे प्रारंभिक स्नॅपशॉट.
  तथापि, अपवाद आहेत. जर कनेक्टर बराच काळ डिस्कनेक्ट अवस्थेत असेल (किंवा मोंगोडीबी उदाहरणापर्यंत पोहोचू शकला नाही), आणि या वेळी oplog फिरवला गेला असेल, तर कनेक्शन पुनर्संचयित केल्यावर, कनेक्टर शांतपणे पहिल्या उपलब्ध स्थितीतून डेटा वाचणे सुरू ठेवेल. , म्हणूनच काफ्कामधील काही डेटा नाही मारेल.

निष्कर्ष

डेबेझियम हा सीडीसी सिस्टीमचा माझा पहिला अनुभव आहे आणि एकूणच खूप सकारात्मक आहे. प्रमुख डीबीएमएस, कॉन्फिगरेशन सुलभता, क्लस्टरिंग समर्थन आणि सक्रिय समुदायासाठी त्याच्या समर्थनासह प्रकल्प जिंकला. सराव मध्ये स्वारस्य असलेल्यांसाठी, मी शिफारस करतो की आपण मार्गदर्शक वाचा काफ्का कनेक्ट и डेबेझियम.

Kafka Connect साठी JDBC कनेक्टरच्या तुलनेत, Debezium चा मुख्य फायदा हा आहे की DBMS लॉगमधून बदल वाचले जातात, जे कमीतकमी लेटन्सीसह डेटा प्राप्त करण्यास अनुमती देतात. JDBC कनेक्टर (काफ्का कनेक्ट कडून) एका निश्चित अंतराने निरीक्षण केलेल्या टेबलची चौकशी करतो आणि (त्याच कारणासाठी) जेव्हा डेटा हटवला जातो तेव्हा संदेश तयार करत नाही (अस्तित्वात नसलेल्या डेटाची तुम्ही चौकशी कशी करू शकता?).

तत्सम समस्यांचे निराकरण करण्यासाठी, आपण खालील उपायांकडे लक्ष देऊ शकता (डेबेझियम व्यतिरिक्त):

जेडीबीसी कनेक्टर काफ्का कनेक्ट;
फक्त MySQL साठी अनेक उपाय:
ओरॅकल गोल्डनगेट, परंतु ही पूर्णपणे भिन्न "वजन श्रेणी" आहे.

PS

आमच्या ब्लॉगवर देखील वाचा:

स्त्रोत: www.habr.com

Apache Kafka साठी Debezium - CDC सादर करत आहे