🥇अपाचे काफ्का के लिए डेबेज़ियम - सीडीसी का परिचय

अपने काम में, मुझे अक्सर नए तकनीकी समाधान/सॉफ़्टवेयर उत्पाद मिलते हैं, जिनके बारे में जानकारी रूसी भाषा के इंटरनेट पर दुर्लभ है। इस लेख के साथ मैं अपने हाल के अभ्यास से एक उदाहरण के साथ एक ऐसे अंतर को भरने की कोशिश करूंगा, जब मुझे डेबेज़ियम का उपयोग करके दो लोकप्रिय डीबीएमएस (पोस्टग्रेएसक्यूएल और मोंगोडीबी) से काफ्का क्लस्टर में सीडीसी ईवेंट भेजने को कॉन्फ़िगर करने की आवश्यकता थी। मुझे आशा है कि यह समीक्षा लेख, जो किए गए कार्यों के परिणामस्वरूप सामने आया है, दूसरों के लिए उपयोगी होगा।

डेबेज़ियम और सीडीसी सामान्यतः क्या है?

डेबेज़ियम - सीडीसी सॉफ्टवेयर श्रेणी का प्रतिनिधि (डेटा परिवर्तन कैप्चर करें), या अधिक सटीक रूप से, यह अपाचे काफ्का कनेक्ट ढांचे के साथ संगत विभिन्न डीबीएमएस के लिए कनेक्टर्स का एक सेट है।

यह ओपन सोर्स प्रोजेक्ट, अपाचे लाइसेंस v2.0 के तहत लाइसेंस प्राप्त और Red Hat द्वारा प्रायोजित। विकास 2016 से जारी है और वर्तमान में यह निम्नलिखित DBMS के लिए आधिकारिक समर्थन प्रदान करता है: MySQL, PostgreSQL, MongoDB, SQL सर्वर। कैसंड्रा और ओरेकल के लिए भी कनेक्टर हैं, लेकिन फिलहाल वे "प्रारंभिक पहुंच" स्थिति में हैं, और नई रिलीज़ बैकवर्ड संगतता की गारंटी नहीं देती हैं।

यदि हम सीडीसी की तुलना पारंपरिक दृष्टिकोण (जब एप्लिकेशन सीधे डीबीएमएस से डेटा पढ़ता है) से करते हैं, तो इसके मुख्य लाभों में कम विलंबता, उच्च विश्वसनीयता और उपलब्धता के साथ पंक्ति स्तर पर डेटा परिवर्तन स्ट्रीमिंग का कार्यान्वयन शामिल है। अंतिम दो अंक सीडीसी घटनाओं के भंडार के रूप में काफ्का क्लस्टर का उपयोग करके प्राप्त किए जाते हैं।

एक अन्य लाभ यह तथ्य है कि घटनाओं को संग्रहीत करने के लिए एक एकल मॉडल का उपयोग किया जाता है, इसलिए अंतिम एप्लिकेशन को विभिन्न डीबीएमएस के संचालन की बारीकियों के बारे में चिंता करने की ज़रूरत नहीं है।

अंत में, एक संदेश ब्रोकर का उपयोग उन अनुप्रयोगों को अनुमति देता है जो डेटा में परिवर्तनों की निगरानी क्षैतिज रूप से करते हैं। साथ ही, डेटा स्रोत पर प्रभाव कम हो जाता है, क्योंकि डेटा सीधे डीबीएमएस से नहीं, बल्कि काफ्का क्लस्टर से प्राप्त किया जाता है।

डेबेज़ियम वास्तुकला के बारे में

डेबेज़ियम का उपयोग इस सरल योजना से होता है:

डीबीएमएस (डेटा स्रोत के रूप में) → काफ्का कनेक्ट में कनेक्टर → अपाचे काफ्का → उपभोक्ता

उदाहरण के तौर पर, यहां प्रोजेक्ट वेबसाइट से एक चित्र दिया गया है:

हालाँकि, मुझे वास्तव में यह योजना पसंद नहीं है, क्योंकि ऐसा लगता है कि केवल सिंक कनेक्टर का उपयोग ही संभव है।

