🥇अपाचे काफ्काको लागि Debezium - CDC प्रस्तुत गर्दै

मेरो काममा, म प्रायः नयाँ प्राविधिक समाधान/सफ्टवेयर उत्पादनहरू भेट्छु, जसको बारेमा रूसी भाषाको इन्टरनेटमा जानकारी कम छ। यस लेखको साथ म मेरो भर्खरको अभ्यासको उदाहरणको साथ एउटा यस्तो खाली ठाउँ भर्ने प्रयास गर्नेछु, जब मलाई दुई लोकप्रिय DBMSs (PostgreSQL र MongoDB) बाट Debezium प्रयोग गरेर काफ्का क्लस्टरमा CDC घटनाहरू पठाउने कन्फिगर गर्न आवश्यक थियो। मलाई आशा छ कि यो समीक्षा लेख, जुन कामको परिणामको रूपमा देखा पर्दछ, अरूलाई उपयोगी हुनेछ।

Debezium र CDC सामान्यतया के हो?

डेबेजियम - CDC सफ्टवेयर कोटि को प्रतिनिधि (डाटा परिवर्तन क्याप्चर गर्नुहोस्), वा अझ स्पष्ट रूपमा, यो Apache Kafka Connect फ्रेमवर्कसँग मिल्दो विभिन्न DBMSs को लागि जडानकर्ताहरूको सेट हो।

यो खुला स्रोत परियोजना, Apache License v2.0 अन्तर्गत इजाजतपत्र प्राप्त र Red Hat द्वारा प्रायोजित। विकास 2016 देखि चलिरहेको छ र हाल यसले निम्न DBMSs को लागि आधिकारिक समर्थन प्रदान गर्दछ: MySQL, PostgreSQL, MongoDB, SQL Server। त्यहाँ क्यासान्ड्रा र ओरेकलका लागि कनेक्टरहरू पनि छन्, तर यस समयमा तिनीहरू "प्रारम्भिक पहुँच" स्थितिमा छन्, र नयाँ रिलीजहरूले पछाडि अनुकूलताको ग्यारेन्टी गर्दैन।

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

अर्को फाइदा यो तथ्य हो कि घटनाहरू भण्डारण गर्न एकल मोडेल प्रयोग गरिन्छ, त्यसैले अन्तिम अनुप्रयोगले विभिन्न DBMS हरू सञ्चालन गर्ने सूक्ष्मताहरूको बारेमा चिन्ता लिनु पर्दैन।

अन्तमा, सन्देश ब्रोकर प्रयोग गरेर डेटामा परिवर्तनहरू ट्र्याक गर्ने स्केल-आउट अनुप्रयोगहरूको लागि स्कोप खोल्छ। एकै समयमा, डाटा स्रोतमा प्रभाव कम गरिएको छ, किनकि डाटा सीधा DBMS बाट प्राप्त गरिएको छैन, तर काफ्का क्लस्टरबाट।

Debezium वास्तुकला को बारे मा

Debezium को प्रयोग यो सरल योजना मा तल आउँछ:

DBMS (डेटा स्रोतको रूपमा) → काफ्का जडानमा कनेक्टर → अपाचे काफ्का → उपभोक्ता

दृष्टान्तको रूपमा, म परियोजना वेबसाइटबाट एउटा रेखाचित्र दिनेछु:

जे होस्, मलाई यो योजना साँच्चै मन पर्दैन, किनकि यो सिङ्क कनेक्टरको मात्र प्रयोग सम्भव छ जस्तो देखिन्छ।

वास्तवमा, स्थिति फरक छ: तपाईंको डाटा लेक भर्दै (माथिको रेखाचित्रमा अन्तिम लिङ्क) Debezium प्रयोग गर्ने एक मात्र तरिका होइन। Apache Kafka लाई पठाइएका घटनाहरू तपाइँका अनुप्रयोगहरूद्वारा विभिन्न परिस्थितिहरूसँग व्यवहार गर्न प्रयोग गर्न सकिन्छ। उदाहरणका लागि:

