Apache Spark SQL-এ Batch এবং Streaming Data একত্রে ব্যবহারের মাধ্যমে ডেটা প্রসেসিংয়ের জন্য অত্যন্ত শক্তিশালী পদ্ধতি তৈরি করা যায়। Batch Data এবং Streaming Data দুটি ভিন্ন ধরনের ডেটা প্রসেসিং পদ্ধতি, কিন্তু Spark SQL-এর মাধ্যমে এগুলিকে একসাথে সংযুক্ত করা সম্ভব, যা ডেটা অ্যানালাইসিস এবং রিয়েল-টাইম প্রসেসিংয়ের একটি ইন্টিগ্রেটেড অ্যাপ্রোচ তৈরি করে।
এখানে, আমরা আলোচনা করব Batch Data এবং Streaming Data এর মধ্যে Integration কিভাবে করা যায় এবং এর পারফরম্যান্স কিভাবে পরিচালনা করা হয়।
Batch Data এবং Streaming Data
১. Batch Data:
Batch Data হচ্ছে সেগুলি যেগুলি একবারে একটি নির্দিষ্ট সময়ের মধ্যে প্রসেস করা হয়। এটি সাধারণত ডেটা লোড এবং প্রসেস করার একটি সিঙ্ক্রোনাস প্রক্রিয়া। Batch Data ব্যবহৃত হয় যখন ডেটা পেতে বা প্রসেস করার জন্য বড় বড় সময়ের ফাঁক থাকে, যেমন দিনের শেষে রিপোর্ট তৈরি বা মাসিক ডেটা প্রসেসিং।
২. Streaming Data:
Streaming Data হলো সেগুলি যা ক্রমাগত প্রবাহিত হয় এবং প্রায় রিয়েল-টাইমে প্রসেস করা হয়। এটি সাধারণত সিস্টেম বা অ্যাপ্লিকেশন দ্বারা রিয়েল-টাইম ডেটা জেনারেশন এবং অ্যানালাইসিস করতে ব্যবহৃত হয়, যেমন সোসাল মিডিয়া ফিড, সেন্সর ডেটা, বা ওয়েব লগ ফাইল।
Batch এবং Streaming Data এর মধ্যে Integration
Apache Spark SQL Structured Streaming এবং Batch Data-কে একসাথে ইন্টিগ্রেট করার জন্য একটি শক্তিশালী ফিচার সরবরাহ করে। Spark Streaming, যা DStreams (Discretized Streams) ভিত্তিক, তাকে একটি আধুনিক Structured Streaming API দ্বারা প্রতিস্থাপন করা হয়েছে, যা সহজে SQL কোয়ারি এবং DataFrame API-এর মাধ্যমে রিয়েল-টাইম ডেটা প্রসেসিং এবং Batch Data-কে একত্রে পরিচালনা করতে সক্ষম।
১. Structured Streaming: Spark SQL-এ স্ট্রাকচারড স্ট্রিমিং (Structured Streaming) একটি শক্তিশালী পদ্ধতি, যা রিয়েল-টাইম ডেটা প্রসেসিংয়ের জন্য অত্যন্ত কার্যকরী। Structured Streaming Batch এবং Streaming ডেটা একত্রিত করার জন্য ব্যবহৃত হয়।
উদাহরণ: Structured Streaming এর মাধ্যমে Batch এবং Streaming ডেটার সংযোগ
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
# SparkSession তৈরি
spark = SparkSession.builder \
.appName("Batch and Streaming Integration") \
.getOrCreate()
# Batch Data (Parquet ফাইল) লোড করা
batch_df = spark.read.parquet("path/to/batch_data")
# Streaming Data (কিন্তু কনসোল থেকে ইনপুট পাচ্ছি, এখানে ফাইল অথবা কনসোল সোর্স ব্যবহার হতে পারে)
streaming_df = spark.readStream \
.format("json") \
.load("path/to/streaming_data")
# Batch Data এবং Streaming Data একত্রিত করা
result_df = batch_df.join(streaming_df, batch_df.id == streaming_df.id)
# Structured Streaming এর মাধ্যমে রিয়েল-টাইম ফলাফল দেখানো
query = result_df.writeStream \
.outputMode("append") \
.format("console") \
.start()
query.awaitTermination()
এখানে:
- Batch Data হলো সেগুলি যা
parquetফাইল থেকে লোড করা হয়েছে। - Streaming Data হলো JSON ফাইল থেকে আসা রিয়েল-টাইম ডেটা, যা
readStreamএর মাধ্যমে লোড হচ্ছে। - Structured Streaming এর মাধ্যমে Batch Data এবং Streaming Data একত্রিত করা হয়েছে এবং ফলাফল কনসোলে দেখানো হচ্ছে।
২. Batch এবং Streaming Data-র মধ্যে Join
Structured Streaming API Batch এবং Streaming Data-র মধ্যে join অপারেশন করতে সক্ষম, যা আপনাকে স্ট্রিমিং ডেটার সঙ্গে ঐতিহাসিক বা ব্যাচ ডেটা একত্রিত করতে দেয়।
উদাহরণ: Batch এবং Streaming Data-র মধ্যে Join অপারেশন
# Batch Data লোড করা
batch_df = spark.read.parquet("path/to/batch_data")
# Streaming Data লোড করা
streaming_df = spark.readStream \
.format("json") \
.load("path/to/streaming_data")
# Batch Data এবং Streaming Data এর মধ্যে Join অপারেশন
joined_df = batch_df.join(streaming_df, "id")
# Output Stream তৈরী করা
query = joined_df.writeStream \
.outputMode("append") \
.format("console") \
.start()
query.awaitTermination()
এখানে:
- Batch Data এবং Streaming Data দুটি টেবিলকে
join()ফাংশনের মাধ্যমে একত্রিত করা হচ্ছে। writeStreamএবংoutputMode("append")ব্যবহার করে রিয়েল-টাইম ফলাফল কনসোলে আউটপুট করা হচ্ছে।
৩. Micro-batching এবং Real-time Processing
Spark Streaming-এর মডেল হল Micro-batching। এখানে, স্ট্রিমিং ডেটা ছোট ছোট ব্যাচে প্রসেস হয়। এটি স্ট্রিমিং ডেটাকে একটি ব্যাচ প্রসেসিং মডেলে রূপান্তরিত করে, যা স্ট্রিমিং ডেটার উপর batch-style কোয়ারি চালাতে সাহায্য করে। Structured Streaming এ মাইক্রো-ব্যাচিং মডেলটি আরো উন্নত এবং স্বাভাবিক হয়ে উঠেছে।
উদাহরণ: Micro-batching in Structured Streaming
# Streaming data লোড করা
streaming_df = spark.readStream \
.format("json") \
.option("maxFilesPerTrigger", 1) \ # Micro-batching (একসাথে ১টি ফাইল প্রক্রিয়া করবে)
.load("path/to/streaming_data")
# Aggregation বা Transformation করা
aggregated_df = streaming_df.groupBy("category").agg(count("id"))
# Write stream to console
query = aggregated_df.writeStream \
.outputMode("complete") \
.format("console") \
.start()
query.awaitTermination()
এখানে:
- maxFilesPerTrigger প্যারামিটার দিয়ে micro-batching সিস্টেম কনফিগার করা হয়েছে, যা প্রতি ব্যাচে একটি করে ফাইল প্রসেস করবে।
- groupBy() এবং agg() ব্যবহার করে স্ট্রিমিং ডেটার উপর aggregation অপারেশন চালানো হচ্ছে।
৪. Watermarking for Handling Late Data
একটি সাধারণ সমস্যা যা স্ট্রিমিং ডেটাতে হয় তা হল "late data" (যেসব ডেটা স্বাভাবিকভাবে প্রাপ্তির সময়ের পরে আসে)। Spark Structured Streaming এ watermarking ব্যবহৃত হয়, যা স্ট্রিমিং ডেটার জন্য দেরি হওয়া ডেটা নির্ধারণ এবং পরিচালনা করতে সাহায্য করে।
উদাহরণ: Watermarking
# Watermark ব্যবহার করে Late Data হ্যান্ডেল করা
streaming_df_with_watermark = streaming_df \
.withWatermark("timestamp", "10 minutes") \
.groupBy("id").agg(count("*").alias("count"))
# Write Stream to console
query = streaming_df_with_watermark.writeStream \
.outputMode("update") \
.format("console") \
.start()
query.awaitTermination()
এখানে:
withWatermark()পদ্ধতিটি স্ট্রিমিং ডেটার জন্য একটি টাইমস্ট্যাম্পের ভিত্তিতে late data ম্যানেজ করে, যার মাধ্যমে ১০ মিনিট পরে আসা ডেটা প্রসেস করা সম্ভব হয়।
সারাংশ
Spark SQL-এর Structured Streaming ব্যবহার করে Batch Data এবং Streaming Data একত্রে ব্যবহৃত হতে পারে। Spark SQL এই দুটি ডেটা টাইপের মধ্যে Join অপারেশন, Aggregation, এবং Watermarking এর মাধ্যমে একটি ইন্টিগ্রেটেড ডেটা প্রসেসিং সিস্টেম তৈরি করতে সক্ষম। Batch Data এবং Streaming Data-র মধ্যে Integration করার ফলে রিয়েল-টাইম ডেটা অ্যানালাইসিস আরও শক্তিশালী এবং ফ্লেক্সিবল হয়ে ওঠে, যা বিভিন্ন ধরনের ডেটা সোর্স থেকে দ্রুত এবং কার্যকরী ফলাফল পেতে সাহায্য করে।
Read more