वास्तव में, स्थिति अलग है: अपना डेटा लेक भरना (उपरोक्त चित्र में अंतिम लिंक) डेबेज़ियम का उपयोग करने का यह एकमात्र तरीका नहीं है। अपाचे काफ्का को भेजे गए ईवेंट का उपयोग आपके एप्लिकेशन द्वारा विभिन्न स्थितियों को संभालने के लिए किया जा सकता है। उदाहरण के लिए:

कैश से अप्रासंगिक डेटा हटाना;
सूचनाएं भेजना;
खोज सूचकांक अद्यतन;
किसी प्रकार के ऑडिट लॉग;
...

यदि आपके पास जावा एप्लिकेशन है और काफ्का क्लस्टर का उपयोग करने की कोई आवश्यकता/संभावना नहीं है, तो इसके माध्यम से काम करने की भी संभावना है एंबेडेड-कनेक्टर. स्पष्ट लाभ यह है कि यह अतिरिक्त बुनियादी ढांचे (कनेक्टर और काफ्का के रूप में) की आवश्यकता को समाप्त कर देता है। हालाँकि, इस समाधान को संस्करण 1.1 के बाद से अप्रचलित कर दिया गया है और अब इसे उपयोग के लिए अनुशंसित नहीं किया गया है (भविष्य के रिलीज़ में इसके लिए समर्थन हटाया जा सकता है)।

यह आलेख डेवलपर्स द्वारा अनुशंसित आर्किटेक्चर पर चर्चा करेगा, जो दोष सहनशीलता और स्केलेबिलिटी प्रदान करता है।

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

सबसे महत्वपूर्ण मूल्य - डेटा - में परिवर्तनों को ट्रैक करना शुरू करने के लिए हमें चाहिए:

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

पहले दो बिंदुओं पर काम करें, यानी। डीबीएमएस और अपाचे काफ्का की स्थापना प्रक्रिया लेख के दायरे से परे है। हालाँकि, जो लोग सैंडबॉक्स में सब कुछ तैनात करना चाहते हैं, उनके लिए उदाहरणों के साथ आधिकारिक रिपॉजिटरी तैयार है docker-compose.yaml.

हम अंतिम दो बिंदुओं पर अधिक विस्तार से ध्यान देंगे।

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

यहां और लेख में आगे, डेबेज़ियम डेवलपर्स द्वारा वितरित डॉकर छवि के संदर्भ में सभी कॉन्फ़िगरेशन उदाहरणों पर चर्चा की गई है। इसमें सभी आवश्यक प्लगइन फ़ाइलें (कनेक्टर) शामिल हैं और पर्यावरण चर का उपयोग करके काफ्का कनेक्ट का कॉन्फ़िगरेशन प्रदान करता है।

यदि आप कॉन्फ्लुएंट से काफ्का कनेक्ट का उपयोग करने का इरादा रखते हैं, तो आपको निर्दिष्ट निर्देशिका में आवश्यक कनेक्टर्स के प्लगइन्स को स्वतंत्र रूप से जोड़ना होगा 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

एवरो के बारे में नोट

डिफ़ॉल्ट रूप से, डेबेज़ियम JSON प्रारूप में डेटा लिखता है, जो सैंडबॉक्स और छोटी मात्रा में डेटा के लिए स्वीकार्य है, लेकिन अत्यधिक लोड किए गए डेटाबेस में एक समस्या बन सकता है। JSON कनवर्टर का एक विकल्प संदेशों को क्रमबद्ध करना है एवरो एक बाइनरी प्रारूप में, जो अपाचे काफ्का में I/O सबसिस्टम पर लोड को कम करता है।

एवरो का उपयोग करने के लिए आपको एक अलग से तैनाती की आवश्यकता है स्कीमा-रजिस्ट्री (आरेख संग्रहीत करने के लिए)। कनवर्टर के लिए वेरिएबल इस तरह दिखेंगे:

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

एवरो का उपयोग करने और इसके लिए रजिस्ट्री स्थापित करने का विवरण इस आलेख के दायरे से परे है - आगे, स्पष्टता के लिए, हम JSON का उपयोग करेंगे।

2. कनेक्टर को स्वयं कॉन्फ़िगर करना

अब आप सीधे कनेक्टर के कॉन्फ़िगरेशन पर जा सकते हैं, जो स्रोत से डेटा पढ़ेगा।

