Source এবং Sink Connectors কনফিগার করা

Apache Kafka Connect একটি শক্তিশালী টুল যা ডেটা শিফট এবং ইন্টিগ্রেশন প্রক্রিয়া সহজ করে। এটি Source এবং Sink Connectors ব্যবহার করে বিভিন্ন ডেটা স্টোর (যেমন, ডেটাবেস, ফাইল সিস্টেম, কনফিগারেশন সার্ভিস) এর সাথে কাফকা সংযোগ স্থাপন করতে সাহায্য করে। এই কনেক্টরগুলি ব্যবহার করে আপনি Kafka ক্লাস্টারের মধ্যে ডেটা ইনপুট এবং আউটপুট পরিচালনা করতে পারেন।

Source Connector

Source Connector কাফকা ক্লাস্টারের জন্য ডেটা সরবরাহ করে। এটি এক বা একাধিক ডেটা সোর্স (যেমন, ডেটাবেস, ফাইল সিস্টেম, লগ ফাইল) থেকে ডেটা সংগ্রহ করে এবং Kafka টপিকে পাঠায়। এটির মাধ্যমে আপনি সহজেই কাফকায় ডেটা ইনজেস্ট করতে পারেন।

Source Connector কনফিগারেশন

Source Connector কনফিগার করতে কিছু স্টেপ অনুসরণ করতে হয়। এখানে সাধারণভাবে JDBC Source Connector ব্যবহার করে ডেটাবেস থেকে ডেটা Kafka তে পাঠানোর উদাহরণ দেখানো হলো।

1. Connector Configuration: প্রথমে Kafka Connect এর কনফিগারেশন ফাইলে JDBC Source Connector এর জন্য কনফিগারেশন সেট করতে হবে।

   JDBC Source Connector Configuration Example:

   {
     "name": "jdbc-source-connector",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
       "tasks.max": "1",
       "topics": "db-topic",
       "connection.url": "jdbc:mysql://localhost:3306/mydb",
       "connection.user": "root",
       "connection.password": "password",
       "table.whitelist": "employees",
       "mode": "incrementing",
       "incrementing.column.name": "employee_id"
     }
   }
   

   কনফিগারেশন ব্যাখ্যা:

   • "connector.class": এটি connector এর class path, যা নির্দিষ্ট করে দেয় কনফিগারেশনের type (এখানে JDBC source connector ব্যবহার করা হয়েছে)।
   • "connection.url": ডেটাবেসের URL।
   • "table.whitelist": কোন টেবিল থেকে ডেটা পেতে হবে।
   • "mode": ডেটা সংগ্রহের ধরন (যেমন, "incrementing" মানে ইনক্রিমেন্টাল ডেটা সংগ্রহ)।
   • "topics": কাফকা টপিক যেখানে ডেটা পাঠানো হবে।

2. Kafka Connect চলানো: এখন কনফিগারেশন ফাইল তৈরি হলে, Kafka Connect রেস্ট API ব্যবহার করে এই কনফিগারেশন প্রোপার্টি পাঠানো হয় এবং কননেক্টরটি চালু করা হয়।

   curl -X POST -H "Content-Type: application/json" --data @jdbc-source-connector.json http://localhost:8083/connectors
   

   এই কমান্ডটি Kafka Connect সার্ভারে কনফিগারেশন পুশ করবে এবং jdbc-source-connector নামক কনেক্টর চালু হবে।

Sink Connector

Sink Connector Kafka থেকে ডেটা গ্রহণ করে এবং তা অন্য ডেটা স্টোরে (যেমন, ডেটাবেস, ফাইল সিস্টেম, ক্লাউড স্টোরেজ) সংরক্ষণ করে। এটি সাধারণত ডেটা আউটপুট করার জন্য ব্যবহৃত হয়।

Sink Connector কনফিগারেশন

Sink Connector কনফিগার করার জন্য Kafka Connect কনফিগারেশন ফাইলের মাধ্যমে ডেটা ড্রেইন করতে হয়। নিচে JDBC Sink Connector ব্যবহার করে Kafka টপিক থেকে ডেটাবেসে ডেটা পাঠানোর কনফিগারেশন দেওয়া হলো।

1. Connector Configuration: JDBC Sink Connector এর কনফিগারেশন:

   JDBC Sink Connector Configuration Example:

   {
     "name": "jdbc-sink-connector",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
       "tasks.max": "1",
       "topics": "db-topic",
       "connection.url": "jdbc:mysql://localhost:3306/mydb",
       "connection.user": "root",
       "connection.password": "password",
       "insert.mode": "insert",
       "table.name.format": "employees_sink",
       "auto.create": "true"
     }
   }
   

   কনফিগারেশন ব্যাখ্যা:

   • "connector.class": এই প্রোপার্টি দ্বারা সিলেক্ট করা হয় Sink Connector এর class path (এখানে JDBC Sink ব্যবহার করা হয়েছে)।
   • "topics": Kafka টপিক থেকে ডেটা নেয়া হবে।
   • "table.name.format": ডেটা যেখানেই ইনসার্ট করা হবে, সেই টেবিলের নাম।
   • "insert.mode": ডেটা ইনসার্টের ধরন, যেমন insert বা upsert।
   • "auto.create": যদি টেবিল না থাকে, তবে এটি সিস্টেমকে টেবিল তৈরি করতে বলে।

2. Kafka Connect চালানো: কনফিগারেশন ফাইল তৈরি হলে, আপনি আবার Kafka Connect রেস্ট API ব্যবহার করে কনফিগারেশনটি পুশ করবেন।

   curl -X POST -H "Content-Type: application/json" --data @jdbc-sink-connector.json http://localhost:8083/connectors
   

   এই কমান্ডটি Kafka Connect সার্ভারে কনফিগারেশন পুশ করবে এবং jdbc-sink-connector চালু হবে।

Source এবং Sink Connector এর পার্থক্য

1. Source Connector:
   • এটি ডেটা সংগ্রহ করে এবং Kafka টপিকে পাঠায়।
   • ডেটা সোর্স হতে পারে ডেটাবেস, ফাইল সিস্টেম, ইত্যাদি।
2. Sink Connector:
   • এটি Kafka টপিক থেকে ডেটা নিয়ে তা অন্য ডেটাবেস বা স্টোরেজ সিস্টেম এ পাঠায়।
   • ডেটা ড্রেইন করতে ব্যবহৃত হয়, যেমন ডেটাবেসে ডেটা ইনসার্ট করা।

সারাংশ

Source এবং Sink Connectors Kafka Connect এ অত্যন্ত গুরুত্বপূর্ণ ভূমিকা পালন করে। Source Connectors ডেটা সংগ্রহ করে Kafka টপিকে ইনপুট পাঠায় এবং Sink Connectors Kafka টপিক থেকে ডেটা নিয়ে অন্য ডেটাবেস বা সিস্টেমে সংরক্ষণ করে। তাদের কনফিগারেশন সহজে JSON ফরম্যাটে করা যায় এবং Kafka Connect API ব্যবহার করে কনফিগারেশন পুশ করে এটি চালানো যায়। এই কনফিগারেশন সিস্টেম ডেটা ইন্টিগ্রেশন এবং ডিস্ট্রিবিউটেড সিস্টেমের মধ্যে ডেটা ট্রান্সফার প্রক্রিয়া সহজ করে।