Apache Kafka-এর জন্য Debezium - CDC উপস্থাপন করা হচ্ছে

আমার কাজে, আমি প্রায়শই নতুন প্রযুক্তিগত সমাধান / সফ্টওয়্যার পণ্যগুলি দেখতে পাই, যার সম্পর্কে তথ্য রাশিয়ান-ভাষী ইন্টারনেটে খুব কম। এই নিবন্ধটির মাধ্যমে, আমি আমার সাম্প্রতিক অনুশীলন থেকে একটি উদাহরণ দিয়ে এমন একটি শূন্যতা পূরণ করার চেষ্টা করব, যখন আমাকে দুটি জনপ্রিয় DBMSs (PostgreSQL এবং MongoDB) থেকে Debezium ব্যবহার করে একটি কাফকা ক্লাস্টারে CDC ইভেন্ট পাঠানোর সেট আপ করতে হয়েছিল। আমি আশা করি যে এই পর্যালোচনা নিবন্ধটি, যা কাজের ফলাফল হিসাবে উপস্থিত হয়েছিল, অন্যদের জন্য দরকারী হবে।

সাধারণভাবে Debezium এবং CDC কি?

ডেবেজিয়াম - সিডিসি সফ্টওয়্যার বিভাগের প্রতিনিধি (তথ্য পরিবর্তন ক্যাপচার), বা আরও স্পষ্টভাবে, এটি বিভিন্ন DBMS-এর জন্য সংযোগকারীর একটি সেট যা Apache Kafka Connect ফ্রেমওয়ার্কের সাথে সামঞ্জস্যপূর্ণ।

এই ওপেন সোর্স প্রকল্প, Apache License v2.0 এর অধীনে লাইসেন্সপ্রাপ্ত এবং Red Hat দ্বারা স্পনসর করা হয়েছে। 2016 সাল থেকে উন্নয়ন চলছে এবং এই মুহুর্তে এটি নিম্নলিখিত DBMS-এর জন্য সরকারী সহায়তা প্রদান করে: MySQL, PostgreSQL, MongoDB, SQL সার্ভার। ক্যাসান্ড্রা এবং ওরাকলের জন্যও সংযোগকারী রয়েছে, কিন্তু তারা বর্তমানে "আর্লি অ্যাক্সেস" অবস্থায় রয়েছে এবং নতুন রিলিজগুলি পশ্চাদপদ সামঞ্জস্যের গ্যারান্টি দেয় না।

যদি আমরা সিডিসিকে ঐতিহ্যগত পদ্ধতির সাথে তুলনা করি (যখন অ্যাপ্লিকেশনটি সরাসরি ডিবিএমএস থেকে ডেটা পড়ে), তবে এর প্রধান সুবিধাগুলির মধ্যে রয়েছে কম বিলম্ব, উচ্চ নির্ভরযোগ্যতা এবং প্রাপ্যতা সহ সারি স্তরে ডেটা পরিবর্তন স্ট্রিমিং বাস্তবায়ন। CDC ইভেন্টগুলির জন্য একটি সংগ্রহস্থল হিসাবে কাফকা ক্লাস্টার ব্যবহার করে শেষ দুটি পয়েন্ট অর্জন করা হয়।

এছাড়াও, সুবিধার মধ্যে রয়েছে যে একটি একক মডেল ইভেন্টগুলি সঞ্চয় করতে ব্যবহৃত হয়, তাই চূড়ান্ত অ্যাপ্লিকেশনটিকে বিভিন্ন ডিবিএমএস পরিচালনার সূক্ষ্মতা সম্পর্কে চিন্তা করতে হবে না।

অবশেষে, একটি বার্তা ব্রোকার ব্যবহার করা অ্যাপ্লিকেশনগুলির অনুভূমিক স্কেলিং করার সুযোগ খুলে দেয় যা ডেটাতে পরিবর্তনগুলি ট্র্যাক করে। একই সময়ে, ডেটা উত্সের উপর প্রভাব হ্রাস করা হয়, যেহেতু ডেটা সরাসরি ডিবিএমএস থেকে নয়, কাফকা ক্লাস্টার থেকে প্রাপ্ত হয়।

