അപ്പാച്ചെ കാഫ്കയ്ക്കായി ഡെബെസിയം - സിഡിസി അവതരിപ്പിക്കുന്നു

എന്റെ ജോലിയിൽ, ഞാൻ പലപ്പോഴും പുതിയ സാങ്കേതിക പരിഹാരങ്ങൾ / സോഫ്റ്റ്വെയർ ഉൽപ്പന്നങ്ങൾ കാണാറുണ്ട്, റഷ്യൻ സംസാരിക്കുന്ന ഇന്റർനെറ്റിൽ ഇവയെക്കുറിച്ചുള്ള വിവരങ്ങൾ കുറവാണ്. ഈ ലേഖനത്തിലൂടെ, എന്റെ സമീപകാല പരിശീലനത്തിൽ നിന്നുള്ള ഒരു വിടവ് നികത്താൻ ഞാൻ ശ്രമിക്കും, രണ്ട് ജനപ്രിയ DBMS-കളിൽ നിന്ന് (PostgreSQL, MongoDB) Debezium ഉപയോഗിച്ച് ഒരു കാഫ്ക ക്ലസ്റ്ററിലേക്ക് CDC ഇവന്റുകൾ അയയ്ക്കുന്നത് സജ്ജീകരിക്കേണ്ടി വന്നപ്പോൾ. ചെയ്ത പ്രവർത്തനത്തിന്റെ ഫലമായി പ്രത്യക്ഷപ്പെട്ട ഈ അവലോകന ലേഖനം മറ്റുള്ളവർക്ക് ഉപയോഗപ്രദമാകുമെന്ന് ഞാൻ പ്രതീക്ഷിക്കുന്നു.

എന്താണ് ഡിബെസിയവും സിഡിസിയും പൊതുവായി?

ഡെബെസിയം - CDC സോഫ്റ്റ്‌വെയർ വിഭാഗത്തിന്റെ പ്രതിനിധി (ഡാറ്റ മാറ്റം ക്യാപ്ചർ ചെയ്യുക), അല്ലെങ്കിൽ കൂടുതൽ കൃത്യമായി പറഞ്ഞാൽ, അപ്പാച്ചെ കാഫ്ക കണക്ട് ഫ്രെയിംവർക്കുമായി പൊരുത്തപ്പെടുന്ന വിവിധ DBMS-കൾക്കായുള്ള കണക്ടറുകളുടെ ഒരു കൂട്ടമാണ് ഇത്.

ഈ ഓപ്പൺ സോഴ്സ് പ്രോജക്റ്റ്, അപ്പാച്ചെ ലൈസൻസ് v2.0 ന് കീഴിൽ ലൈസൻസുള്ളതും Red Hat സ്പോൺസർ ചെയ്യുന്നതും. 2016 മുതൽ വികസനം നടക്കുന്നുണ്ട്, ഇപ്പോൾ ഇത് ഇനിപ്പറയുന്ന DBMS-ന് ഔദ്യോഗിക പിന്തുണ നൽകുന്നു: MySQL, PostgreSQL, MongoDB, SQL സെർവർ. കസാന്ദ്രയ്ക്കും ഒറാക്കിളിനും കണക്ടറുകൾ ഉണ്ട്, എന്നാൽ അവ നിലവിൽ "ഏർലി ആക്‌സസ്" നിലയിലാണ്, പുതിയ പതിപ്പുകൾ പിന്നാക്ക അനുയോജ്യത ഉറപ്പുനൽകുന്നില്ല.

ഞങ്ങൾ സിഡിസിയെ പരമ്പരാഗത സമീപനവുമായി താരതമ്യപ്പെടുത്തുകയാണെങ്കിൽ (അപ്ലിക്കേഷൻ ഡിബിഎംഎസിൽ നിന്നുള്ള ഡാറ്റ നേരിട്ട് വായിക്കുമ്പോൾ), കുറഞ്ഞ ലേറ്റൻസി, ഉയർന്ന വിശ്വാസ്യത, ലഭ്യത എന്നിവ ഉപയോഗിച്ച് വരി തലത്തിൽ ഡാറ്റ മാറ്റ സ്ട്രീമിംഗ് നടപ്പിലാക്കുന്നത് അതിന്റെ പ്രധാന നേട്ടങ്ങളിൽ ഉൾപ്പെടുന്നു. സിഡിസി ഇവന്റുകളുടെ ഒരു ശേഖരമായി കാഫ്ക ക്ലസ്റ്റർ ഉപയോഗിച്ചാണ് അവസാന രണ്ട് പോയിന്റുകൾ നേടിയത്.

