Apache Kafka માટે Debezium - CDC નો પરિચય

મારા કાર્યમાં, હું વારંવાર નવા તકનીકી ઉકેલો/સોફ્ટવેર ઉત્પાદનોને જોઉં છું, જેની માહિતી રશિયન-ભાષાના ઇન્ટરનેટ પર ખૂબ જ દુર્લભ છે. આ લેખ સાથે હું મારી તાજેતરની પ્રેક્ટિસના ઉદાહરણ સાથે આવા એક તફાવતને ભરવાનો પ્રયાસ કરીશ, જ્યારે મને બે લોકપ્રિય DBMSs (PostgreSQL અને MongoDB) માંથી Debezium નો ઉપયોગ કરીને કાફકા ક્લસ્ટરમાં CDC ઇવેન્ટ્સ મોકલવાનું ગોઠવવાની જરૂર પડી. હું આશા રાખું છું કે આ સમીક્ષા લેખ, જે કરેલા કાર્યના પરિણામે દેખાય છે, તે અન્ય લોકો માટે ઉપયોગી થશે.

સામાન્ય રીતે ડેબેઝિયમ અને સીડીસી શું છે?

ડેબેઝિયમ - સીડીસી સોફ્ટવેર શ્રેણીના પ્રતિનિધિ (કેપ્ચર ડેટા ચેન્જ), અથવા વધુ સ્પષ્ટ રીતે, તે Apache Kafka Connect ફ્રેમવર્ક સાથે સુસંગત વિવિધ DBMSs માટે કનેક્ટર્સનો સમૂહ છે.

આ ઓપન સોર્સ પ્રોજેક્ટ, Apache License v2.0 હેઠળ લાઇસન્સ અને Red Hat દ્વારા પ્રાયોજિત. વિકાસ 2016 થી ચાલુ છે અને હાલમાં તે નીચેના DBMSs માટે સત્તાવાર સમર્થન પ્રદાન કરે છે: MySQL, PostgreSQL, MongoDB, SQL સર્વર. Cassandra અને Oracle માટે કનેક્ટર્સ પણ છે, પરંતુ આ ક્ષણે તેઓ "પ્રારંભિક ઍક્સેસ" સ્થિતિમાં છે, અને નવા પ્રકાશનો પછાત સુસંગતતાની ખાતરી આપતા નથી.

જો આપણે પરંપરાગત અભિગમ (જ્યારે એપ્લિકેશન સીધા DBMS માંથી ડેટા વાંચે છે) સાથે CDC ની સરખામણી કરીએ, તો તેના મુખ્ય ફાયદાઓમાં ઓછી વિલંબતા, ઉચ્ચ વિશ્વસનીયતા અને ઉપલબ્ધતા સાથે પંક્તિ સ્તર પર ડેટા ચેન્જ સ્ટ્રીમિંગનો અમલ શામેલ છે. છેલ્લા બે મુદ્દાઓ CDC ઇવેન્ટ્સ માટે ભંડાર તરીકે કાફકા ક્લસ્ટરનો ઉપયોગ કરીને પ્રાપ્ત થાય છે.

બીજો ફાયદો એ હકીકત છે કે એક જ મોડેલનો ઉપયોગ ઇવેન્ટ્સને સંગ્રહિત કરવા માટે થાય છે, તેથી અંતિમ એપ્લિકેશનને વિવિધ DBMS ને ચલાવવાની ઘોંઘાટ વિશે ચિંતા કરવાની જરૂર નથી.

છેલ્લે, મેસેજ બ્રોકરનો ઉપયોગ કરીને એપ્લીકેશનને મંજૂરી આપે છે જે ડેટામાં થતા ફેરફારોને આડી રીતે માપવા માટે મોનિટર કરે છે. તે જ સમયે, ડેટા સ્ત્રોત પરની અસર ઓછી કરવામાં આવે છે, કારણ કે ડેટા સીધો DBMS થી નહીં, પરંતુ કાફકા ક્લસ્ટરમાંથી મેળવવામાં આવે છે.

ડેબેઝિયમ આર્કિટેક્ચર વિશે

ડેબેઝિયમનો ઉપયોગ આ સરળ યોજનામાં આવે છે:

DBMS (ડેટા સ્ત્રોત તરીકે) → કાફકા કનેક્ટમાં કનેક્ટર → અપાચે કાફકા → ઉપભોક્તા

એક ઉદાહરણ તરીકે, અહીં પ્રોજેક્ટ વેબસાઇટ પરથી એક આકૃતિ છે:

જો કે, મને આ યોજના ખરેખર ગમતી નથી, કારણ કે એવું લાગે છે કે ફક્ત સિંક કનેક્ટરનો ઉપયોગ શક્ય છે.

વાસ્તવમાં, પરિસ્થિતિ અલગ છે: તમારા ડેટા લેકને ભરવા (ઉપરના ચિત્રમાં છેલ્લી લિંક) ડેબેઝિયમનો ઉપયોગ કરવાનો આ એકમાત્ર રસ્તો નથી. Apache Kafka ને મોકલવામાં આવેલ ઇવેન્ટ્સનો ઉપયોગ તમારી એપ્લિકેશન દ્વારા વિવિધ પરિસ્થિતિઓને હેન્ડલ કરવા માટે કરી શકાય છે. દાખ્લા તરીકે:

• કેશમાંથી અપ્રસ્તુત ડેટા દૂર કરી રહ્યા છીએ;
• સૂચનાઓ મોકલવી;
• અનુક્રમણિકા અપડેટ્સ શોધો;
• અમુક પ્રકારના ઓડિટ લોગ;
• ...

જો તમારી પાસે જાવા એપ્લિકેશન હોય અને કાફકા ક્લસ્ટરનો ઉપયોગ કરવાની કોઈ જરૂર/સંભવ ન હોય, તો તેના દ્વારા કામ કરવાની શક્યતા પણ છે. એમ્બેડેડ-કનેક્ટર. સ્પષ્ટ ફાયદો એ છે કે તે વધારાના ઇન્ફ્રાસ્ટ્રક્ચરની જરૂરિયાતને દૂર કરે છે (એક કનેક્ટર અને કાફકાના સ્વરૂપમાં). જો કે, આ સોલ્યુશન સંસ્કરણ 1.1 થી નાપસંદ કરવામાં આવ્યું છે અને તેનો ઉપયોગ કરવા માટે હવે ભલામણ કરવામાં આવતી નથી (તેના માટે સમર્થન ભવિષ્યના પ્રકાશનોમાં દૂર કરવામાં આવી શકે છે).

આ લેખ વિકાસકર્તાઓ દ્વારા ભલામણ કરાયેલ આર્કિટેક્ચરની ચર્ચા કરશે, જે ખામી સહિષ્ણુતા અને માપનીયતા પ્રદાન કરે છે.

કનેક્ટર ગોઠવણી

સૌથી મહત્વપૂર્ણ મૂલ્ય - ડેટા - માં ફેરફારોને ટ્રૅક કરવાનું શરૂ કરવા માટે અમને જરૂર છે:

1. ડેટા સ્ત્રોત, જે વર્ઝન 5.7, PostgreSQL 9.6+, MongoDB 3.2+ થી શરૂ કરીને MySQL હોઈ શકે છે (સંપૂર્ણ સૂચિ);
2. અપાચે કાફકા ક્લસ્ટર;
3. કાફકા કનેક્ટ ઉદાહરણ (સંસ્કરણ 1.x, 2.x);
4. રૂપરેખાંકિત ડેબેઝિયમ કનેક્ટર.

પ્રથમ બે મુદ્દાઓ પર કામ કરો, એટલે કે. DBMS અને Apache Kafka ની ઇન્સ્ટોલેશન પ્રક્રિયા લેખના અવકાશની બહાર છે. જો કે, જેઓ સેન્ડબોક્સમાં બધું જમાવવા માંગે છે, તેમના માટે ઉદાહરણો સાથેના સત્તાવાર ભંડારમાં તૈયાર છે. 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. કનેક્ટરને જ રૂપરેખાંકિત કરી રહ્યું છે

હવે તમે સીધા જ કનેક્ટરની ગોઠવણી પર જઈ શકો છો, જે સ્રોતમાંથી ડેટા વાંચશે.