आइए दो DBMS के लिए कनेक्टर्स का उदाहरण देखें: PostgreSQL और MongoDB, जिनमें मुझे अनुभव है और जिनमें अंतर हैं (यद्यपि छोटे, लेकिन कुछ मामलों में महत्वपूर्ण!)।

कॉन्फ़िगरेशन को JSON नोटेशन में वर्णित किया गया है और POST अनुरोध का उपयोग करके काफ्का कनेक्ट पर अपलोड किया गया है।

2.1. पोस्टग्रेएसक्यूएल

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 — कनेक्टर का नाम जिसके लिए नीचे वर्णित कॉन्फ़िगरेशन का उपयोग किया जाता है; भविष्य में, इस नाम का उपयोग काफ्का कनेक्ट रेस्ट एपीआई के माध्यम से कनेक्टर के साथ काम करने के लिए किया जाता है (यानी, स्थिति देखें/रीस्टार्ट/कॉन्फ़िगरेशन अपडेट करें);
connector.class - डीबीएमएस कनेक्टर वर्ग जिसका उपयोग कॉन्फ़िगर किए गए कनेक्टर द्वारा किया जाएगा;
plugin.name - वाल फ़ाइलों से डेटा की तार्किक डिकोडिंग के लिए प्लगइन का नाम। चुनने के लिए उपलब्ध है 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 में, जो कनेक्टर उपयोग करता है। यदि यह अस्तित्व में नहीं है, तो डेबेज़ियम इसे बनाने का प्रयास करेगा। यदि जिस उपयोगकर्ता के तहत कनेक्शन बनाया गया है, उसके पास इस क्रिया के लिए पर्याप्त अधिकार नहीं हैं, तो कनेक्टर एक त्रुटि के साथ समाप्त हो जाएगा;
transforms यह निर्धारित करता है कि लक्षित विषय का नाम कैसे बदला जाए:
- transforms.AddPrefix.type इंगित करता है कि हम नियमित अभिव्यक्तियों का उपयोग करेंगे;
- transforms.AddPrefix.regex - एक मुखौटा जो लक्ष्य विषय के नाम को फिर से परिभाषित करता है;
- transforms.AddPrefix.replacement - सीधे तौर पर हम क्या पुनर्परिभाषित कर रहे हैं।

दिल की धड़कन और परिवर्तनों के बारे में अधिक जानकारी

डिफ़ॉल्ट रूप से, कनेक्टर प्रत्येक प्रतिबद्ध लेनदेन के लिए काफ्का को डेटा भेजता है, और इसका एलएसएन (लॉग अनुक्रम संख्या) सेवा विषय में दर्ज किया जाता है offset. लेकिन क्या होता है यदि कनेक्टर को पूरे डेटाबेस को नहीं, बल्कि उसकी तालिकाओं के केवल एक हिस्से को पढ़ने के लिए कॉन्फ़िगर किया गया है (जिसमें डेटा अपडेट अक्सर नहीं होते हैं)?

कनेक्टर वाल फ़ाइलों को पढ़ेगा और जिन तालिकाओं की वह निगरानी कर रहा है, उनमें किसी भी लेन-देन का पता नहीं लगाएगा।
इसलिए, यह विषय या प्रतिकृति स्लॉट में अपनी वर्तमान स्थिति को अपडेट नहीं करेगा।
इसके परिणामस्वरूप, वाल फ़ाइलें डिस्क पर रखी जाएंगी और संभवतः डिस्क स्थान से बाहर हो जाएंगी।

और यहीं पर विकल्प बचाव के लिए आते हैं। heartbeat.interval.ms и heartbeat.action.query. जोड़े में इन विकल्पों का उपयोग करने से हर बार दिल की धड़कन का संदेश भेजे जाने पर एक अलग तालिका में डेटा बदलने का अनुरोध करना संभव हो जाता है। इस प्रकार, एलएसएन जिस पर कनेक्टर वर्तमान में स्थित है (प्रतिकृति स्लॉट में) लगातार अद्यतन किया जाता है। यह DBMS को उन WAL फ़ाइलों को हटाने की अनुमति देता है जिनकी अब आवश्यकता नहीं है। आप इस बारे में अधिक जान सकते हैं कि विकल्प कैसे काम करते हैं प्रलेखन.