കൂടാതെ, ഇവന്റുകൾ സംഭരിക്കുന്നതിന് ഒരൊറ്റ മോഡൽ ഉപയോഗിക്കുന്നു എന്ന വസ്തുതയും ഗുണങ്ങളിൽ ഉൾപ്പെടുന്നു, അതിനാൽ വ്യത്യസ്ത ഡിബിഎംഎസ് പ്രവർത്തിപ്പിക്കുന്നതിന്റെ സൂക്ഷ്മതകളെക്കുറിച്ച് അന്തിമ ആപ്ലിക്കേഷന് വിഷമിക്കേണ്ടതില്ല.

അവസാനമായി, ഒരു സന്ദേശ ബ്രോക്കർ ഉപയോഗിക്കുന്നത് ഡാറ്റയിലെ മാറ്റങ്ങൾ ട്രാക്കുചെയ്യുന്ന ആപ്ലിക്കേഷനുകളുടെ തിരശ്ചീന സ്കെയിലിംഗിനുള്ള സാധ്യത തുറക്കുന്നു. അതേസമയം, ഡിബിഎംഎസിൽ നിന്നല്ല, കാഫ്ക ക്ലസ്റ്ററിൽ നിന്നാണ് ഡാറ്റ ലഭിക്കുന്നത് എന്നതിനാൽ, ഡാറ്റാ ഉറവിടത്തിലെ ആഘാതം കുറയ്ക്കുന്നു.

ഡെബെസിയം വാസ്തുവിദ്യയെക്കുറിച്ച്

Debezium ഉപയോഗിക്കുന്നത് ഈ ലളിതമായ സ്കീമിലേക്ക് വരുന്നു:

DBMS (ഡാറ്റ ഉറവിടമായി) → കാഫ്ക കണക്റ്റിലെ കണക്റ്റർ → അപ്പാച്ചെ കാഫ്ക → ഉപഭോക്താവ്

ഒരു ചിത്രീകരണമെന്ന നിലയിൽ, പ്രോജക്റ്റ് വെബ്സൈറ്റിൽ നിന്ന് ഞാൻ ഒരു ഡയഗ്രം നൽകും:

എന്നിരുന്നാലും, ഈ സ്കീം എനിക്ക് ശരിക്കും ഇഷ്ടമല്ല, കാരണം ഒരു സിങ്ക് കണക്റ്റർ മാത്രമേ സാധ്യമാകൂ എന്ന് തോന്നുന്നു.

വാസ്തവത്തിൽ, സ്ഥിതി വ്യത്യസ്തമാണ്: നിങ്ങളുടെ ഡാറ്റ തടാകം നിറയ്ക്കുക (മുകളിലുള്ള ഡയഗ്രാമിലെ അവസാന ലിങ്ക്) Debezium ഉപയോഗിക്കാനുള്ള ഒരേയൊരു മാർഗ്ഗമല്ല. അപ്പാച്ചെ കാഫ്കയിലേക്ക് അയച്ച ഇവന്റുകൾ വിവിധ സാഹചര്യങ്ങൾ കൈകാര്യം ചെയ്യാൻ നിങ്ങളുടെ ആപ്ലിക്കേഷനുകൾക്ക് ഉപയോഗിക്കാനാകും. ഉദാഹരണത്തിന്:

• കാഷെയിൽ നിന്ന് അപ്രസക്തമായ ഡാറ്റ നീക്കംചെയ്യൽ;
• അറിയിപ്പുകൾ അയയ്ക്കുന്നു;
• തിരയൽ സൂചിക അപ്ഡേറ്റുകൾ;
• ചില തരത്തിലുള്ള ഓഡിറ്റ് ലോഗുകൾ;
• പങ്ക് € |