क्यासबाट अप्रासंगिक डाटा हटाउने;
सूचनाहरू पठाउँदै;
खोज अनुक्रमणिका अपडेटहरू;
केहि प्रकारको अडिट लगहरू;
...

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

यस लेखले विकासकर्ताहरू द्वारा सिफारिस गरिएको वास्तुकलाको बारेमा छलफल गर्नेछ, जसले दोष सहिष्णुता र स्केलेबिलिटी प्रदान गर्दछ।

कनेक्टर कन्फिगरेसन

सबैभन्दा महत्त्वपूर्ण मान - डाटा - मा परिवर्तनहरू ट्र्याक गर्न सुरु गर्न हामीलाई आवश्यक छ:

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

पहिलो दुई बिन्दुहरूमा काम गर्नुहोस्, अर्थात्। DBMS र Apache Kafka स्थापना गर्ने प्रक्रिया लेखको दायराभन्दा बाहिर छ। जे होस्, जो स्यान्डबक्समा सबै कुरा डिप्लोय गर्न चाहन्छन्, त्यहाँ आधिकारिक भण्डारमा उदाहरणहरू सहित तयार-बनाइएको छ। docker-compose.yaml.

हामी थप विवरणमा अन्तिम दुई बिन्दुहरूमा ध्यान केन्द्रित गर्नेछौं।

0. काफ्का जडान

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

यदि तपाइँ कन्फ्लुएन्टबाट काफ्का जडान प्रयोग गर्न चाहानुहुन्छ भने, तपाइँले निर्दिष्ट निर्देशिकामा आवश्यक कनेक्टरहरूको प्लगइनहरू थप्न आवश्यक छ। plugin.path वा वातावरण चर मार्फत सेट गर्नुहोस् CLASSPATH। काफ्का जडान कार्यकर्ता र कनेक्टरहरूका लागि सेटिङहरू कन्फिगरेसन फाइलहरू मार्फत परिभाषित गरिन्छ जुन कार्यकर्ता स्टार्ट आदेशमा तर्कको रूपमा पास गरिन्छ। विवरणको लागि हेर्नुहोस् कागजात.

कनेक्टर संस्करणमा Debeizum सेटअप गर्ने सम्पूर्ण प्रक्रिया दुई चरणहरूमा गरिन्छ। तिनीहरूमध्ये प्रत्येकलाई विचार गरौं:

1. काफ्का जडान फ्रेमवर्क सेटअप गर्दै

Apache Kafka क्लस्टरमा डाटा स्ट्रिम गर्न, विशिष्ट प्यारामिटरहरू काफ्का जडान फ्रेमवर्कमा सेट गरिएका छन्, जस्तै:

क्लस्टर जडान सेटिङहरू,
विषयहरूको नाम जसमा जडानकर्ताको कन्फिगरेसन सीधै भण्डार गरिनेछ,
समूहको नाम जसमा कनेक्टर चलिरहेको छ (यदि वितरित मोड प्रयोग गरिएको छ)।

परियोजनाको आधिकारिक डकर छविले वातावरण चर प्रयोग गरेर कन्फिगरेसनलाई समर्थन गर्दछ - यो हामीले प्रयोग गर्नेछौं। त्यसैले छवि डाउनलोड गरौं:

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 को बारेमा नोट

पूर्वनिर्धारित रूपमा, Debezium ले JSON ढाँचामा डाटा लेख्छ, जुन स्यान्डबक्स र थोरै मात्रामा डाटाका लागि स्वीकार्य छ, तर धेरै लोड गरिएका डाटाबेसमा समस्या हुन सक्छ। JSON कन्भर्टरको विकल्प भनेको सन्देशहरू प्रयोग गरेर क्रमबद्ध गर्नु हो Avro बाइनरी ढाँचामा, जसले Apache Kafka मा I/O सबसिस्टममा लोड घटाउँछ।