ডেবেজিয়াম স্থাপত্য সম্পর্কে

Debezium ব্যবহার করা এই সহজ স্কিমে নেমে আসে:

DBMS (ডেটা উৎস হিসেবে) → কাফকা কানেক্টে সংযোগকারী → অ্যাপাচি কাফকা → ভোক্তা

একটি উদাহরণ হিসাবে, আমি প্রকল্পের ওয়েবসাইট থেকে একটি ডায়াগ্রাম দেব:

যাইহোক, আমি এই স্কিমটি সত্যিই পছন্দ করি না, কারণ মনে হচ্ছে শুধুমাত্র একটি সিঙ্ক সংযোগকারী সম্ভব।

বাস্তবে, পরিস্থিতি ভিন্ন: আপনার ডেটা লেক ভরাট করা (উপরের চিত্রের শেষ লিঙ্ক) Debezium ব্যবহার করার একমাত্র উপায় নয়। Apache Kafka-তে পাঠানো ইভেন্টগুলি বিভিন্ন পরিস্থিতিতে মোকাবেলা করার জন্য আপনার অ্যাপ্লিকেশনগুলি ব্যবহার করতে পারে। উদাহরণ স্বরূপ:

• ক্যাশে থেকে অপ্রাসঙ্গিক তথ্য অপসারণ;
• বিজ্ঞপ্তি পাঠানো;
• অনুসন্ধান সূচক আপডেট;
• কিছু ধরনের অডিট লগ;
• ...

আপনার যদি একটি জাভা অ্যাপ্লিকেশন থাকে এবং কাফকা ক্লাস্টার ব্যবহার করার কোন প্রয়োজন/সম্ভাবনা না থাকে, তাহলে এর মাধ্যমে কাজ করার সম্ভাবনাও রয়েছে এমবেডেড সংযোগকারী. সুস্পষ্ট প্লাস হল যে এটির সাথে আপনি অতিরিক্ত অবকাঠামো (একটি সংযোগকারী এবং কাফকার আকারে) প্রত্যাখ্যান করতে পারেন। যাইহোক, এই সমাধানটি সংস্করণ 1.1 থেকে অবমূল্যায়িত করা হয়েছে এবং এটি আর ব্যবহারের জন্য সুপারিশ করা হয় না (এটি ভবিষ্যতের প্রকাশগুলিতে সরানো হতে পারে)।

এই নিবন্ধটি বিকাশকারীদের দ্বারা সুপারিশকৃত আর্কিটেকচার নিয়ে আলোচনা করবে, যা ত্রুটি সহনশীলতা এবং মাপযোগ্যতা প্রদান করে।

সংযোগকারী কনফিগারেশন

সবচেয়ে গুরুত্বপূর্ণ মান - ডেটা - এর পরিবর্তনগুলি ট্র্যাক করা শুরু করার জন্য আমাদের প্রয়োজন:

1. ডেটা উৎস, যা MySQL হতে পারে সংস্করণ 5.7 থেকে শুরু করে, PostgreSQL 9.6+, MongoDB 3.2+ (সম্পূর্ণ তালিকা);
2. অ্যাপাচি কাফকা ক্লাস্টার
3. কাফকা কানেক্ট ইনস্ট্যান্স (সংস্করণ 1.x, 2.x);
4. কনফিগার করা Debezium সংযোগকারী.

প্রথম দুটি পয়েন্টে কাজ করুন, যেমন একটি DBMS এবং Apache Kafka ইনস্টল করার প্রক্রিয়া নিবন্ধের সুযোগের বাইরে। যাইহোক, যারা একটি স্যান্ডবক্সে সবকিছু স্থাপন করতে চান, তাদের জন্য অফিসিয়াল রিপোজিটরিতে উদাহরণ সহ একটি রেডিমেড রয়েছে docker-compose.yaml.

আমরা আরও বিশদে শেষ দুটি পয়েন্টে ফোকাস করব।

0. কাফকা কানেক্ট