നിങ്ങൾക്ക് ഒരു ജാവ ആപ്ലിക്കേഷൻ ഉണ്ടെങ്കിൽ ഒരു കാഫ്ക ക്ലസ്റ്റർ ഉപയോഗിക്കേണ്ട ആവശ്യമില്ല/സാധ്യത ഇല്ലെങ്കിൽ, പ്രവർത്തിക്കാനുള്ള സാധ്യതയും ഉണ്ട്. ഉൾച്ചേർത്ത കണക്റ്റർ. വ്യക്തമായ പ്ലസ്, അതുപയോഗിച്ച് നിങ്ങൾക്ക് അധിക ഇൻഫ്രാസ്ട്രക്ചർ നിരസിക്കാൻ കഴിയും (ഒരു കണക്ടറിന്റെയും കാഫ്കയുടെയും രൂപത്തിൽ). എന്നിരുന്നാലും, ഈ സൊല്യൂഷൻ പതിപ്പ് 1.1 മുതൽ അവസാനിപ്പിച്ചതിനാൽ ഉപയോഗത്തിന് ഇനി ശുപാർശ ചെയ്യുന്നില്ല (ഭാവിയിലെ റിലീസുകളിൽ ഇത് നീക്കം ചെയ്തേക്കാം).

ഈ ലേഖനം ഡെവലപ്പർമാർ ശുപാർശ ചെയ്യുന്ന വാസ്തുവിദ്യയെക്കുറിച്ച് ചർച്ച ചെയ്യും, അത് തെറ്റ് സഹിഷ്ണുതയും സ്കേലബിളിറ്റിയും നൽകുന്നു.

കണക്റ്റർ കോൺഫിഗറേഷൻ

ഏറ്റവും പ്രധാനപ്പെട്ട മൂല്യമായ ഡാറ്റയിലെ മാറ്റങ്ങൾ ട്രാക്കുചെയ്യാൻ ആരംഭിക്കുന്നതിന് ഞങ്ങൾക്ക് ഇത് ആവശ്യമാണ്:

1. ഡാറ്റാ ഉറവിടം, പതിപ്പ് 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (MongDB XNUMX+) മുതൽ MySQL ആകാംപട്ടിക പൂർത്തിയാക്കുക);
2. അപ്പാച്ചെ കാഫ്ക ക്ലസ്റ്റർ
3. കാഫ്ക കണക്ട് ഇൻസ്റ്റൻസ് (പതിപ്പുകൾ 1.x, 2.x);
4. കോൺഫിഗർ ചെയ്ത Debezium കണക്റ്റർ.

ആദ്യത്തെ രണ്ട് പോയിന്റുകളിൽ പ്രവർത്തിക്കുക, അതായത്. ഒരു ഡിബിഎംഎസും അപ്പാച്ചെ കാഫ്കയും ഇൻസ്റ്റാൾ ചെയ്യുന്ന പ്രക്രിയ ലേഖനത്തിന്റെ പരിധിക്കപ്പുറമാണ്. എന്നിരുന്നാലും, എല്ലാം ഒരു സാൻഡ്‌ബോക്‌സിൽ വിന്യസിക്കാൻ ആഗ്രഹിക്കുന്നവർക്ക്, ഉദാഹരണങ്ങളുള്ള ഒരു റെഡിമെയ്ഡ് ഒഫീഷ്യൽ റിപ്പോസിറ്ററിയിൽ ഉണ്ട്. ഡോക്കർ-compose.yaml.

അവസാന രണ്ട് പോയിന്റുകളിൽ ഞങ്ങൾ കൂടുതൽ വിശദമായി ശ്രദ്ധ കേന്ദ്രീകരിക്കും.

0. കാഫ്ക കണക്ട്

ഇവിടെയും പിന്നീടുള്ള ലേഖനത്തിലും, എല്ലാ കോൺഫിഗറേഷൻ ഉദാഹരണങ്ങളും Debezium ഡെവലപ്പർമാർ വിതരണം ചെയ്ത ഡോക്കർ ഇമേജിന്റെ പശ്ചാത്തലത്തിൽ പരിഗണിക്കുന്നു. ആവശ്യമായ എല്ലാ പ്ലഗിൻ ഫയലുകളും (കണക്ടറുകൾ) ഇതിൽ അടങ്ങിയിരിക്കുന്നു കൂടാതെ എൻവയോൺമെന്റ് വേരിയബിളുകൾ ഉപയോഗിച്ച് കാഫ്ക കണക്ട് കോൺഫിഗറേഷൻ നൽകുന്നു.