ચાલો બે DBMSs માટે કનેક્ટર્સનું ઉદાહરણ જોઈએ: 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 — પોસ્ટગ્રેએસક્યુએલ ઇન્સ્ટન્સ નામનો ઉપયોગ કાફકા ક્લસ્ટરમાં વિષયનું નામ બનાવવા માટે થાય છે;
• 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 નીચે મુજબ થાય છે: મોનિટર કરેલ ડેટાબેઝમાંથી તમામ સીડીસી ઇવેન્ટ નામ સાથેના વિષય પર જશે data.cdc.dbname. નહિંતર (આ સેટિંગ્સ વિના), Debezium મૂળભૂત રીતે દરેક કોષ્ટક માટે એક વિષય બનાવશે જેમ કે: pg-dev.public.<table_name>.

કનેક્ટર મર્યાદાઓ

PostgreSQL માટે કનેક્ટર રૂપરેખાંકનનું વર્ણન પૂર્ણ કરવા માટે, તેના ઓપરેશનની નીચેની સુવિધાઓ/મર્યાદાઓ વિશે વાત કરવી યોગ્ય છે:

1. PostgreSQL માટે કનેક્ટરની કાર્યક્ષમતા લોજિકલ ડીકોડિંગની વિભાવના પર આધાર રાખે છે. તેથી તેમણે ડેટાબેઝ માળખું બદલવાની વિનંતીઓને ટ્રૅક કરતું નથી (DDL) - તદનુસાર, આ ડેટા વિષયોમાં રહેશે નહીં.
2. પ્રતિકૃતિ સ્લોટનો ઉપયોગ થતો હોવાથી, કનેક્ટરને કનેક્ટ કરવું શક્ય છે માત્ર અગ્રણી DBMS ઉદાહરણ માટે.
3. જો વપરાશકર્તા કે જેના હેઠળ કનેક્ટર ડેટાબેઝ સાથે જોડાય છે તેની પાસે ફક્ત વાંચવાના અધિકારો છે, તો પછી પ્રથમ લોન્ચ પહેલાં તમારે મેન્યુઅલી પ્રતિકૃતિ સ્લોટ બનાવવાની અને ડેટાબેઝ પર પ્રકાશિત કરવાની જરૂર પડશે.

રૂપરેખાંકન લાગુ કરી રહ્યા છીએ

તો, ચાલો આપણું રૂપરેખાંકન કનેક્ટરમાં લોડ કરીએ:

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

અમે તપાસીએ છીએ કે ડાઉનલોડ સફળ થયું અને કનેક્ટર શરૂ થયું:

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

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

સરસ: તે સેટ થઈ ગયું છે અને જવા માટે તૈયાર છે. હવે ચાલો ઉપભોક્તા હોવાનો ડોળ કરીએ અને કાફકા સાથે કનેક્ટ થઈએ, ત્યારબાદ આપણે કોષ્ટકમાં એક એન્ટ્રી ઉમેરીશું અને બદલીશું:

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

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', '[email protected]');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

અમારા વિષયમાં તે નીચે મુજબ પ્રદર્શિત થશે:

અમારા ફેરફારો સાથે ખૂબ જ લાંબી JSON

{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":null,
"after":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600374991648,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":602,
"lsn":34088472,
"xmin":null
},
"op":"c",
"ts_ms":1600374991762,
"transaction":null
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600375609365,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":603,
"lsn":34089688,
"xmin":null
},
"op":"u",
"ts_ms":1600375609778,
"transaction":null
}
}

બંને કિસ્સાઓમાં, રેકોર્ડ્સમાં બદલાયેલ રેકોર્ડની કી (PK) અને ફેરફારોનો ખૂબ જ સાર હોય છે: રેકોર્ડ પહેલા શું હતો અને તે પછી શું બન્યું.

• ના કિસ્સામાં INSERT: મૂલ્ય પહેલા (before) બરાબર null, અને પછી - જે લાઇન દાખલ કરવામાં આવી હતી.
• ના કિસ્સામાં UPDATE: અંતે payload.before લાઇનની પાછલી સ્થિતિ પ્રદર્શિત થાય છે, અને માં 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>.

સહનશીલતા દોષ