এখানে এবং পরে নিবন্ধে, সমস্ত কনফিগারেশন উদাহরণ Debezium ডেভেলপারদের দ্বারা বিতরণ করা ডকার চিত্রের প্রসঙ্গে বিবেচনা করা হয়েছে। এটিতে সমস্ত প্রয়োজনীয় প্লাগইন ফাইল (সংযোগকারী) রয়েছে এবং পরিবেশ ভেরিয়েবল ব্যবহার করে কাফকা কানেক্ট কনফিগারেশন প্রদান করে।

আপনি যদি কনফ্লুয়েন্ট থেকে কাফকা কানেক্ট ব্যবহার করতে চান, তাহলে আপনাকে প্রয়োজনীয় কানেক্টরের প্লাগইনগুলিকে নির্দিষ্ট ডিরেক্টরিতে যোগ করতে হবে 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

অভ্র সম্পর্কে নোট

ডিফল্টরূপে, Debezium JSON ফর্ম্যাটে ডেটা লেখে, যা স্যান্ডবক্স এবং অল্প পরিমাণ ডেটার জন্য গ্রহণযোগ্য, কিন্তু ভারী লোড করা ডেটাবেসে সমস্যা হতে পারে। JSON রূপান্তরকারীর একটি বিকল্প হল ব্যবহার করে বার্তাগুলিকে সিরিয়ালাইজ করা অভ্র একটি বাইনারি বিন্যাসে, যা Apache Kafka-তে 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 — সংযোগকারীর নাম যার জন্য নীচে বর্ণিত কনফিগারেশন ব্যবহার করা হয়েছে; ভবিষ্যতে, এই নামটি কাফকা কানেক্ট REST API-এর মাধ্যমে সংযোগকারীর সাথে কাজ করতে ব্যবহার করা হবে (অর্থাৎ স্থিতি দেখুন / পুনরায় চালু করুন / কনফিগারেশন আপডেট করুন);
• connector.class — কনফিগার করা সংযোগকারী দ্বারা ব্যবহৃত DBMS সংযোগকারী শ্রেণী;
• plugin.name WAL ফাইল থেকে ডেটার যৌক্তিক ডিকোডিংয়ের জন্য প্লাগইনটির নাম। থেকে চয়ন করার জন্য উপলব্ধ wal2json, decoderbuffs и pgoutput. প্রথম দুটির জন্য ডিবিএমএসে উপযুক্ত এক্সটেনশন ইনস্টল করা প্রয়োজন, এবং pgoutput PostgreSQL সংস্করণ 10 এবং উচ্চতর জন্য অতিরিক্ত ম্যানিপুলেশন প্রয়োজন হয় না;
• database.* — ডাটাবেসের সাথে সংযোগ করার বিকল্প, যেখানে database.server.name - কাফকা ক্লাস্টারে বিষয়ের নাম গঠন করতে ব্যবহৃত PostgreSQL উদাহরণের নাম;
• table.include.list - টেবিলের একটি তালিকা যেখানে আমরা পরিবর্তনগুলি ট্র্যাক করতে চাই; বিন্যাসে দেওয়া schema.table_name; সঙ্গে একসঙ্গে ব্যবহার করা যাবে না table.exclude.list;
• heartbeat.interval.ms - ব্যবধান (মিলিসেকেন্ডে) যার সাথে সংযোগকারী একটি বিশেষ বিষয়ে হার্টবিট বার্তা পাঠায়;
• heartbeat.action.query - একটি অনুরোধ যা প্রতিটি হার্টবিট বার্তা পাঠানোর সময় কার্যকর করা হবে (বিকল্পটি সংস্করণ 1.1 থেকে উপস্থিত হয়েছে);
• slot.name — প্রতিলিপি স্লটের নাম যা সংযোগকারী দ্বারা ব্যবহৃত হবে;
• publication.name - নাম প্রকাশন PostgreSQL-এ যা সংযোগকারী ব্যবহার করে। যদি এটি বিদ্যমান না থাকে, Debezium এটি তৈরি করার চেষ্টা করবে। যে ব্যবহারকারীর অধীনে সংযোগটি করা হয়েছে তার যদি এই ক্রিয়াকলাপের জন্য পর্যাপ্ত অধিকার না থাকে তবে সংযোগকারী একটি ত্রুটি সহ প্রস্থান করবে;
• transforms লক্ষ্য বিষয়ের নাম ঠিক কিভাবে পরিবর্তন করতে হবে তা নির্ধারণ করে:
  • transforms.AddPrefix.type নির্দেশ করে যে আমরা রেগুলার এক্সপ্রেশন ব্যবহার করব;
  • transforms.AddPrefix.regex — মুখোশ যার দ্বারা লক্ষ্য বিষয়ের নাম পুনরায় সংজ্ঞায়িত করা হয়েছে;
  • transforms.AddPrefix.replacement - সরাসরি আমরা কি পুনরায় সংজ্ঞায়িত করি।