നിങ്ങൾ Confluent-ൽ നിന്ന് Kafka Connect ഉപയോഗിക്കാൻ ഉദ്ദേശിക്കുന്നുവെങ്കിൽ, നിർദ്ദേശിച്ചിരിക്കുന്ന ഡയറക്ടറിയിലേക്ക് ആവശ്യമായ കണക്ടറുകളുടെ പ്ലഗിനുകൾ നിങ്ങൾ തന്നെ ചേർക്കേണ്ടതുണ്ട്. plugin.path അല്ലെങ്കിൽ ഒരു പരിസ്ഥിതി വേരിയബിൾ വഴി സജ്ജമാക്കുക CLASSPATH. വർക്കർ സ്റ്റാർട്ട് കമാൻഡിലേക്ക് ആർഗ്യുമെന്റുകളായി കൈമാറുന്ന കോൺഫിഗറേഷൻ ഫയലുകളിലൂടെയാണ് കാഫ്ക കണക്ട് വർക്കറിനും കണക്ടറുകൾക്കുമുള്ള ക്രമീകരണങ്ങൾ നിർവചിച്ചിരിക്കുന്നത്. വിശദാംശങ്ങൾക്ക് കാണുക പ്രമാണീകരണം.

കണക്റ്റർ പതിപ്പിൽ Debeizum സജ്ജീകരിക്കുന്നതിനുള്ള മുഴുവൻ പ്രക്രിയയും രണ്ട് ഘട്ടങ്ങളിലായാണ് നടത്തുന്നത്. അവ ഓരോന്നും നമുക്ക് പരിഗണിക്കാം:

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 സബ്സിസ്റ്റത്തിലെ ലോഡ് കുറയ്ക്കുന്നു.

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. കണക്റ്റർ തന്നെ സജ്ജീകരിക്കുന്നു

ഇപ്പോൾ നിങ്ങൾക്ക് കണക്റ്ററിന്റെ കോൺഫിഗറേഷനിലേക്ക് നേരിട്ട് പോകാം, അത് ഉറവിടത്തിൽ നിന്നുള്ള ഡാറ്റ വായിക്കും.

രണ്ട് ഡിബിഎംഎസുകൾക്കായുള്ള കണക്റ്ററുകളുടെ ഉദാഹരണം നോക്കാം: PostgreSQL, MongoDB, ഇവയ്ക്ക് എനിക്ക് അനുഭവമുണ്ട്, അവയ്ക്ക് വ്യത്യാസങ്ങളുണ്ട് (ചെറുതാണെങ്കിലും ചില സന്ദർഭങ്ങളിൽ പ്രാധാന്യമുണ്ട്!).

കോൺഫിഗറേഷൻ JSON നൊട്ടേഷനിൽ വിവരിക്കുകയും ഒരു 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 - താഴെ വിവരിച്ചിരിക്കുന്ന കോൺഫിഗറേഷൻ ഉപയോഗിക്കുന്ന കണക്ടറിന്റെ പേര്; ഭാവിയിൽ, ഈ പേര് കാഫ്ക കണക്ട് 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 (ലോഗ് സീക്വൻസ് നമ്പർ) എഴുതുകയും ചെയ്യുന്നു. offset. എന്നാൽ കണക്ടർ മുഴുവൻ ഡാറ്റാബേസും അല്ല, അതിന്റെ ടേബിളുകളുടെ ഒരു ഭാഗം മാത്രം വായിക്കാൻ കോൺഫിഗർ ചെയ്‌താൽ എന്ത് സംഭവിക്കും (ഇതിൽ ഡാറ്റ ഇടയ്ക്കിടെ അപ്‌ഡേറ്റ് ചെയ്യപ്പെടുന്നു)?

• കണക്റ്റർ WAL ഫയലുകൾ വായിക്കുകയും അത് നിരീക്ഷിക്കുന്ന ടേബിളുകളിൽ ഇടപാടുകൾ കണ്ടെത്തുകയും ചെയ്യും.
• അതിനാൽ, വിഷയത്തിലോ റെപ്ലിക്കേഷൻ സ്ലോട്ടിലോ അതിന്റെ നിലവിലെ സ്ഥാനം അപ്ഡേറ്റ് ചെയ്യില്ല.
• ഇത്, WAL ഫയലുകൾ ഡിസ്കിൽ "സ്റ്റക്ക്" ആകാൻ ഇടയാക്കുകയും ഡിസ്കിൽ ഇടം ഇല്ലാതാകുകയും ചെയ്യും.