एक और विकल्प जो करीब से ध्यान देने योग्य है वह है transforms. हालाँकि यह सुविधा और सुंदरता के बारे में अधिक है...

डिफ़ॉल्ट रूप से, डेबेज़ियम निम्नलिखित नामकरण नीति का उपयोग करके विषय बनाता है: serverName.schemaName.tableName. यह हमेशा सुविधाजनक नहीं हो सकता. विकल्प transforms आप तालिकाओं की सूची को परिभाषित करने के लिए नियमित अभिव्यक्ति का उपयोग कर सकते हैं, जिनमें से घटनाओं को एक विशिष्ट नाम के साथ किसी विषय पर रूट करने की आवश्यकता होती है।

हमारे विन्यास में धन्यवाद transforms निम्नलिखित होता है: मॉनिटर किए गए डेटाबेस से सभी सीडीसी इवेंट नाम के साथ एक विषय पर जाएंगे data.cdc.dbname. अन्यथा (इन सेटिंग्स के बिना), डेबेज़ियम डिफ़ॉल्ट रूप से प्रत्येक तालिका के लिए एक विषय बनाएगा जैसे: pg-dev.public.<table_name>.

कनेक्टर सीमाएँ

PostgreSQL के लिए कनेक्टर कॉन्फ़िगरेशन के विवरण को समाप्त करने के लिए, इसके संचालन की निम्नलिखित विशेषताओं/सीमाओं के बारे में बात करना उचित है:

PostgreSQL के लिए कनेक्टर की कार्यक्षमता तार्किक डिकोडिंग की अवधारणा पर निर्भर करती है। इसलिए वह डेटाबेस संरचना को बदलने के अनुरोधों को ट्रैक नहीं करता है (डीडीएल) - तदनुसार, यह डेटा विषयों में नहीं होगा।
चूंकि प्रतिकृति स्लॉट का उपयोग किया जाता है, इसलिए कनेक्टर को कनेक्ट करना संभव है केवल अग्रणी 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 नोड के ओपलॉग से जानकारी पढ़ता है।

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

दोष सहिष्णुता

हमारे समय में दोष सहनशीलता और उच्च उपलब्धता का मुद्दा पहले से कहीं अधिक गंभीर है - खासकर जब हम डेटा और लेनदेन के बारे में बात कर रहे हैं, और डेटा परिवर्तनों को ट्रैक करना इस मुद्दे से अलग नहीं है। आइए देखें कि सैद्धांतिक रूप से क्या गलत हो सकता है और प्रत्येक मामले में डेबेज़ियम का क्या होगा।

तीन ऑप्ट-आउट विकल्प हैं:

काफ्का कनेक्ट विफलता. यदि कनेक्ट को वितरित मोड में काम करने के लिए कॉन्फ़िगर किया गया है, तो इसके लिए एक ही ग्रुप.आईडी सेट करने के लिए कई श्रमिकों की आवश्यकता होती है। फिर, यदि उनमें से एक विफल हो जाता है, तो कनेक्टर को दूसरे कार्यकर्ता पर पुनः आरंभ किया जाएगा और काफ्का में विषय में अंतिम प्रतिबद्ध स्थिति से पढ़ना जारी रखा जाएगा।
काफ्का क्लस्टर के साथ कनेक्टिविटी का नुकसान. कनेक्टर बस उस स्थिति में पढ़ना बंद कर देगा जो काफ्का को भेजने में विफल रहा, और प्रयास सफल होने तक समय-समय पर इसे फिर से भेजने का प्रयास करेगा।
डेटा स्रोत अनुपलब्धता. कनेक्टर कॉन्फ़िगर किए गए स्रोत से पुनः कनेक्ट करने का प्रयास करेगा। डिफ़ॉल्ट 16 प्रयासों का उपयोग है घातीय बैकऑफ़. 16वें असफल प्रयास के बाद, कार्य को इस रूप में चिह्नित किया जाएगा में विफल रहा है और आपको काफ्का कनेक्ट REST इंटरफ़ेस के माध्यम से इसे मैन्युअल रूप से पुनः आरंभ करने की आवश्यकता होगी।
- के मामले में पोस्टग्रेएसक्यूएल डेटा नष्ट नहीं होगा, क्योंकि प्रतिकृति स्लॉट का उपयोग आपको उन वाल फ़ाइलों को हटाने से रोकेगा जो कनेक्टर द्वारा नहीं पढ़ी जाती हैं। इस मामले में, सिक्के का एक नकारात्मक पक्ष भी है: यदि कनेक्टर और डीबीएमएस के बीच नेटवर्क कनेक्टिविटी लंबे समय तक बाधित रहती है, तो संभावना है कि डिस्क स्थान खत्म हो जाएगा, और इससे विफलता हो सकती है संपूर्ण डीबीएमएस.
- के मामले में MySQL कनेक्टिविटी बहाल होने से पहले बिनलॉग फ़ाइलों को डीबीएमएस द्वारा ही घुमाया जा सकता है। इससे कनेक्टर विफल स्थिति में चला जाएगा, और सामान्य ऑपरेशन को बहाल करने के लिए, आपको बिनलॉग से पढ़ना जारी रखने के लिए प्रारंभिक स्नैपशॉट मोड में पुनः आरंभ करने की आवश्यकता होगी।
- पर MongoDB. दस्तावेज़ में कहा गया है: उस स्थिति में कनेक्टर का व्यवहार जब लॉग/ओप्लॉग फ़ाइलें हटा दी गई हैं और कनेक्टर उस स्थिति से पढ़ना जारी नहीं रख सकता जहां इसे छोड़ा गया था, सभी डीबीएमएस के लिए समान है। इसका मतलब है कि कनेक्टर राज्य में चला जाएगा में विफल रहा है और मोड में पुनः प्रारंभ करने की आवश्यकता होगी प्रारंभिक स्नैपशॉट.
  हालाँकि, कुछ अपवाद भी हैं। यदि कनेक्टर लंबे समय तक डिस्कनेक्ट हो गया था (या MongoDB इंस्टेंस तक नहीं पहुंच सका), और इस दौरान ओपलॉग रोटेशन से गुजरा, तो जब कनेक्शन बहाल हो जाता है, तो कनेक्टर शांति से पहली उपलब्ध स्थिति से डेटा पढ़ना जारी रखेगा, यही कारण है कि काफ्का में कुछ डेटा नहीं मारेंगे.

निष्कर्ष

डेबेज़ियम सीडीसी सिस्टम के साथ मेरा पहला अनुभव है और कुल मिलाकर बहुत सकारात्मक है। परियोजना ने प्रमुख डीबीएमएस के समर्थन, कॉन्फ़िगरेशन में आसानी, क्लस्टरिंग समर्थन और सक्रिय समुदाय के साथ जीत हासिल की। अभ्यास में रुचि रखने वालों के लिए, मेरा सुझाव है कि आप इसके लिए मार्गदर्शिकाएँ पढ़ें काफ्का कनेक्ट и डेबेज़ियम.

काफ्का कनेक्ट के लिए जेडीबीसी कनेक्टर की तुलना में, डेबेज़ियम का मुख्य लाभ यह है कि परिवर्तन डीबीएमएस लॉग से पढ़े जाते हैं, जो न्यूनतम विलंबता के साथ डेटा प्राप्त करने की अनुमति देता है। जेडीबीसी कनेक्टर (काफ्का कनेक्ट से) एक निश्चित अंतराल पर मॉनिटर की गई तालिका पर सवाल उठाता है और (उसी कारण से) डेटा हटाए जाने पर संदेश उत्पन्न नहीं करता है (आप उस डेटा को कैसे क्वेरी कर सकते हैं जो मौजूद नहीं है?)।

समान समस्याओं को हल करने के लिए, आप निम्नलिखित समाधानों पर ध्यान दे सकते हैं (डेबेज़ियम के अतिरिक्त):

जेडीबीसी कनेक्टर काफ्का कनेक्ट;
केवल MySQL के लिए कई समाधान:
ओरेकल गोल्डनगेट, लेकिन यह एक पूरी तरह से अलग "वजन श्रेणी" है।

पुनश्च

हमारे ब्लॉग पर भी पढ़ें:

स्रोत: www.habr.com

पेश है डेबेज़ियम - अपाचे काफ्का के लिए सीडीसी