হার্টবিট এবং রূপান্তর সম্পর্কে আরও

ডিফল্টরূপে, সংযোগকারী প্রতিটি প্রতিশ্রুতিবদ্ধ লেনদেনের জন্য কাফকার কাছে ডেটা পাঠায় এবং পরিষেবার বিষয়ে তার LSN (লগ সিকোয়েন্স নম্বর) লিখে দেয়। offset. তবে কি হবে যদি সংযোগকারীটি সম্পূর্ণ ডাটাবেস পড়ার জন্য কনফিগার করা হয় না, তবে শুধুমাত্র তার টেবিলের একটি অংশ (যাতে ডেটা কদাচিৎ আপডেট করা হয়)?

• সংযোগকারী WAL ফাইলগুলি পড়বে এবং সেগুলির মধ্যে লেনদেনের প্রতিশ্রুতি সনাক্ত করবে না যে টেবিলগুলি এটি নিরীক্ষণ করে৷
• অতএব, এটি বিষয় বা প্রতিলিপি স্লটে তার বর্তমান অবস্থান আপডেট করবে না।
• এর ফলে, WAL ফাইলগুলি ডিস্কে "আটকে" যাবে এবং সম্ভবত ডিস্কের স্থান ফুরিয়ে যাবে।

এবং এখানে বিকল্পগুলি উদ্ধার করতে আসে। heartbeat.interval.ms и heartbeat.action.query. জোড়ায় জোড়ায় এই বিকল্পগুলি ব্যবহার করে প্রতিবার হার্টবিট মেসেজ পাঠানো হলে একটি পৃথক টেবিলে ডেটা পরিবর্তন করার অনুরোধ কার্যকর করা সম্ভব করে। এইভাবে, যে LSNটিতে সংযোগকারীটি বর্তমানে অবস্থিত (প্রতিলিপি স্লটে) তা ক্রমাগত আপডেট করা হয়। এটি DBMS কে WAL ফাইলগুলি সরাতে দেয় যা আর প্রয়োজন নেই৷ বিকল্পগুলি কীভাবে কাজ করে সে সম্পর্কে আরও তথ্যের জন্য, দেখুন ডকুমেন্টেশন.

ঘনিষ্ঠ মনোযোগ প্রাপ্য আরেকটি বিকল্প হয় transforms. যদিও এটি সুবিধা এবং সৌন্দর্য সম্পর্কে আরও ...

ডিফল্টরূপে, Debezium নিম্নলিখিত নামকরণ নীতি ব্যবহার করে বিষয় তৈরি করে: serverName.schemaName.tableName. এটি সবসময় সুবিধাজনক নাও হতে পারে। অপশন transforms রেগুলার এক্সপ্রেশন ব্যবহার করে, আপনি টেবিলের একটি তালিকা সংজ্ঞায়িত করতে পারেন যার ইভেন্টগুলিকে একটি নির্দিষ্ট নামের সাথে একটি বিষয়ে রাউট করতে হবে।

আমাদের কনফিগারেশন ধন্যবাদ transforms নিম্নলিখিতটি ঘটে: ট্র্যাক করা ডাটাবেস থেকে সমস্ত CDC ইভেন্ট নাম সহ টপিকে যাবে data.cdc.dbname. অন্যথায় (এই সেটিংস ছাড়া), Debezium ডিফল্টরূপে ফর্মের প্রতিটি টেবিলের জন্য একটি বিষয় তৈরি করবে: pg-dev.public.<table_name>.