ഇവിടെ ഓപ്ഷനുകൾ രക്ഷാപ്രവർത്തനത്തിലേക്ക് വരുന്നു. heartbeat.interval.ms и heartbeat.action.query. ജോഡികളായി ഈ ഓപ്‌ഷനുകൾ ഉപയോഗിക്കുന്നത് ഓരോ തവണയും ഹൃദയമിടിപ്പ് സന്ദേശം അയയ്‌ക്കുമ്പോൾ ഒരു പ്രത്യേക പട്ടികയിൽ ഡാറ്റ മാറ്റുന്നതിനുള്ള അഭ്യർത്ഥന നടപ്പിലാക്കുന്നത് സാധ്യമാക്കുന്നു. അങ്ങനെ, കണക്റ്റർ നിലവിൽ സ്ഥിതിചെയ്യുന്ന എൽഎസ്എൻ (റെപ്ലിക്കേഷൻ സ്ലോട്ടിൽ) നിരന്തരം അപ്ഡേറ്റ് ചെയ്യുന്നു. ഇത് ഇനി ആവശ്യമില്ലാത്ത WAL ഫയലുകൾ നീക്കം ചെയ്യാൻ DBMS-നെ അനുവദിക്കുന്നു. ഓപ്ഷനുകൾ എങ്ങനെ പ്രവർത്തിക്കുന്നു എന്നതിനെക്കുറിച്ചുള്ള കൂടുതൽ വിവരങ്ങൾക്ക്, കാണുക പ്രമാണീകരണം.

കൂടുതൽ ശ്രദ്ധ അർഹിക്കുന്ന മറ്റൊരു ഓപ്ഷൻ 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
}
}

രണ്ട് സാഹചര്യങ്ങളിലും, റെക്കോർഡുകൾ മാറ്റിയ റെക്കോർഡിന്റെ കീയും (പികെ) മാറ്റങ്ങളുടെ സാരാംശവും ഉൾക്കൊള്ളുന്നു: റെക്കോർഡ് മുമ്പ് എന്തായിരുന്നു, അതിനുശേഷം എന്തായിത്തീർന്നു.

• കേസിൽ INSERT: മുമ്പത്തെ മൂല്യം (before) തുല്യമാണ് nullതിരുകിയ ചരട് പിന്നാലെ.
• കേസിൽ UPDATE: at payload.before വരിയുടെ മുമ്പത്തെ അവസ്ഥ പ്രദർശിപ്പിക്കും, ഒപ്പം ഇൻ payload.after - മാറ്റത്തിന്റെ സത്തയിൽ പുതിയത്.

2.2 മോംഗോഡിബി

