অ্যাপাচি ফ্লুম (Apache Flume) একটি শক্তিশালী ডেটা সংগ্রহ সিস্টেম যা বিভিন্ন উৎস (sources) থেকে ডেটা সংগ্রহ করে, তা প্রক্রিয়াকরণ করে এবং সিঙ্ক (sinks) তে প্রেরণ করে। Flume Interceptors ব্যবহার করে ডেটা ফিল্টারিং (filtering) এবং রাউটিং (routing) করা যায়, যা ডেটা ইনজেশন প্রক্রিয়াকে আরও কাস্টমাইজড এবং কার্যকরী করে তোলে।
ডেটা ফিল্টারিং এবং রাউটিং ইন্টারসেপ্টরগুলির মাধ্যমে খুব সহজে করা যায়। ইন্টারসেপ্টর (interceptor) ফ্লুমের এমন একটি উপাদান, যা ডেটা সংগ্রহ ও প্রসেসিংয়ের মধ্যে কাজ করে এবং ডেটা ফিল্টার বা রাউট করার জন্য কাস্টম লজিক প্রয়োগ করতে সাহায্য করে।
Flume Interceptors দিয়ে Data Filtering
ডেটা ফিল্টারিংয়ের মাধ্যমে, আপনি কিছু নির্দিষ্ট ডেটা বাদ দিতে পারেন বা শুধুমাত্র নির্দিষ্ট ধরনের ডেটা প্রসেস করতে পারেন। যেমন, যদি কোনো ইভেন্টের মধ্যে নির্দিষ্ট শব্দ বা প্যাটার্ন থাকে, তবে কেবল সেগুলোই প্রসেস করা হবে। অন্যথায় সেগুলো ফিল্টার হয়ে যাবে।
ফিল্টারিংয়ের জন্য কাস্টম ইন্টারসেপ্টর তৈরি করা
ফ্লুমে কাস্টম ইন্টারসেপ্টর তৈরি করে ডেটা ফিল্টার করার জন্য intercept() মেথডের মধ্যে আপনার লজিক প্রয়োগ করতে হবে। উদাহরণস্বরূপ, আপনি যদি এমন ডেটা ফিল্টার করতে চান যা ERROR বা WARN কনটেন্ট ধারণ করে, তবে নিচের কোড ব্যবহার করতে পারেন।
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.Event;
import java.util.List;
import java.util.ArrayList;
public class LogLevelInterceptor implements Interceptor {
@Override
public void initialize() {
// কাস্টম লজিকের জন্য ইনিশিয়ালাইজেশন
}
@Override
public List<Event> intercept(List<Event> events) {
List<Event> filteredEvents = new ArrayList<>();
for (Event event : events) {
String eventBody = new String(event.getBody());
// শুধুমাত্র ERROR বা WARN লগ গুলি রাখুন
if (eventBody.contains("ERROR") || eventBody.contains("WARN")) {
filteredEvents.add(event);
}
}
return filteredEvents;
}
@Override
public void close() {
// কোন ক্লিনআপ বা রিসোর্স রিলিজ প্রয়োজন হলে এখানে করবেন
}
}
এখানে, intercept() মেথডে কাস্টম লজিক প্রয়োগ করা হয়েছে যাতে কেবল ERROR বা WARN শব্দ সম্বলিত ইভেন্টগুলো প্রসেস হয়, অন্যগুলো বাদ পড়ে যাবে।
Flume Interceptors দিয়ে Data Routing
ডেটা রাউটিংয়ের মাধ্যমে, আপনি নির্দিষ্ট ডেটাকে আলাদা আলাদা সিঙ্কে পাঠাতে পারেন। অর্থাৎ, ডেটা উৎস (source) থেকে আসার পর, আপনি চয়েস করতে পারবেন কোন ডেটা কোন সিঙ্কে যাবে। এক্ষেত্রে, ফ্লুমের ইন্টারসেপ্টর ব্যবহৃত হয় ডেটা সিঙ্ক নির্বাচনের জন্য।
রাউটিংয়ের জন্য কাস্টম ইন্টারসেপ্টর তৈরি করা
ডেটা রাউট করার জন্য আপনি একটি কাস্টম ইন্টারসেপ্টর তৈরি করতে পারেন যা ডেটার মধ্যে থাকা কোনো বিশেষ বৈশিষ্ট্যের উপর ভিত্তি করে আলাদা আলাদা সিঙ্কে ডেটা পাঠায়। নিচের উদাহরণে, আমরা ডেটার মধ্যে level ফিল্ড অনুসারে আলাদা সিঙ্কে রাউটিং করছি:
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.Event;
import java.util.List;
import java.util.ArrayList;
public class LevelBasedRoutingInterceptor implements Interceptor {
@Override
public void initialize() {
// কাস্টম লজিকের জন্য ইনিশিয়ালাইজেশন
}
@Override
public List<Event> intercept(List<Event> events) {
List<Event> routedEvents = new ArrayList<>();
for (Event event : events) {
String eventBody = new String(event.getBody());
if (eventBody.contains("INFO")) {
// INFO level ডেটাকে infoSink এ পাঠানো হবে
event.getHeaders().put("sink", "infoSink");
} else if (eventBody.contains("ERROR")) {
// ERROR level ডেটাকে errorSink এ পাঠানো হবে
event.getHeaders().put("sink", "errorSink");
} else {
// অন্য কোনো ডেটা সাধারণ সিঙ্কে যাবে
event.getHeaders().put("sink", "defaultSink");
}
routedEvents.add(event);
}
return routedEvents;
}
@Override
public void close() {
// কোন ক্লিনআপ বা রিসোর্স রিলিজ প্রয়োজন হলে এখানে করবেন
}
}
এখানে, level ফিল্ডের উপর ভিত্তি করে ডেটাকে আলাদা সিঙ্কে রাউট করা হচ্ছে। যদি INFO থাকে, তাহলে সেটি infoSink সিঙ্কে যাবে এবং যদি ERROR থাকে, তাহলে তা errorSink সিঙ্কে যাবে।
কনফিগারেশন ফাইলে ইন্টারসেপ্টর রেজিস্টার করা
ফ্লুম কনফিগারেশন ফাইলে ইন্টারসেপ্টরগুলি রেজিস্টার করা হয় যাতে ফ্লুম সেগুলিকে ব্যবহার করতে পারে। উদাহরণস্বরূপ, যদি আপনি LogLevelInterceptor এবং LevelBasedRoutingInterceptor ব্যবহার করতে চান, তবে কনফিগারেশন ফাইলে নিচের মতো লিখবেন:
# সোর্স কনফিগারেশন
agent.sources = source1
agent.sources.source1.type = exec
agent.sources.source1.command = tail -F /var/log/syslog
# ফিল্টারিং এবং রাউটিং ইন্টারসেপ্টর রেজিস্টার করা
agent.sources.source1.interceptors = filterInterceptor routingInterceptor
agent.sources.source1.interceptors.filterInterceptor.type = com.example.LogLevelInterceptor
agent.sources.source1.interceptors.routingInterceptor.type = com.example.LevelBasedRoutingInterceptor
এখানে, দুটি কাস্টম ইন্টারসেপ্টর filterInterceptor এবং routingInterceptor রেজিস্টার করা হয়েছে। ফ্লুম এই ইন্টারসেপ্টরগুলিকে সোর্স থেকে আগত ডেটা প্রক্রিয়া করার জন্য ব্যবহার করবে।
Flume Interceptors দিয়ে Data Filtering এবং Routing এর সুবিধা
- কাস্টম লজিক প্রয়োগ: ডেটা ফিল্টারিং বা রাউটিং করার মাধ্যমে আপনার ডেটা ইনজেশন প্রক্রিয়ায় কাস্টম লজিক প্রয়োগ করতে পারবেন।
- বিভিন্ন সিঙ্কে ডেটা পাঠানো: ডেটার ধরন বা বৈশিষ্ট্যের ওপর ভিত্তি করে ডেটাকে আলাদা আলাদা সিঙ্কে পাঠানোর সুবিধা।
- স্কেলেবিলিটি এবং দক্ষতা: ডেটা ফিল্টারিং এবং রাউটিংয়ের মাধ্যমে শুধুমাত্র প্রয়োজনীয় ডেটা প্রসেস হবে, যা সিস্টেমের স্কেলেবিলিটি এবং কর্মক্ষমতা বাড়াবে।
Flume Interceptors দিয়ে Data Filtering এবং Routing আপনার ডেটা ইনজেশন প্রক্রিয়াকে আরও স্মার্ট এবং কাস্টমাইজড করতে সাহায্য করে, এবং এটি বিশেষ করে বড় ডেটা এনভায়রনমেন্টে ব্যবহৃত হলে কার্যকরী হয়।
Read more
