Apache Storm হল একটি ওপেন-সোর্স ডিসট্রিবিউটেড রিয়েল-টাইম ডেটা প্রসেসিং সিস্টেম, যা স্ট্রিমিং ডেটা প্রক্রিয়া করার জন্য ব্যবহৃত হয়। Storm এর একটি গুরুত্বপূর্ণ বৈশিষ্ট্য হলো Trident API, যা Storm-এর উচ্চ-স্তরের স্ট্রিম প্রক্রিয়াকরণের জন্য ব্যবহৃত হয়। Trident ব্যবহার করে আপনি সহজে জটিল স্ট্রিমিং ডেটা প্রসেসিং টপোলজি তৈরি করতে পারেন।
Trident Topology তৈরি করার মাধ্যমে আপনি বিভিন্ন ধরনের কাজ যেমন ডেটা ফিল্টারিং, ট্রান্সফরমেশন, অ্যাগ্রিগেশন এবং স্টোরেজ পরিচালনা করতে পারেন। Trident-এ টপোলজি তৈরি করা Storm-এর সাধারণ টপোলজি তৈরি করার তুলনায় আরো সহজ এবং উন্নত।
Trident API কী?
Trident হলো Storm-এর একটি API যা স্ট্রিম প্রক্রিয়াকরণের জন্য উচ্চ স্তরের অপারেশন প্রদান করে। এটি Storm টপোলজির মধ্যে পার্টিশনিং, ফিল্টারিং, এবং অ্যাগ্রিগেশনসহ বিভিন্ন ধরনের কাজ সম্পাদন করতে সাহায্য করে। Trident স্ট্রিমিং ডেটা অ্যানালিটিক্সের জন্য বিশেষভাবে উপযুক্ত, কারণ এটি অ্যাপ্লিকেশন নির্মাণের জন্য একটি সহজ এবং declarative ইন্টারফেস প্রদান করে।
Trident API Storm-এর ওপর ভিত্তি করে কাজ করে, তবে এটি ব্যবহারকারীদের জন্য জটিল স্ট্রিম প্রক্রিয়াকরণ যেমন ট্রান্সফরমেশন, অ্যাগ্রিগেশন এবং স্টেটফুল অপারেশন সরবরাহ করে।
Trident Topology তৈরি করার স্টেপ-by-স্টেপ প্রক্রিয়া
Trident ব্যবহার করে একটি টপোলজি তৈরি করার জন্য কিছু নির্দিষ্ট ধাপ অনুসরণ করতে হবে। এই ধাপগুলো নিম্নরূপ:
১. Maven Dependency যোগ করা
প্রথমে, আপনার প্রজেক্টে Storm এবং Trident এর Maven ডিপেনডেন্সি যোগ করতে হবে। আপনার pom.xml ফাইলে নিম্নলিখিত ডিপেনডেন্সি যোগ করুন:
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-trident</artifactId>
<version>2.4.0</version>
</dependency>
এখানে, আপনি Storm এর উপযুক্ত ভার্সনটি নির্বাচন করতে পারেন।
২. Trident Topology তৈরি করা
Trident টপোলজি তৈরি করতে, আমরা প্রথমে একটি TridentTopology অবজেক্ট তৈরি করি। এরপর, আমরা স্পাউট এবং বোল্ট সংজ্ঞায়িত করি, এবং বিভিন্ন স্ট্রিম অপারেশন যোগ করি।
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.spout.SpoutSpec;
import org.apache.storm.trident.spout.Scheme;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.operation.builtin.Fields;
import org.apache.storm.tuple.Fields;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.spout.BatchSpout;
import org.apache.storm.topology.TopologyBuilder;
public class TridentExample {
public static void main(String[] args) {
TridentTopology topology = new TridentTopology();
// স্পাউট সংজ্ঞায়িত করা
SpoutSpec spoutSpec = new SpoutSpec(new BatchSpout());
// Trident স্ট্রিম তৈরি করা
topology.newStream("spout", spoutSpec)
.each(new Fields("input_field"), new Count(), new Fields("count"))
.each(new Fields("count"), new SomeOtherOperation(), new Fields("output_field"));
// টপোলজি চালানো
Config conf = new Config();
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("Trident-Topology", conf, topology.build());
}
}
এখানে:
SpoutSpec: Trident স্পাউট স্পেসিফিকেশন।each(): এটি একটি ট্রান্সফরমেশন অপারেশন, যা স্ট্রিমের উপর প্রক্রিয়া সম্পন্ন করে। এখানে আমরাCountঅপারেশন ব্যবহার করছি।new Fields(): এটি স্ট্রিমের মাধ্যমে প্রেরিত বিভিন্ন ফিল্ডকে নির্ধারণ করে।
৩. Trident Spout তৈরি করা
Trident টপোলজির জন্য স্পাউট তৈরি করতে, আপনি একটি কাস্টম স্পাউট তৈরি করতে পারেন যা ডেটা পাঠানোর কাজ করবে। উদাহরণস্বরূপ:
import org.apache.storm.trident.spout.BatchSpout;
import org.apache.storm.trident.operation.TridentTuple;
import org.apache.storm.trident.spout.BatchOutputCollector;
public class CustomBatchSpout extends BatchSpout {
@Override
public void open(Map conf, TopologyContext context) {
// স্পাউট খোলার সময় প্রয়োজনীয় কনফিগারেশন সেটআপ
}
@Override
public void nextBatch(Map batchMeta, BatchOutputCollector collector) {
// ডেটা প্রেরণ করতে হবে
collector.emit(new Values("some_data"));
}
@Override
public void ack(Object batchId) {
// ব্যাচ সফলভাবে প্রক্রিয়া হলে এখানে কোন কাজ করতে পারেন
}
@Override
public void close() {
// স্পাউট বন্ধ করার সময় কোনো কাজ করলে
}
}
এই কাস্টম স্পাউট nextBatch() মেথডে ডেটা পাঠাবে যা টপোলজির বাকি অংশে প্রক্রিয়া করার জন্য প্রেরিত হবে।
৪. Trident Bolt তৈরি করা
Trident টপোলজিতে কাজ করার জন্য আপনি কাস্টম বোল্ট তৈরি করতে পারেন। উদাহরণস্বরূপ, এখানে একটি কাস্টম ট্রান্সফরমেশন বোল্টের কোড দেখানো হলো:
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.tuple.TridentTuple;
public class CustomFunction extends BaseFunction {
@Override
public void execute(TridentTuple tuple) {
String input = tuple.getStringByField("input_field");
String output = "Transformed: " + input;
collector.emit(new Values(output));
}
}
এই কাস্টম বোল্ট ইনপুট ফিল্ড থেকে ডেটা নিয়ে সেটি প্রসেস করে আউটপুট প্রদান করবে।
৫. Trident টপোলজি চালানো
সবশেষে, আপনি ট্রাইডেন্ট টপোলজিটি চালানোর জন্য LocalCluster বা ক্লাস্টার সেটআপ ব্যবহার করতে পারেন।
Config conf = new Config();
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("Trident-Topology", conf, topology.build());
সারাংশ
Trident Storm-এর একটি উচ্চ-স্তরের API, যা স্ট্রিমিং ডেটা প্রসেসিংকে সহজ করে তোলে। Trident Topology তৈরি করতে, প্রথমে আপনি স্পাউট এবং বোল্ট তৈরি করবেন, এরপর সেগুলোর মাধ্যমে ডেটা স্ট্রিমের উপর বিভিন্ন প্রক্রিয়া সম্পন্ন করবেন। Trident-এর মাধ্যমে আপনি স্ট্রিমের ডেটা ফিল্টারিং, অ্যাগ্রিগেশন এবং ট্রান্সফরমেশন সহ বিভিন্ন ধরনের অপারেশন খুব সহজে বাস্তবায়ন করতে পারবেন।
Read more