ഡിബിഎംഎസ് പ്രൈമറി നോഡിന്റെ ഒപ്ലോഗിൽ നിന്നുള്ള വിവരങ്ങൾ വായിക്കുന്ന സ്റ്റാൻഡേർഡ് മോംഗോഡിബി റെപ്ലിക്കേഷൻ മെക്കാനിസം ഈ കണക്റ്റർ ഉപയോഗിക്കുന്നു.

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. കാഫ്ക കണക്ട് പരാജയം. വിതരണ മോഡിൽ പ്രവർത്തിക്കാൻ കണക്റ്റ് കോൺഫിഗർ ചെയ്‌തിട്ടുണ്ടെങ്കിൽ, ഇതിന് ഒരേ group.id സജ്ജീകരിക്കാൻ ഒന്നിലധികം തൊഴിലാളികൾ ആവശ്യമാണ്. തുടർന്ന്, അവയിലൊന്ന് പരാജയപ്പെടുകയാണെങ്കിൽ, മറ്റൊരു തൊഴിലാളിയിൽ കണക്റ്റർ പുനരാരംഭിക്കുകയും കാഫ്കയിലെ വിഷയത്തിലെ അവസാന പ്രതിബദ്ധതയുള്ള സ്ഥാനത്ത് നിന്ന് വായന തുടരുകയും ചെയ്യും.
2. കാഫ്ക ക്ലസ്റ്ററുമായുള്ള ബന്ധം നഷ്ടപ്പെടുന്നു. കണക്റ്റർ കാഫ്കയിലേക്ക് അയയ്‌ക്കുന്നതിൽ പരാജയപ്പെട്ട സ്ഥാനത്ത് വായന നിർത്തുകയും ശ്രമം വിജയിക്കുന്നതുവരെ ഇടയ്‌ക്കിടെ അത് വീണ്ടും അയയ്‌ക്കാൻ ശ്രമിക്കുകയും ചെയ്യും.
3. ഡാറ്റ ഉറവിടം ലഭ്യമല്ല. കോൺഫിഗറേഷൻ അനുസരിച്ച് ഉറവിടത്തിലേക്ക് വീണ്ടും കണക്റ്റുചെയ്യാൻ കണക്റ്റർ ശ്രമിക്കും. 16 ശ്രമങ്ങൾ ഉപയോഗിച്ചാണ് ഡിഫോൾട്ട് എക്സ്പോണൻഷ്യൽ ബാക്ക്ഓഫ്. 16-ാമത്തെ ശ്രമം പരാജയപ്പെട്ടതിന് ശേഷം, ടാസ്‌ക് ഇതായി അടയാളപ്പെടുത്തും പരാജയപ്പെട്ടു കാഫ്ക കണക്ട് REST ഇന്റർഫേസ് വഴി ഇത് സ്വമേധയാ പുനരാരംഭിക്കേണ്ടതുണ്ട്.
   • കേസിൽ PostgreSQL എന്നീ ഡാറ്റ നഷ്ടപ്പെടില്ല, കാരണം റെപ്ലിക്കേഷൻ സ്ലോട്ടുകൾ ഉപയോഗിക്കുന്നത് കണക്റ്റർ വായിക്കാത്ത WAL ഫയലുകൾ ഇല്ലാതാക്കുന്നത് തടയും. ഈ സാഹചര്യത്തിൽ, ഒരു പോരായ്മയുണ്ട്: കണക്ടറും ഡിബിഎംഎസും തമ്മിലുള്ള നെറ്റ്‌വർക്ക് കണക്റ്റിവിറ്റി വളരെക്കാലം തടസ്സപ്പെട്ടാൽ, ഡിസ്ക് സ്പേസ് തീർന്നുപോകാനുള്ള അവസരമുണ്ട്, ഇത് മുഴുവൻ ഡിബിഎംഎസിന്റെയും പരാജയത്തിലേക്ക് നയിച്ചേക്കാം.
   • കേസിൽ MySQL കണക്റ്റിവിറ്റി പുനഃസ്ഥാപിക്കുന്നതിനുമുമ്പ് binlog ഫയലുകൾ DBMS-ന് തന്നെ തിരിക്കാൻ കഴിയും. ഇത് കണക്റ്റർ പരാജയപ്പെട്ട അവസ്ഥയിലേക്ക് പോകുന്നതിന് കാരണമാകും, സാധാരണ പ്രവർത്തനം പുനഃസ്ഥാപിക്കുന്നതിന് ബിൻലോഗുകളിൽ നിന്ന് വായന തുടരുന്നതിന് അത് പ്രാരംഭ സ്നാപ്പ്ഷോട്ട് മോഡിൽ പുനരാരംഭിക്കേണ്ടതുണ്ട്.
   • ഓൺ മോംഗോഡിബി. ഡോക്യുമെന്റേഷൻ പറയുന്നു: ലോഗ്/ഒപ്ലോഗ് ഫയലുകൾ ഇല്ലാതാക്കുകയും കണക്ടറിന് അത് നിർത്തിയ സ്ഥാനത്ത് നിന്ന് റീഡിംഗ് തുടരാൻ കഴിയാതിരിക്കുകയും ചെയ്താൽ കണക്ടറിന്റെ പെരുമാറ്റം എല്ലാ DBMS-നും സമാനമാണ്. കണക്റ്റർ സംസ്ഥാനത്തിലേക്ക് പോകുമെന്ന വസ്തുതയിലാണ് ഇത് സ്ഥിതിചെയ്യുന്നത് പരാജയപ്പെട്ടു കൂടാതെ മോഡിൽ പുനരാരംഭിക്കേണ്ടതുണ്ട് പ്രാരംഭ സ്നാപ്പ്ഷോട്ട്.

     എന്നിരുന്നാലും, ഒഴിവാക്കലുകൾ ഉണ്ട്. കണക്റ്റർ വളരെക്കാലം വിച്ഛേദിക്കപ്പെട്ട അവസ്ഥയിലാണെങ്കിൽ (അല്ലെങ്കിൽ മോംഗോഡിബി ഉദാഹരണത്തിൽ എത്താൻ കഴിഞ്ഞില്ല), ഈ സമയത്ത് ഓപ്ലോഗ് തിരിക്കുകയാണെങ്കിൽ, കണക്ഷൻ പുനഃസ്ഥാപിക്കുമ്പോൾ, കണക്റ്റർ ആദ്യം ലഭ്യമായ സ്ഥാനത്ത് നിന്ന് ഡാറ്റ വായിക്കുന്നത് തുടരും. , അതുകൊണ്ടാണ് കാഫ്കയിലെ ചില ഡാറ്റ അല്ല അടിക്കും.