সংযোগকারীর সীমাবদ্ধতা

PostgreSQL এর জন্য সংযোগকারী কনফিগারেশনের বর্ণনার শেষে, এটির কাজের নিম্নলিখিত বৈশিষ্ট্য / সীমাবদ্ধতা সম্পর্কে কথা বলা মূল্যবান:

1. PostgreSQL-এর সংযোগকারী কার্যকারিতা লজিক্যাল ডিকোডিংয়ের ধারণার উপর নির্ভর করে। তাই সে ডাটাবেসের গঠন পরিবর্তন করার অনুরোধ ট্র্যাক করে না (DDL) - সেই অনুযায়ী, এই ডেটা বিষয়গুলিতে থাকবে না।
2. যেহেতু প্রতিলিপি স্লট ব্যবহার করা হয়, সংযোগকারীর সংযোগ সম্ভব শুধুমাত্র মাস্টার ডিবিএমএস উদাহরণে।
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', 'foo@bar.com');
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":"foo@bar.com"
    },
    "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":"foo@bar.com"
    },
    "after":{
      "id":1005,
      "first_name":"egg",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "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 মঙ্গোডিবি

এই সংযোগকারীটি DBMS প্রাইমারি নোডের অপলগ থেকে তথ্য পড়া স্ট্যান্ডার্ড MongoDB রেপ্লিকেশন মেকানিজম ব্যবহার করে।

একইভাবে 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 সেট করতে একাধিক কর্মীদের প্রয়োজন। তারপর, যদি তাদের মধ্যে একটি ব্যর্থ হয়, সংযোগকারী অন্য কর্মীর উপর পুনরায় চালু হবে এবং Kafka বিষয়ের শেষ প্রতিশ্রুতিবদ্ধ অবস্থান থেকে পড়া চালিয়ে যাবে।
2. কাফকা ক্লাস্টারের সাথে সংযোগ বিচ্ছিন্ন. সংযোগকারীটি কাফকাকে পাঠাতে ব্যর্থ হওয়া অবস্থানে পড়া বন্ধ করবে এবং প্রচেষ্টা সফল না হওয়া পর্যন্ত পর্যায়ক্রমে এটি পুনরায় পাঠানোর চেষ্টা করবে।
3. ডেটা উৎস অনুপলব্ধ. সংযোগকারী কনফিগারেশন অনুযায়ী উৎসের সাথে পুনরায় সংযোগ করার চেষ্টা করবে। ডিফল্ট হল 16টি ব্যবহার করে প্রচেষ্টা সূচকীয় ব্যাকঅফ. 16 তম ব্যর্থ প্রচেষ্টার পরে, টাস্কটি হিসাবে চিহ্নিত করা হবে৷ ব্যর্থ এবং এটি কাফকা কানেক্ট REST ইন্টারফেসের মাধ্যমে ম্যানুয়ালি রিস্টার্ট করতে হবে।
   • ক্ষেত্রে পোস্টগ্রি ডেটা হারিয়ে যাবে না, কারণ প্রতিলিপি স্লট ব্যবহার করে সংযোগকারী দ্বারা পড়া হয়নি এমন WAL ফাইল মুছে ফেলা প্রতিরোধ করবে। এই ক্ষেত্রে, একটি নেতিবাচক দিক রয়েছে: যদি সংযোগকারী এবং DBMS-এর মধ্যে নেটওয়ার্ক সংযোগ দীর্ঘ সময়ের জন্য ব্যাহত হয়, তাহলে ডিস্কের স্থান ফুরিয়ে যাওয়ার সম্ভাবনা রয়েছে এবং এটি সম্পূর্ণ DBMS-এর ব্যর্থতার দিকে নিয়ে যেতে পারে।
   • ক্ষেত্রে মাইএসকিউএল সংযোগ পুনরুদ্ধার করার আগে binlog ফাইলগুলি DBMS নিজেই ঘোরানো যেতে পারে। এর ফলে সংযোগকারী ব্যর্থ অবস্থায় চলে যাবে এবং স্বাভাবিক ক্রিয়াকলাপ পুনরুদ্ধার করতে বিনলগ থেকে পড়া চালিয়ে যেতে এটিকে প্রাথমিক স্ন্যাপশট মোডে পুনরায় চালু করতে হবে।
   • উপর MongoDB. ডকুমেন্টেশনে বলা হয়েছে: লগ/অপলগ ফাইলগুলি মুছে ফেলার ক্ষেত্রে সংযোগকারীর আচরণ এবং সংযোগকারী যে অবস্থান থেকে পড়া চালিয়ে যেতে পারে না তা সব DBMS-এর জন্য একই। এটা সত্য যে সংযোগকারী রাষ্ট্র যেতে হবে ব্যর্থ এবং মোডে একটি পুনরায় চালু করতে হবে প্রাথমিক স্ন্যাপশট.

     যাইহোক, ব্যতিক্রম আছে. যদি সংযোগকারীটি দীর্ঘ সময়ের জন্য সংযোগ বিচ্ছিন্ন অবস্থায় থাকে (বা মঙ্গোডিবি উদাহরণে পৌঁছাতে পারেনি), এবং এই সময়ে অপলগটি ঘোরানো হয়েছিল, তখন সংযোগটি পুনরুদ্ধার করা হলে, সংযোগকারী শান্তভাবে প্রথম উপলব্ধ অবস্থান থেকে ডেটা পড়তে থাকবে। , যে কারণে কাফকার কিছু তথ্য না আঘাত করবে

উপসংহার

Debezium হল CDC সিস্টেমের সাথে আমার প্রথম অভিজ্ঞতা এবং সামগ্রিকভাবে খুবই ইতিবাচক। প্রকল্পটি প্রধান DBMS-এর সমর্থন, কনফিগারেশনের সহজতা, ক্লাস্টারিংয়ের জন্য সমর্থন এবং একটি সক্রিয় সম্প্রদায়কে ঘুষ দিয়েছে। অনুশীলনে আগ্রহীদের জন্য, আমি সুপারিশ করছি যে আপনি গাইডগুলি পড়ুন কাফকা কানেক্ট и ডেবেজিয়াম.

কাফকা সংযোগের জন্য JDBC সংযোগকারীর তুলনায়, Debezium-এর প্রধান সুবিধা হল যে পরিবর্তনগুলি DBMS লগগুলি থেকে পড়া হয়, যা ন্যূনতম বিলম্বের সাথে ডেটা গ্রহণ করার অনুমতি দেয়। JDBC সংযোগকারী (কাফকা কানেক্ট দ্বারা সরবরাহ করা) একটি নির্দিষ্ট ব্যবধানে ট্র্যাক করা টেবিলটি জিজ্ঞাসা করে এবং (একই কারণে) ডেটা মুছে ফেলার সময় বার্তা তৈরি করে না (যেটি নেই এমন ডেটার জন্য আপনি কীভাবে অনুসন্ধান করতে পারেন?)।

অনুরূপ সমস্যা সমাধানের জন্য, আপনি নিম্নলিখিত সমাধানগুলিতে মনোযোগ দিতে পারেন (ডেবেজিয়াম ছাড়াও):

• JDBC কানেক্টর কাফকা কানেক্ট
• কিছু MySQL-শুধু সমাধান:
  • মাইএসকিউএল স্ট্রীমার;
  • ম্যাক্সওয়েল;
  • মেরুদণ্ডের আংটা;
• ওরাকল গোল্ডেনগেট, কিন্তু এটি একটি সম্পূর্ণ ভিন্ন "ওজন বিভাগ"।

দ্রষ্টব্য

আমাদের ব্লগেও পড়ুন:

• «Kubernetes-এ একটি কাফকা ক্লাস্টারের জন্য উপযুক্ত আকার নির্ধারণ করুন";
• «আমাদের SRE দৈনন্দিন জীবন থেকে ব্যবহারিক গল্প. অংশ ২";
• «Kubernetes, আমাদের পছন্দ এবং অভিজ্ঞতার জন্য PostgreSQL বিবৃতিগুলির একটি সংক্ষিপ্ত বিবরণ».

উত্স: www.habr.com