Apache Spark একটি শক্তিশালী ডিস্ট্রিবিউটেড ডেটা প্রসেসিং ফ্রেমওয়ার্ক যা ডেটা বিশ্লেষণ, ট্রান্সফরমেশন, এবং মেশিন লার্নিংসহ বিভিন্ন ডেটা প্রসেসিং কার্যক্রমে ব্যবহৃত হয়। Real-time data processing বা streaming হলো এমন একটি প্রক্রিয়া যেখানে ডেটা আসার সাথে সাথে সেটি প্রক্রিয়া করা হয়। স্পার্কে Structured Streaming এর মাধ্যমে আমরা DataFrame API ব্যবহার করে রিয়েল-টাইম ডেটা প্রসেসিং করতে পারি।
এই টিউটোরিয়ালে, আমরা DataFrame API ব্যবহার করে Real-time Data Processing নিয়ে আলোচনা করব এবং কিভাবে স্পার্কে রিয়েল-টাইম ডেটা প্রসেসিং করা যায় তা দেখব।
Structured Streaming in Apache Spark
Structured Streaming হলো স্পার্কের একটি API যা স্ট্রাকচারড ডেটা বা DataFrame/Dataset এর মাধ্যমে রিয়েল-টাইম ডেটা প্রসেসিং করার সুবিধা প্রদান করে। এটি ডেটা স্ট্রিমিংয়ের জন্য স্পার্কের এক ধরনের DataFrame API যা স্ট্রিমিং ডেটার সাথে একইভাবে কাজ করে যেমনটি আপনি ব্যাচ ডেটার সাথে কাজ করেন।
Structured Streaming একটি ফ্রেমওয়ার্ক হিসেবে কাজ করে যেখানে আপনি DataFrame বা Dataset API ব্যবহার করে ডেটা প্রসেস করেন এবং সেই ডেটা continuous ভাবে আপডেট হয়।
Structured Streaming Basic Example
স্পার্ক স্ট্রাকচারড স্ট্রিমিংয়ের মাধ্যমে রিয়েল-টাইম ডেটা প্রসেসিং করার জন্য, প্রথমে আপনি ইনপুট সোর্স (যেমন কনসোল, Kafka, হাডুপ ফাইল সিস্টেম) থেকে ডেটা পড়েন এবং তারপর ডেটা ট্রান্সফর্মেশন ও কুয়েরি চালান।
Step 1: SparkSession and Structured Streaming Setup
প্রথমে, SparkSession তৈরি করতে হবে এবং স্পার্ক স্ট্রাকচারড স্ট্রিমিং সক্রিয় করতে হবে।
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
val spark = SparkSession.builder()
.appName("Real-time Data Processing Example")
.getOrCreate()
Step 2: Reading Streaming Data
স্পার্ক স্ট্রাকচারড স্ট্রিমিং এ readStream() ফাংশন ব্যবহার করে আপনি কনসোল, Kafka, অথবা ফাইল সিস্টেম থেকে রিয়েল-টাইম ডেটা পড়তে পারেন।
// Reading streaming data from a directory (file stream)
val inputDF = spark.readStream
.format("csv")
.option("header", "true")
.schema("name STRING, age INT")
.load("path/to/your/input/data")
inputDF.printSchema()
এখানে:
- readStream() ফাংশনটি স্ট্রিমিং ডেটা পড়তে ব্যবহৃত হয়।
- format("csv") কনফিগারেশনটি ডেটার ফরম্যাট সেট করে।
- option("header", "true"): প্রথম লাইনটি হেডার হিসেবে গ্রহণ করে।
- schema(): ডেটার স্কিমা নির্ধারণ করে।
Step 3: Transformations on Streaming Data
স্ট্রিমিং ডেটার ওপর transformations যেমন map(), filter(), groupBy() ইত্যাদি ব্যবহার করা যেতে পারে।
val transformedDF = inputDF.filter(col("age") > 30)
এখানে:
- filter() ফাংশনটি age > 30 শর্তে ডেটা ফিল্টার করছে।
Step 4: Writing Stream to Sink
স্ট্রিমিং ডেটা প্রসেস করার পর সেই ডেটাকে sink এ লিখতে হবে। এখানে console একটি সাধারণ sink হিসেবে ব্যবহৃত হয়, যেখানে আপনি রিয়েল-টাইম আউটপুট দেখতে পাবেন।
val query = transformedDF.writeStream
.outputMode("append") // This defines how the output is written: 'append', 'complete', or 'update'
.format("console")
.start()
query.awaitTermination()
এখানে:
- writeStream(): স্ট্রিমিং ডেটা আউটপুট করার জন্য ব্যবহৃত হয়।
- outputMode("append"): আউটপুট মুছে না ফেলে নতুন ডেটা অ্যাড করবে।
- format("console"): আউটপুট কনসোলে প্রিন্ট হবে।
Step 5: Running and Stopping the Stream
রিয়েল-টাইম ডেটা প্রসেসিংয়ের জন্য স্পার্ক স্ট্রিমিং অ্যাপ্লিকেশনটি চালানো এবং থামানো খুবই গুরুত্বপূর্ণ। awaitTermination() ফাংশনটি স্ট্রিমিং প্রক্রিয়া চালু রাখে যতক্ষণ না আপনি নিজে এটি থামান।
query.awaitTermination() // Keep the stream running
এটি স্ট্রিমিং কুয়েরির কার্যক্রম চালু রাখে যতক্ষণ না আপনি সেটি থামান।
Advanced Operations with Structured Streaming
- Aggregations on Streaming Data: আপনি স্ট্রিমিং ডেটার উপর বিভিন্ন ধরনের অ্যাগ্রিগেশন করতে পারেন, যেমন গড়, মোট, গুণফল ইত্যাদি।
val aggregatedDF = inputDF.groupBy("name").agg(avg("age").alias("avg_age"))
এখানে, avg("age") ফাংশনটি স্ট্রিমিং ডেটার উপর age কলামের গড় হিসাব করছে।
- Windowing Operations: window() ফাংশন ব্যবহার করে আপনি একটি নির্দিষ্ট সময়সীমার মধ্যে ডেটা অ্যাগ্রিগেট করতে পারেন। এটি রিয়েল-টাইম ডেটা প্রসেসিংয়ে খুবই গুরুত্বপূর্ণ।
val windowedDF = inputDF
.withWatermark("timestamp", "10 minutes")
.groupBy(window(col("timestamp"), "5 minutes"))
.agg(sum("value").alias("total_value"))
এখানে:
- withWatermark(): ডেটা স্ট্রিমিংয়ের মধ্যে লেট ডেটা পরিচালনা করার জন্য ব্যবহৃত হয়।
- window(): একটি সময়ের জানুয়ারি উইন্ডো তৈরি করে।
- Streaming Joins: আপনি স্ট্রিমিং ডেটা সেটগুলির মধ্যে join অপারেশনও চালাতে পারেন। তবে, এখানে কিছু সীমাবদ্ধতা থাকতে পারে, যেমন ডেটা ফ্রেমের মধ্যে একটি সময়সীমা নির্ধারণ করা।
val stream1 = spark.readStream.format("kafka").option("subscribe", "topic1").load()
val stream2 = spark.readStream.format("kafka").option("subscribe", "topic2").load()
val joinedStream = stream1.join(stream2, stream1("key") === stream2("key"))
এখানে:
- join() ফাংশনটি দুইটি স্ট্রিমিং ডেটাসেটের মধ্যে যুক্ত হচ্ছে।
Fault Tolerance and Exactly Once Semantics
স্পার্ক স্ট্রাকচারড স্ট্রিমিং fault tolerance সমর্থন করে, এবং এটি Exactly Once Semantics (EOS) নিশ্চিত করতে পারে। এটি নিশ্চিত করে যে স্ট্রিমিং ডেটার কোনো রেকর্ড একাধিক বার প্রসেস হবে না, যদিও সিস্টেমের মধ্যে কোনো ত্রুটি ঘটে।
স্পার্ক স্ট্রাকচারড স্ট্রিমিংয়ে checkpointing এবং write-ahead logs (WAL) ব্যবহার করা হয়, যা স্ট্রিমিং প্রসেসিংয়ের মধ্যে ত্রুটির ক্ষেত্রেও ডেটাকে সঠিকভাবে সংরক্ষণ করতে সহায়তা করে।
val query = transformedDF.writeStream
.outputMode("append")
.format("parquet")
.option("checkpointLocation", "path_to_checkpoint_directory")
.start()
এখানে, checkpointLocation ডিরেক্টরি ব্যবহার করা হচ্ছে যাতে স্ট্রিমিং প্রসেসটি ত্রুটিপূর্ণভাবে পুনরায় শুরু করা গেলে সঠিক অবস্থান থেকে শুরু হতে পারে।
Conclusion
Structured Streaming স্পার্কের একটি শক্তিশালী ফিচার যা DataFrame API ব্যবহার করে রিয়েল-টাইম ডেটা প্রসেসিং সম্পাদন করতে সহায়তা করে। এটি ডেটা স্ট্রিমিংয়ের ক্ষেত্রে streaming joins, windowing, aggregations, fault tolerance, এবং exactly once semantics সমর্থন করে, যা রিয়েল-টাইম ডেটা প্রসেসিংয়ের জন্য অত্যন্ত কার্যকর। স্পার্ক স্ট্রাকচারড স্ট্রিমিংয়ের মাধ্যমে আপনি সহজে রিয়েল-টাইম ডেটা প্রসেসিং করতে পারেন এবং ডেটাকে কার্যকরভাবে ট্রান্সফর্ম, অ্যানালাইজ এবং স্টোর করতে পারবেন।
Read more