અમારા સમયમાં ખામી સહિષ્ણુતા અને ઉચ્ચ ઉપલબ્ધતાનો મુદ્દો પહેલા કરતાં વધુ તીવ્ર છે - ખાસ કરીને જ્યારે આપણે ડેટા અને વ્યવહારો વિશે વાત કરી રહ્યા છીએ, અને ડેટાના ફેરફારોને ટ્રૅક કરવાનું આ મુદ્દામાં બાજુ પર રહેતું નથી. ચાલો જોઈએ કે સિદ્ધાંતમાં શું ખોટું થઈ શકે છે અને દરેક કિસ્સામાં ડેબેઝિયમનું શું થશે.

ત્યાં ત્રણ નાપસંદ વિકલ્પો છે:

1. કાફકા કનેક્ટ નિષ્ફળતા. જો કનેક્ટને વિતરિત મોડમાં કામ કરવા માટે ગોઠવેલું હોય, તો આ માટે બહુવિધ કામદારોને સમાન group.id સેટ કરવાની જરૂર છે. પછી, જો તેમાંથી એક નિષ્ફળ જાય, તો કનેક્ટર બીજા કાર્યકર પર પુનઃપ્રારંભ થશે અને કાફકામાં વિષયમાં છેલ્લી પ્રતિબદ્ધ સ્થિતિમાંથી વાંચવાનું ચાલુ રાખશે.
2. કાફકા ક્લસ્ટર સાથે જોડાણ ગુમાવવું. કનેક્ટર ફક્ત તે સ્થાન પર વાંચવાનું બંધ કરશે જે કાફ્કાને મોકલવામાં નિષ્ફળ ગયું છે, અને પ્રયાસ સફળ થાય ત્યાં સુધી સમયાંતરે તેને ફરીથી મોકલવાનો પ્રયાસ કરશે.
3. ડેટા સ્ત્રોત અનુપલબ્ધતા. કનેક્ટર રૂપરેખાંકિત કર્યા મુજબ સ્રોત સાથે ફરીથી કનેક્ટ કરવાનો પ્રયાસ કરશે. ડિફૉલ્ટ 16 પ્રયાસો છે ઘાતાંકીય બેકઓફ. 16મા અસફળ પ્રયાસ પછી, કાર્ય તરીકે ચિહ્નિત કરવામાં આવશે નિષ્ફળ અને તમારે તેને કાફકા કનેક્ટ REST ઇન્ટરફેસ દ્વારા મેન્યુઅલી પુનઃપ્રારંભ કરવાની જરૂર પડશે.
   • ના કિસ્સામાં પોસ્ટગ્રે એસક્યુએલ ડેટા ખોવાઈ જશે નહીં, કારણ કે પ્રતિકૃતિ સ્લોટ્સનો ઉપયોગ તમને કનેક્ટર દ્વારા વાંચવામાં આવતી ન હોય તેવી WAL ફાઇલોને કાઢી નાખવાથી અટકાવશે. આ કિસ્સામાં, સિક્કાનું નુકસાન પણ છે: જો કનેક્ટર અને DBMS વચ્ચેની નેટવર્ક કનેક્ટિવિટી લાંબા સમય સુધી વિક્ષેપિત થાય છે, તો ડિસ્ક સ્પેસ સમાપ્ત થવાની સંભાવના છે, અને આ નિષ્ફળતા તરફ દોરી શકે છે. સમગ્ર DBMS.
   • ના કિસ્સામાં MySQL કનેક્ટિવિટી પુનઃસ્થાપિત થાય તે પહેલાં binlog ફાઇલોને DBMS દ્વારા જ ફેરવી શકાય છે. આનાથી કનેક્ટર નિષ્ફળ સ્થિતિમાં જશે, અને સામાન્ય કામગીરીને પુનઃસ્થાપિત કરવા માટે, તમારે બિનલોગ્સમાંથી વાંચવાનું ચાલુ રાખવા માટે પ્રારંભિક સ્નેપશોટ મોડમાં પુનઃપ્રારંભ કરવાની જરૂર પડશે.
   • પર MongoDB. દસ્તાવેજીકરણ જણાવે છે: લોગ/ઓપલોગ ફાઇલો કાઢી નાખવામાં આવી હોય અને કનેક્ટર જ્યાંથી છોડી દીધું હોય ત્યાંથી વાંચવાનું ચાલુ રાખી શકતું નથી તે ઘટનામાં કનેક્ટરનું વર્તન તમામ DBMS માટે સમાન છે. તેનો અર્થ એ છે કે કનેક્ટર રાજ્યમાં જશે નિષ્ફળ અને મોડમાં પુનઃપ્રારંભ કરવાની જરૂર પડશે પ્રારંભિક સ્નેપશોટ.

     જો કે, ત્યાં અપવાદો છે. જો કનેક્ટર લાંબા સમય સુધી ડિસ્કનેક્ટ થયેલું હતું (અથવા MongoDB દાખલા સુધી પહોંચી શક્યું ન હતું), અને આ સમય દરમિયાન oplog પરિભ્રમણમાંથી પસાર થયું હતું, તો પછી જ્યારે કનેક્શન પુનઃસ્થાપિત કરવામાં આવે છે, ત્યારે કનેક્ટર શાંતિથી પ્રથમ ઉપલબ્ધ સ્થિતિમાંથી ડેટા વાંચવાનું ચાલુ રાખશે, તેથી જ કાફકાના કેટલાક ડેટા નથી ફટકો પડશે.