Avro प्रयोग गर्नको लागि तपाईंले छुट्टै प्रयोग गर्न आवश्यक छ schema-Registry (स्कीमाहरू भण्डारण गर्नका लागि)। कन्भर्टरका लागि चरहरू यस्तो देखिनेछन्:

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 अनुरोध प्रयोग गरेर काफ्का जडानमा अपलोड गरिएको छ।

२.१। 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 — कन्फिगर गरिएको कनेक्टरद्वारा प्रयोग गरिने 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 - एक अनुरोध जुन प्रत्येक हृदयघात सन्देश पठाउँदा कार्यान्वयन गरिनेछ (विकल्प संस्करण १.१ देखि देखा परेको छ);
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
}
}

दुबै अवस्थामा, रेकर्डहरूमा परिवर्तन भएको रेकर्डको कुञ्जी (PK) र परिवर्तनहरूको सार समावेश हुन्छ: रेकर्ड पहिले के थियो र पछि के भयो।

को मामला मा INSERT: मूल्य पहिले (before) बराबर null, र पछि - सम्मिलित गरिएको रेखा।
को मामला मा UPDATE: at payload.before पङ्क्तिको अघिल्लो स्थिति प्रदर्शित हुन्छ, र मा payload.after - परिवर्तनको सार संग नयाँ।

2.2 MongoDB

यो कनेक्टरले मानक 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>.

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

हाम्रो समयमा गल्ती सहिष्णुता र उच्च उपलब्धताको मुद्दा पहिले भन्दा बढी तीव्र छ - विशेष गरी जब हामी डाटा र लेनदेनको बारेमा कुरा गर्छौं, र डाटा परिवर्तन ट्र्याकिङ यस मामिलामा साइडलाइनमा छैन। सैद्धान्तिक रूपमा के गलत हुन सक्छ र प्रत्येक मामला मा Debezium के हुनेछ हेरौं।

त्यहाँ तीन अप्ट-आउट विकल्पहरू छन्:

काफ्का जडान विफलता। यदि Connect लाई वितरित मोडमा काम गर्न कन्फिगर गरिएको छ भने, यसका लागि एउटै group.id सेट गर्न धेरै कामदारहरू आवश्यक पर्दछ। त्यसपछि, यदि तिनीहरू मध्ये एक असफल भयो भने, कनेक्टर अर्को कार्यकर्तामा पुन: सुरु हुनेछ र Kafka मा शीर्षकमा अन्तिम प्रतिबद्ध स्थितिबाट पढ्न जारी राख्नेछ।
काफ्का क्लस्टरसँग जडानको हानि। कनेक्टरले काफ्कालाई पठाउन असफल भएको स्थितिमा पढ्न बन्द गर्नेछ, र प्रयास सफल नभएसम्म यसलाई पुन: पठाउने प्रयास गर्नेछ।
डाटा स्रोत अनुपलब्ध। कनेक्टरले कन्फिगरेसन अनुसार स्रोतमा पुन: जडान गर्ने प्रयास गर्नेछ। पूर्वनिर्धारित प्रयोग गरेर 16 प्रयासहरू छन् घातीय ब्याकअफ। 16 औं असफल प्रयास पछि, कार्य रूपमा चिन्ह लगाइनेछ असफल भयो र यसलाई Kafka Connect REST इन्टरफेस मार्फत म्यानुअल रूपमा पुन: सुरु गर्न आवश्यक हुनेछ।
- को मामला मा PostgreSQL डाटा हराउने छैन, किनभने प्रतिकृति स्लटहरू प्रयोग गर्नाले तपाईंलाई WAL फाइलहरू मेटाउनबाट रोक्नेछ जुन कनेक्टरले पढेका छैनन्। यस अवस्थामा, त्यहाँ सिक्काको एक नकारात्मक पक्ष पनि छ: यदि कनेक्टर र DBMS बीचको नेटवर्क जडान लामो समयको लागि अवरुद्ध भयो भने, डिस्क स्पेस समाप्त हुने सम्भावना छ, र यसले असफलता निम्त्याउन सक्छ। सम्पूर्ण DBMS।
- को मामला मा MySQL binlog फाइलहरू जडान पुनर्स्थापित हुनु अघि DBMS आफैले घुमाउन सकिन्छ। यसले कनेक्टरलाई असफल अवस्थामा लैजान्छ, र सामान्य सञ्चालन पुनर्स्थापना गर्न binlogs बाट पढ्न जारी राख्न यसलाई प्रारम्भिक स्न्यापसट मोडमा पुन: सुरु गर्न आवश्यक हुनेछ।
- बारेमा MongoDB। कागजातले भन्छ: लग/ओप्लग फाइलहरू मेटिएको अवस्थामा कनेक्टरको व्यवहार र कनेक्टरले छोडेको स्थितिबाट पढ्न जारी राख्न सक्दैन भने सबै DBMS को लागि समान हुन्छ। यो तथ्यमा छ कि कनेक्टर राज्यमा जानेछ असफल भयो र मोडमा पुन: सुरु गर्न आवश्यक हुनेछ प्रारम्भिक स्न्यापसट.
  यद्यपि, त्यहाँ अपवादहरू छन्। यदि कनेक्टर लामो समयको लागि विच्छेदन गरिएको थियो (वा MongoDB उदाहरणमा पुग्न सकेन), र oplog यस समयमा रोटेशन मार्फत गयो, त्यसपछि जब जडान पुनर्स्थापित हुन्छ, कनेक्टरले शान्त रूपमा पहिलो उपलब्ध स्थितिबाट डेटा पढ्न जारी राख्छ, जसका कारण काफ्काका केही तथ्याङ्कहरू छैन प्रहार गर्नेछ।

निष्कर्षमा

Debezium CDC प्रणाली र समग्र धेरै सकारात्मक संग मेरो पहिलो अनुभव हो। परियोजनाले प्रमुख DBMSs, कन्फिगरेसनको सहजता, क्लस्टरिङ समर्थन, र सक्रिय समुदायका लागि यसको समर्थनको साथ जित्यो। अभ्यासमा रुचि राख्नेहरूका लागि, म सुझाव दिन्छु कि तपाइँ गाइडहरू पढ्नुहोस् काफ्का जडान и डेबेजियम.

काफ्का जडानको लागि JDBC कनेक्टरको तुलनामा, Debezium को मुख्य फाइदा यो हो कि परिवर्तनहरू DBMS लगहरूबाट पढिन्छ, जसले डाटालाई न्यूनतम ढिलाइमा प्राप्त गर्न अनुमति दिन्छ। JDBC कनेक्टर (काफ्का जडानद्वारा प्रदान गरिएको) ले ट्र्याक गरिएको तालिकालाई निश्चित अन्तरालमा सोध्छ र (त्यही कारणले) डाटा मेटाउँदा सन्देशहरू उत्पन्न गर्दैन (तपाईले त्यहाँ नभएको डाटाको लागि कसरी क्वेरी गर्न सक्नुहुन्छ?)।

समान समस्याहरू समाधान गर्न, तपाइँ निम्न समाधानहरूमा ध्यान दिन सक्नुहुन्छ (Debezium को अतिरिक्त):

JDBC कनेक्टर काफ्का जडान;
MySQL को लागि मात्र धेरै समाधानहरू:
ओरेकल गोल्डेनगेट, तर यो एकदम फरक "वजन वर्ग" हो।

PS

हाम्रो ब्लगमा पनि पढ्नुहोस्:

स्रोत: www.habr.com

Apache Kafka को लागि Debezium - CDC प्रस्तुत गर्दै