തീരുമാനം

സിഡിസി സിസ്റ്റങ്ങളുമായുള്ള എന്റെ ആദ്യ അനുഭവമാണ് ഡെബെസിയം, മൊത്തത്തിൽ വളരെ പോസിറ്റീവ് ആണ്. പ്രധാന DBMS-ന്റെ പിന്തുണ, കോൺഫിഗറേഷൻ എളുപ്പം, ക്ലസ്റ്ററിംഗിനുള്ള പിന്തുണ, ഒരു സജീവ കമ്മ്യൂണിറ്റി എന്നിവയ്ക്ക് പ്രോജക്റ്റ് കൈക്കൂലി നൽകി. പരിശീലനത്തിൽ താൽപ്പര്യമുള്ളവർക്കായി, ഗൈഡുകൾ വായിക്കാൻ ഞാൻ ശുപാർശ ചെയ്യുന്നു കാഫ്ക കണക്ട് и ഡെബെസിയം.

Kafka Connect-നുള്ള JDBC കണക്റ്ററുമായി താരതമ്യപ്പെടുത്തുമ്പോൾ, Debezium-ന്റെ പ്രധാന നേട്ടം DBMS ലോഗുകളിൽ നിന്ന് മാറ്റങ്ങൾ വായിക്കുന്നു എന്നതാണ്, ഇത് കുറഞ്ഞ കാലതാമസത്തോടെ ഡാറ്റ സ്വീകരിക്കാൻ അനുവദിക്കുന്നു. JDBC കണക്റ്റർ (കാഫ്ക കണക്ട് നൽകിയത്) ഒരു നിശ്ചിത ഇടവേളയിൽ ട്രാക്ക് ചെയ്‌ത പട്ടികയെ അന്വേഷിക്കുന്നു, (അതേ കാരണത്താൽ) ഡാറ്റ ഇല്ലാതാക്കുമ്പോൾ സന്ദേശങ്ങൾ സൃഷ്‌ടിക്കുന്നില്ല (ഇല്ലാത്ത ഡാറ്റയ്‌ക്കായി നിങ്ങൾക്ക് എങ്ങനെ അന്വേഷിക്കാനാകും?).

സമാനമായ പ്രശ്നങ്ങൾ പരിഹരിക്കുന്നതിന്, നിങ്ങൾക്ക് ഇനിപ്പറയുന്ന പരിഹാരങ്ങൾ ശ്രദ്ധിക്കാവുന്നതാണ് (Debezium കൂടാതെ):

• JDBC കണക്റ്റർ കാഫ്ക കണക്ട്
• കുറച്ച് MySQL-മാത്രം പരിഹാരങ്ങൾ:
  • MySQL സ്ട്രീമർ;
  • മാക്സ്വെൽ;
  • നട്ടെല്ല് ടാപ്പ്;
• ഒറാക്കിൾ ഗോൾഡൻഗേറ്റ്, എന്നാൽ ഇത് തികച്ചും വ്യത്യസ്തമായ "ഭാര വിഭാഗം" ആണ്.

പി.എസ്

ഞങ്ങളുടെ ബ്ലോഗിലും വായിക്കുക:

• «കുബർനെറ്റസിലെ ഒരു കാഫ്ക ക്ലസ്റ്ററിന് അനുയോജ്യമായ വലുപ്പം നിർണ്ണയിക്കുക";
• «ഞങ്ങളുടെ SRE ദൈനംദിന ജീവിതത്തിൽ നിന്നുള്ള പ്രായോഗിക കഥകൾ. ഭാഗം 2";
• «Kubernetes, ഞങ്ങളുടെ തിരഞ്ഞെടുപ്പുകൾ, അനുഭവങ്ങൾ എന്നിവയ്‌ക്കായുള്ള PostgreSQL പ്രസ്താവനകളുടെ ഒരു ഹ്രസ്വ അവലോകനം".

അവലംബം: www.habr.com