નિષ્કર્ષ

ડેબેઝિયમ એ સીડીસી સિસ્ટમ્સ સાથેનો મારો પ્રથમ અનુભવ છે અને એકંદરે ખૂબ જ સકારાત્મક છે. મુખ્ય DBMS, રૂપરેખાંકનની સરળતા, ક્લસ્ટરિંગ સપોર્ટ અને સક્રિય સમુદાય માટે તેના સમર્થન સાથે પ્રોજેક્ટ જીત્યો. પ્રેક્ટિસમાં રસ ધરાવતા લોકો માટે, હું ભલામણ કરું છું કે તમે માટે માર્ગદર્શિકાઓ વાંચો કાફકા કનેક્ટ и ડેબેઝિયમ.

કાફકા કનેક્ટ માટેના JDBC કનેક્ટરની તુલનામાં, ડેબેઝિયમનો મુખ્ય ફાયદો એ છે કે ફેરફારોને DBMS લૉગ્સમાંથી વાંચવામાં આવે છે, જે ન્યૂનતમ લેટન્સી સાથે ડેટા પ્રાપ્ત કરવાની મંજૂરી આપે છે. JDBC કનેક્ટર (કાફ્કા કનેક્ટમાંથી) એક નિશ્ચિત અંતરાલ પર મોનિટર કરેલ કોષ્ટકની ક્વેરી કરે છે અને (તે જ કારણસર) જ્યારે ડેટા કાઢી નાખવામાં આવે ત્યારે સંદેશા જનરેટ કરતું નથી (તમે અસ્તિત્વમાં ન હોય તેવા ડેટાને કેવી રીતે ક્વેરી કરી શકો?).

સમાન સમસ્યાઓ હલ કરવા માટે, તમે નીચેના ઉકેલો પર ધ્યાન આપી શકો છો (ડેબેઝિયમ ઉપરાંત):

• JDBC કનેક્ટર કાફકા કનેક્ટ;
• માત્ર MySQL માટે કેટલાક ઉકેલો:
  • MySQL સ્ટ્રીમર;
  • મેક્સવેલ;
  • સ્પાઇનલટેપ;
• ઓરેકલ ગોલ્ડનગેટ, પરંતુ આ એક સંપૂર્ણપણે અલગ "વજન શ્રેણી" છે.

PS

અમારા બ્લોગ પર પણ વાંચો:

• «કુબરનેટીસમાં કાફકા ક્લસ્ટર માટે યોગ્ય કદ નક્કી કરવું»;
• «અમારા SRE રોજિંદા જીવનમાંથી વ્યવહારુ વાર્તાઓ. ભાગ 2»;
• «કુબરનેટ્સ, અમારી પસંદગીઓ અને અનુભવો માટે પોસ્ટગ્રેએસક્યુએલ નિવેદનોની સંક્ષિપ્ત ઝાંખી».

સોર્સ: www.habr.com