Apache Kafka Streams হল একটি শক্তিশালী লাইব্রেরি যা Kafka থেকে আসা ডেটা স্ট্রিম প্রসেসিংয়ের জন্য ব্যবহৃত হয়। যখন Kafka Streams-এর সাথে Avro ফরম্যাট ব্যবহার করা হয়, তখন ডেটা স্ট্রিম প্রসেসিং আরও কার্যকরী, স্কেলেবল এবং ইন্টিগ্রেটেড হয়। Avro-এর প্রধান সুবিধাগুলি যেমন স্কিমা ইভোলিউশন, কম্প্রেশন এবং সঠিক ডেটা সিরিয়ালাইজেশন Kafka Streams-এর সাথে যুক্ত হলে শক্তিশালী ডেটা প্রসেসিং সিস্টেম গঠন করা যায়।
Avro-এর স্কিমা ভিত্তিক ডিজাইন Kafka স্ট্রিমসে ব্যবহৃত ডেটার গঠন নির্ধারণ করতে সহায়তা করে, যা স্ট্রিম প্রসেসিংয়ের ক্ষেত্রে ডেটার সঠিকতা নিশ্চিত করে এবং স্কিমা ইভোলিউশনের ফলে ডেটার পরিবর্তনেও সিস্টেম কার্যকরী থাকে।
Kafka Streams এবং Avro Integration কেন গুরুত্বপূর্ণ?
- ডেটা স্কিমা বজায় রাখা: Avro-এর মাধ্যমে আপনি ডেটার স্কিমা নিয়ন্ত্রণ করতে পারেন, যার ফলে স্ট্রিম ডেটা প্রসেসিংয়ের সময় ডেটা স্ট্রাকচার বজায় থাকে।
- সেন্সিটিভ ডেটার নিরাপত্তা: Avro ব্যবহার করে ডেটার সিরিয়ালাইজেশন এবং ডেসিরিয়ালাইজেশন নিশ্চিত করা হয়, যা নিরাপদভাবে ডেটা স্ট্রিম প্রক্রিয়া করতে সাহায্য করে।
- স্কিমা ইভোলিউশন: Avro স্কিমা ইভোলিউশনের সাথে Kafka Streams-এর মাধ্যমে ডেটা প্রসেসিং সিস্টেমে কোনো ধরনের বিভ্রান্তি তৈরি হয় না। স্কিমার পরিবর্তন এবং নতুন ফিল্ড যোগ করা সহজে সমর্থিত হয়।
- কম্প্রেশন সুবিধা: Avro ফরম্যাটের সাথে বিভিন্ন কম্প্রেশন স্কিমা যেমন Snappy এবং Deflate ব্যবহার করলে ডেটা প্রসেসিং আরও দ্রুত এবং কম্প্যাক্ট হতে পারে।
Kafka Streams এবং Avro Integration কিভাবে কাজ করে?
Kafka Streams-এ Avro ইনটিগ্রেট করার জন্য সাধারণত Avro Serializer এবং Deserializer ব্যবহার করতে হয়, যা Kafka-কে Avro ফরম্যাটে ডেটা পাঠানোর এবং গ্রহণ করার সক্ষমতা প্রদান করে। এছাড়া, Schema Registry ব্যবহার করে স্কিমার রেজিস্ট্রেশন এবং ম্যানেজমেন্ট করতে হয়, যা ডেটার গঠন এবং ভার্সন ট্র্যাকিং নিশ্চিত করে।
১. Avro স্কিমা রেজিস্ট্রেশন
Kafka Streams-এ Avro ডেটা ব্যবহার করতে হলে, প্রথমে একটি স্কিমা রেজিস্টার করতে হয়। এর জন্য Schema Registry ব্যবহার করা হয়, যেখানে Avro স্কিমা সংরক্ষিত থাকে এবং Kafka-র মাধ্যমে ডেটা প্রেরণ বা গ্রহণের সময় Avro স্কিমা ব্যবহার হয়।
Schema Registry সেটআপ:
curl -X POST -H "Content-Type: application/json" \
--data @user-schema.avsc \
http://localhost:8081/subjects/user/versions
এখানে user-schema.avsc হল Avro স্কিমা ফাইল এবং localhost:8081 হল Schema Registry সার্ভারের URL।
২. Avro Serializer এবং Deserializer কনফিগারেশন
Kafka Streams-এ ডেটা প্রেরণের জন্য AvroSerializer এবং AvroDeserializer ব্যবহার করতে হয়। এই Serializer এবং Deserializer ডেটাকে Avro ফরম্যাটে সিরিয়ালাইজ এবং ডেসিরিয়ালাইজ করতে সাহায্য করে।
Kafka Streams Avro Serializer/Deserializer কনফিগারেশন:
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Properties;
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "avro-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put("schema.registry.url", "http://localhost:8081");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, KafkaAvroSerializer.class);
এই কনফিগারেশনটি Kafka Streams অ্যাপ্লিকেশনকে Avro সিরিয়ালাইজেশন এবং ডেসিরিয়ালাইজেশন সক্ষম করে। schema.registry.url প্রপার্টি Schema Registry-এর অবস্থান নির্ধারণ করে।
৩. Kafka Streams এ Avro ডেটা প্রসেসিং
Kafka Streams ব্যবহার করে আপনি Avro ডেটা প্রসেস করতে পারেন, যেমন ডেটা ট্রান্সফর্মেশন, অ্যাগ্রিগেশন এবং ফিল্টারিং ইত্যাদি।
Kafka Streams Avro ডেটা প্রসেসিং উদাহরণ:
KStream<String, GenericRecord> stream = builder.stream("input-topic");
stream.mapValues(value -> {
// আপনার লজিক এখানে
String name = value.get("name").toString();
int age = (int) value.get("age");
return new GenericData.Record(value.getSchema()); // নতুন Avro রেকর্ড তৈরি
}).to("output-topic", Produced.with(Serdes.String(), new KafkaAvroSerializer()));
এখানে, mapValues() মেথডের মাধ্যমে ডেটার উপর অপারেশন করা হচ্ছে এবং পরবর্তী স্টেপে তা অন্য Kafka টপিকে প্রেরণ করা হচ্ছে।
৪. Avro ডেটা স্ট্রিমে ইভেন্ট প্রসেসিং
Avro ফরম্যাটের ডেটা Kafka Streams-এর মাধ্যমে প্রসেস করার সময় Schema Registry স্কিমা থেকে ডেটার গঠন নিশ্চিত করে। এর ফলে ডেটা প্রেরণের সময় কাঙ্খিত স্কিমা ভ্যালিডেশন করা হয়।
Event Processing উদাহরণ:
stream.filter((key, value) -> value.get("age").equals(30))
.to("filtered-output-topic", Produced.with(Serdes.String(), new KafkaAvroSerializer()));
এখানে, filter() মেথড ব্যবহার করে Avro ডেটা ফিল্টার করা হচ্ছে, যাতে শুধুমাত্র বয়স ৩০ হওয়া রেকর্ডগুলিই পরবর্তী টপিকে প্রেরিত হয়।
সারাংশ
Kafka Streams এবং Avro এর ইনটিগ্রেশন দ্বারা ডেটা স্ট্রিম প্রসেসিং আরও কার্যকরী এবং স্কেলেবল হয়ে ওঠে। Avro ফরম্যাটের স্কিমা ব্যবহার করে আপনি ডেটার গঠন এবং স্কিমা ইভোলিউশন নিয়ন্ত্রণ করতে পারেন। Schema Registry ব্যবহার করলে Avro স্কিমার ইন্টিগ্রেশন সহজ হয় এবং Kafka Streams-এ ডেটা সিরিয়ালাইজেশন এবং ডেসিরিয়ালাইজেশন নিশ্চিত হয়। বিভিন্ন Kafka টপিকে ডেটা প্রেরণ এবং গ্রহণের সময় স্কিমা ভিত্তিক ডেটা প্রসেসিং পারফরম্যান্স উন্নত করে এবং ডেটার সঠিকতা বজায় রাখে।
Read more
