একটি Serverless Big Data Ingestion Pipeline

এই লেসন শেষে তুমি যা পারবে

• Classic serverless big data ingestion pipeline-টা না দেখে এঁকে ফেলতে পারবে: IoT device → IoT Core → Kinesis Data Streams → Data Firehose (Lambda transformation সহ) → S3 ingestion bucket → SQS → Lambda → Athena → S3 reporting bucket → QuickSight আর Redshift।
• প্রতিটা ধাপ কী কাজ করে, কেন সেখানে আছে, আর কোন আগের লেসনে সেটা বিস্তারিত পড়েছ, বলতে পারবে।
• কোন অংশ real time আর কোনটা near real time, ধরতে পারবে, আর device থেকে report পর্যন্ত কত দেরি হয় তার আন্দাজ করতে পারবে।
• প্রতিটা ধাপের খরচ কিসে বাড়ে বলতে পারবে, আর সংখ্যা দিয়ে দেখাতে পারবে কেন file format আর partition-ই Athena-র bill ঠিক করে দেয়।
• চেনা গোলমালগুলো ধরতে পারবে: ভাঙা record, schema বদলে যাওয়া, দেরিতে আসা data, ছোট ছোট file, partition ছাড়া scan, throttle হওয়া Lambda আর Athena, poison message, duplicate event আর নিজেকেই trigger করা loop।
• Pattern-টা দরকারমতো বদলাতে পারবে: কোনো ধাপ বাদ দেওয়া বা বদলানো, hot path যোগ করা, আর কয়েকটা pipeline একই রকম ঠিক মনে হলে পরীক্ষায় সঠিকটা বেছে নেওয়া।

#একটা service না, পুরো একটা pipeline কেন

#ছয়টা শর্ত

১৩, ১৫ আর ১৭ নম্বর chapter-এ টুকরোগুলো একটা একটা করে শিখেছ। এই লেসনে সেগুলো জোড়া লাগিয়ে সেই pattern-টা বানাব, যেটা দিয়ে পরীক্ষার Data & Analytics অংশ শেষ হয়। প্রশ্নে চাহিদাটা সাধারণত এভাবে আসে:

1. Ingestion pipeline হবে পুরোপুরি serverless: size করা, patch করা বা চালু রাখার মতো কিছুই থাকবে না।
2. Data real time-এ জোগাড় করতে হবে।
3. পথেই data রূপান্তর (transform) করতে হবে।
4. রূপান্তর করা data-তে SQL দিয়ে query করা যাবে।
5. ওই query থেকে যে report হবে, সেটা S3-এ জমা থাকবে।
6. Report-গুলো একটা data warehouse-এ load হবে আর dashboard-এ দেখানো হবে।

প্রতিটা লাইন একটা করে service-এর দিকে আঙুল তোলে। Real time-এ জোগাড় মানে একটা stream; code ছাড়া storage-এ পৌঁছানো মানে Firehose; file-এর ওপর SQL মানে Athena; dashboard মানে QuickSight; warehouse মানে Redshift। আসল দক্ষতা হলো এগুলো এমনভাবে জোড়া যাতে প্রতিটা ধাপ একটাই কাজ করে, পরিষ্কারভাবে পরের জনের হাতে তুলে দেয়, আর বিগড়ালে সেটা চোখে পড়ে।

#একটা service দিয়ে সব হয় না কেন

একটা service দিয়েই সব করে ফেলা যায় না? ভালোভাবে যায় না:

• Kinesis Data Streams real time-এ data ধরে, replay-এর জন্য record রেখেও দেয়, কিন্তু file লেখে না, SQL-এর উত্তরও দেয় না (লেসন ১৩.৫)।
• Firehose code ছাড়াই মাপমতো, compress করা, partition করা file লেখে, কিন্তু সে buffer করে, তাই near real time, আর replay করা যায় না (লেসন ১৩.৬)।
• S3 যা খুশি সস্তায় আর টেকসইভাবে রাখে, কিন্তু নিজে কোনো হিসাব করে না।
• Athena কোনো server ছাড়াই S3-এর ওপর SQL চালায়, কিন্তু query কাউকে চালু করে দিতে হয়, আর টাকা লাগে যত data scan হয় তত (লেসন ১৭.১)।
• Redshift হলো business data জুড়ে দ্রুত join করার warehouse, আর chart আঁকে QuickSight (লেসন ১৭.২, লেসন ১৭.৫)।

প্রত্যেকে একটা ধাপে দারুণ, বাকিগুলোতে দুর্বল। Pipeline হলো সেই শিকল, যেটা প্রতিটা service-কে শুধু তার সেরা কাজটাই করতে দেয়।

#এক নজরে পুরো pipeline

তিনটা পর্বে ভাগ করে পড়ো: ingest (ধাপ ১–৪, data এসে file হয়ে নামে), process (ধাপ ৫–৮, file থেকে report হয়) আর serve (ধাপ ৯, মানুষ report ব্যবহার করে)।

#ধাপে ধাপে

নিচের প্রতিটা অংশে বলা আছে এই pipeline-এ ধাপটা কী করে আর কী configure করতে হয়। গভীরে যাওয়া আছে link করা লেসনগুলোতে; এখানে নজর এক ধাপ থেকে আরেক ধাপে হাতবদলের ওপর।

#জোগাড়: device থেকে IoT Core

AWS IoT Core হলো device-দের জন্য managed সদর দরজা। আগের কোনো লেসনে এটা পড়ানো হয়নি, তাই পরীক্ষায় যা লাগে সেটুকু এখানে:

• Device-রা connect করে MQTT দিয়ে (MQTT over WebSocket আর HTTPS-ও চলে), আর প্রত্যেকে নিজের একটা X.509 certificate দিয়ে পরিচয় দেয়। স্ক্রুড্রাইভার দিয়ে যে কেউ খুলে ফেলতে পারে এমন meter-এর ভেতরে কোনো AWS access key রাখতে হয় না।
• Message publish হয় একটা topic-এ, যেমন factory/gazipur-1/floor-3/meter-2217। Payload সর্বোচ্চ ১২৮ KB।
• Rules engine topic-এর ওপর একটা SQL statement চালায় (SELECT * FROM 'factory/+/+/+'), আর মিলে যাওয়া message পাঠায় action-এ: Kinesis Data Streams, Firehose, S3, Lambda, SQS, SNS, DynamoDB, Timestream ইত্যাদি, একটা rule-এ সর্বোচ্চ ১০টা action। প্রতিটা rule-এ একটা error action-ও রাখা যায়, যেমন একটা backup S3 bucket, যাতে হাতবদল ব্যর্থ হলে data চুপচাপ হারিয়ে না যায়।
• এটা serverless: টাকা দাও প্রতি লাখ-দশলাখ message আর connection-এর মিনিট ধরে।

Slide-এর ভাষায়, IoT Core দিয়ে তুমি IoT device থেকে data সংগ্রহ (harvest) করো। পরীক্ষায় “হাজার হাজার sensor”, “MQTT” আর “device certificate” দেখলেই এদিকে তাকাও; IoT-এর machine learning বা device management-এর দিক পরীক্ষার বাইরে।

#ধরে রাখা: Kinesis Data Streams

IoT rule প্রতিটা reading একটা Kinesis data stream-এ লেখে, partition key হলো meter ID, তাই প্রতিটা meter-এর reading একটা shard-এর ভেতরে ক্রম ধরে থাকে। কিন্তু এখানে stream কেন, সরাসরি Firehose-এ না কেন?

• Real time। লেখার মোটামুটি ২০০ ms পরেই একজন consumer record-টা পড়তে পারে (enhanced fan-out-এ প্রায় ৭০ ms)। “real time-এ জোগাড়” শর্তটা পূরণ হয় এখানেই।
• কয়েকজন পাঠক। Firehose একজন consumer; একটা alert function বা একটা Flink application (লেসন ১৭.৭) একই সময়ে একই record পড়তে পারে, প্রত্যেকে নিজের গতিতে।
• Replay আর নিরাপত্তার buffer। Record থেকে যায় retention-এর পুরো সময় (default ২৪ ঘণ্টা, সর্বোচ্চ ৩৬৫ দিন)। নিচের দিকে সব থেমে গেলেও data এখানে অপেক্ষা করে।

Garments group-এর হিসাব: ৪,০০০ meter, প্রতিটা ১০ সেকেন্ডে একবার, মানে সেকেন্ডে ৪০০ record, প্রতিটা প্রায় ০.৫ KB, অর্থাৎ ২০০ KB/s। একটা shard নেয় সেকেন্ডে ১ MB বা ১,০০০ record, আর দেয় ২ MB/s, যেটা standard consumer-রা ভাগ করে নেয়। কাগজে-কলমে একটা shard-ই যথেষ্ট, কিন্তু কোনো কারখানার internet বন্ধ হয়ে ফিরলে শত শত meter কয়েক ঘণ্টার জমানো reading একসাথে পাঠায়। তাই team on-demand mode বেছে নেয়, যেটা নিজেই shard বাড়ায়-কমায়, আর pipeline-কে capacity-র হিসাব থেকে মুক্ত রাখে।

#পৌঁছানো আর রূপান্তর: Lambda সহ Data Firehose

Amazon Data Firehose (আগের নাম Kinesis Data Firehose) stream পড়ে আর ingestion bucket-এ তুলে দেয়। নিজের হাতে লেখা একটা consumer-কে যা যা করতে হতো, সব সে করে: batch, compress, রূপান্তর, partition, retry আর scale।

• Buffering। S3-এর জন্য buffer-এর একটা size (১–১২৮ MB, default ৫ MB) আর একটা interval (০–৯০০ সেকেন্ড, default ৩০০) থাকে; যেটা আগে পূরণ হয় সেটাই file লেখায়। Team দেয় ১২৮ MB আর ৩০০ সেকেন্ড। ২০০ KB/s-এ interval আগে আসে, তাই মোটামুটি ৫ মিনিট পরপর একটা file, দিনে প্রায় ২৯০টা।
• Lambda transformation। Firehose batch (default প্রায় ১ MB বা ৬০ সেকেন্ড) একটা Lambda function-কে synchronously পাঠায়, আর সর্বোচ্চ ৫ মিনিট অপেক্ষা করে। প্রতিটা record-এর জন্য function ফেরত দেয় recordId, একটা result (Ok, Dropped বা ProcessingFailed) আর নতুন data। এখানে সে watt-hour-কে kWh বানায়, meter ID দেখে floor-এর নাম যোগ করে, আর ভাঙা JSON বাতিল করে। Slide-এর “Lambda can help Firehose with data transformations” কথাটা এই feature-টাই।
• Format রূপান্তর। Firehose নিজেই JSON-কে Parquet (বা ORC) বানায়, Glue Data Catalog-এর একটা table schema দেখে (লেসন ১৭.৬)। পরে Athena সস্তা হয় এই columnar file-এর জন্যই।
• Dynamic partitioning। Firehose প্রতিটা record থেকে একটা key পড়ে (এখানে reading-এর দিন), আর লেখে data/day=2026-10-29/-এ। এমন key দিয়ে ভাগ করো যার মান কম; প্রতিটা মান নিজের আলাদা buffer পায়।
• Error হারায় না, জমা থাকে। যে record Lambda-তে ফেল করে সেটা processing-failed হিসেবে যায় একটা error prefix-এ; schema-র সাথে না মিললে যায় format-conversion-failed হিসেবে।

#নামানো: ingestion bucket

Ingestion bucket (data lake-এর raw zone) প্রতিটা reading রেখে দেয়, business-এর যতদিন দরকার। পরে যেসব সিদ্ধান্ত কাজে লাগে:

• Hive ধাঁচের partition দিয়ে prefix সাজানো: data/day=2026-10-29/। Query-তে day দিয়ে filter থাকলে Athena পুরো folder বাদ দিয়ে যেতে পারে (লেসন ১৭.১)।
• কম কিন্তু বড় file। প্রতিটা file কয়েক দশ থেকে কয়েকশো MB হলে সবচেয়ে ভালো; হাজার হাজার ছোট্ট file প্রতিটা query ধীর করে, S3 request-এর খরচ বাড়ায় (লেসন ১০.৬)।
• SSE-S3 বা SSE-KMS দিয়ে encryption; KMS হলে Firehose-এর role-কে key ব্যবহারের অনুমতি দিতে হয় (লেসন ২১.২)।
• Lifecycle rule পুরোনো কাঁচা data সস্তা storage class-এ সরিয়ে দেয় (লেসন ১০.৫); report ছোট, তাই Standard-এই থাকে।
• দুই রকম output-এর জন্য দুটো prefix: ভালো file-এর জন্য data/, আর Firehose-এর ফেল করা record-এর জন্য errors/। পরের ধাপের event rule শোনে শুধু data/।

#Trigger: S3 event, SQS, Lambda

কাউকে খেয়াল করতে হবে যে নতুন file এসেছে, আর query চালু করতে হবে। Classic design-এ থাকে:

1. s3:ObjectCreated:*-এর জন্য একটা S3 event notification, prefix filter data/, পাঠানো হয় একটা SQS queue-তে (লেসন ১০.৭)।
2. Queue-এর ওপর event source mapping সহ একটা Lambda function: Lambda-র poller-রা batch ধরে message পড়ে আর function চালায় (লেসন ১৫.১)।
3. File-টা যে দিনের, সেই দিনের জন্য function Athena-র StartQueryExecution ডাকে।

Slide বলে Lambda SQS-এ “subscribe” করতে পারে। আসলে queue-তে কেউ subscribe করে না: SQS pull-ভিত্তিক, আর event source mapping function-এর হয়ে queue poll করে। ফল একই, কিন্তু যে প্রশ্নে SNS (push, subscription) আর SQS (pull)-কে পাশাপাশি রাখা হয়, সেখানে শব্দটা গুরুত্বপূর্ণ।

Slide SQS-কে ঐচ্ছিক বলেছে, আর সত্যিই তাই: S3 সরাসরি Lambda ডাকতে পারে। তবু queue-টা জায়গা পায় কারণ:

• এটা ঢল সামলায়: ৩,০০০ file-এর একটা backfill একসাথে ৩,০০০ query না হয়ে ৩,০০০টা অপেক্ষমাণ message হয়;
• Event source mapping-এর maximum concurrency (২–১,০০০) ঠিক করে কতগুলো copy একসাথে চলবে, ফলে Athena তার active query-র quota-র নিচে থাকে (লেসন ১৫.২);
• ফেল করা message visibility timeout শেষে ফিরে আসে, আর maxReceiveCount পেরোলে চলে যায় একটা dead-letter queue-তে, যেটা তুমি খুলে দেখতে আর redrive করতে পারো (লেসন ১৩.২);
• Partial batch response (ReportBatchItemFailures) দিলে একটা খারাপ message পুরো batch-কে আবার চালাতে বাধ্য করে না।

Queue-এর visibility timeout রাখো function timeout-এর অন্তত ছয় গুণ, লেসন ১৩.২-এ যেমন বলা আছে।

#Query: Athena

Amazon Athena সরাসরি S3-এর file-এর ওপর SQL চালায়, table-এর সংজ্ঞা পড়ে Glue Data Catalog থেকে। এই pipeline-এ:

• Function এমন একটা query চালায়: UNLOAD (SELECT floor, date_trunc('hour', ts) AS hour, sum(kwh) … WHERE day = '2026-10-29' GROUP BY 1, 2) TO 's3://meter-reports-example/hourly/day=2026-10-29/' WITH (format = 'PARQUET')। UNLOAD ফলাফলটা Parquet হিসেবে reporting bucket-এ লেখে; CTAS (CREATE TABLE AS SELECT) একই কাজ করে, সাথে একটা table-ও register করে।
• প্রতিটা query নিজের ফলাফল workgroup-এর query result location-এও লেখে, যেটা S3-এ। Slide যে বলে Athena-র ফলাফল S3-এ জমা থাকে, তার মানে এটাই।
• Athena asynchronous: StartQueryExecution সাথে সাথে একটা query ID ফেরত দেয়, আর GetQueryExecution জানায় QUEUED, RUNNING, SUCCEEDED নাকি FAILED। Lambda function-এর উচিত query চালু করে ফিরে যাওয়া, মিনিটের পর মিনিট বসে poll করা না। লম্বা বা কয়েক ধাপের report হলে Step Functions .sync integration দিয়ে Athena চালায়, আর কোনো compute খরচ ছাড়াই অপেক্ষা করে (লেসন ১৫.৭)।
• দাম: বেশিরভাগ Region-এ scan হওয়া প্রতি TB-তে ৫ মার্কিন ডলার, MB পর্যন্ত ওপরে গোল করে, আর প্রতি query-তে কমপক্ষে ১০ MB ধরা হয় (বিস্তারিত লেসন ১৭.১-এ)। হাতল হলো Parquet, compression আর partition। নিরাপত্তার জাল হিসেবে একটা workgroup প্রতি query-তে scan-এর সীমা বেঁধে দিতে পারে।
• Quota: account প্রতি active query-র সংখ্যা সীমিত (চলমান আর লাইনে থাকা মিলিয়ে, Region ভেদে আলাদা); তার বেশি হলে StartQueryExecution TooManyRequestsException দিয়ে ফেল করে। Default-এ একটা query ৩০ মিনিটে timeout হয় (৪ ঘণ্টা পর্যন্ত বাড়ানো যায়)।

#Report: reporting bucket, QuickSight আর Redshift

Reporting bucket-এ থাকে বিশ্লেষণ করা data: ছোট, পরিষ্কার, মানুষের জন্য তৈরি। সেখান থেকে:

• QuickSight dashboard বানায়। সে report পড়ে Athena দিয়ে (direct query, তাজা, কিন্তু প্রতিবার দেখলেই একটা query চলে), অথবা নির্দিষ্ট সময় পরপর refresh করে SPICE-এ import করে, যেটা তার in-memory engine (প্রতিবার দেখা দ্রুত আর সস্তা, data ততটাই তাজা যতটা শেষ refresh)। QuickSight Redshift-ও পড়তে পারে। অক্টোবর ২০২৫ থেকে QuickSight হলো Amazon Quick Suite-এর অংশ (২০২৬-এ নামটা ছোট হয়ে Amazon Quick), যেখানে BI অংশটার নাম Quick Sight; API আর পরীক্ষার প্রশ্নে এখনো QuickSight-ই লেখা থাকে (লেসন ১৭.৫)।
• Redshift S3 থেকে COPY দিয়ে report load করে, order, বেতন আর খরচের table-এর সাথে join করতে; অথবা Redshift Spectrum দিয়ে জায়গাতেই query করে (লেসন ১৭.২)।

Slide-এ warehouse-এর গায়ে লেখা Redshift Serverless। শর্ত ১ রক্ষা হয় এতেই: provisioned Redshift cluster-এর node তুমি বেছে নাও, ঘণ্টা ধরে টাকা দাও, তাই শেষে সেটা থাকলে pipeline পুরোপুরি serverless না। প্রশ্নে “fully serverless” জোর দিয়ে বলা থাকলে আর দুটোই option-এ থাকলে Redshift Serverless বেছে নাও।

#Real time নাকি near real time: দেরির হিসাব

প্রশ্নে “real time” মানে সাধারণত millisecond থেকে মোটামুটি এক সেকেন্ড; “near real time” মানে কয়েক সেকেন্ড থেকে কয়েক মিনিট। Garments pipeline-এর ধাপগুলো যোগ করে দেখো:

তাই meter থেকে ঘণ্টার dashboard পর্যন্ত লাগে মোটামুটি ৫–১০ মিনিট: near real time। Real time শুধু hot path, মানে যে consumer stream পড়ে। প্রশ্নে এক সেকেন্ডের কম সময়ে সাড়া চাইলে উত্তর হলো Kinesis Data Streams-এর ওপর একজন consumer (Lambda, Flink বা KCL), কখনোই “Firehose-এর buffer কমাও” বা “Athena দিয়ে আরও দ্রুত query করো” না।

#প্রতিটা ধাপের খরচ

Serverless মানে যতটুকু ব্যবহার ততটুকু টাকা, তাই প্রতিটা ধাপের একটা খরচের চালক আছে, যেটা তুমি কমাতে পারো:

#Scan-এর bill: একটা হাতে-কলমে হিসাব

Meter-গুলো তৈরি করে সেকেন্ডে ৪০০ reading × ০.৫ KB = ২০০ KB/s, মানে দিনে প্রায় ১৭ GB JSON, আর মাসে ৫২০ GB।

• কাঁচা JSON, partition নেই। “গতকাল প্রতি floor-এ কত kWh” জানতে পুরো মাস পড়তে হয়: ০.৫২ TB × ৫ ডলার ≈ প্রতি query-তে ২.৬০ ডলার। ১৫ মিনিট পরপর refresh হওয়া একটা dashboard এটা দিনে ৯৬ বার চালায়: একটা chart-এর জন্য দিনে প্রায় ২৫০ ডলার।
• Parquet, দিন ধরে partition। Query পড়ে একটা দিন (রূপান্তর আর compression-এর পর প্রায় ১.৭ GB), আর বারোটা column-এর মধ্যে শুধু তিনটা: প্রায় ০.৪ GB, এক সেন্টের পাঁচ ভাগের এক ভাগ। দিনে ৯৬ বার চালালেও ২০ সেন্টের কম।
• কাঁচা data-র বদলে report। QuickSight পড়ে ঘণ্টার report (কয়েক MB), reading না। দর্শকদের দেখাতে খরচ প্রায় শূন্য।

SQL একটুও বদলায়নি। Data কীভাবে সাজানো সেটাই খরচ হাজার গুণেরও বেশি কমিয়েছে, আর সেই সাজানো ঠিক হয় উজানে, Firehose-এ।

#কোথায় গোলমাল হয়

#এই pattern দিয়ে design করা

#যেসব রূপভেদ দেখবে

Classic শিকলটা শুরু মাত্র। চাহিদা বদলালে বাস্তব system আর পরীক্ষার উত্তর ধাপ বদলে নেয়:

• Real time-এর দরকার নেই: IoT Core → সরাসরি Firehose → S3। এক ধাপ কম, সস্তা, কিন্তু hot path নেই, replay-ও নেই।
• আগে থেকেই Kafka producer আছে: Kinesis Data Streams-এর বদলে Amazon MSK; Firehose MSK থেকেও পড়তে পারে (১৭.৭)। MSK provisioned serverless না; MSK Serverless হলো serverless।
• Stream-এর ওপরেই প্রতি মিনিটের হিসাব: Managed Service for Apache Flink stream পড়ে, ফলাফল লেখে S3-এ বা আরেকটা stream-এ।
• Data-তে search: একটা Firehose stream বা আলাদা একটা শাখা OpenSearch-এ (১৭.৩)।
• S3 ধাপ ছাড়া warehouse: Firehose সরাসরি Redshift-এ পৌঁছাতে পারে (সে S3-এ file রেখে তোমার হয়ে COPY চালায়)। Firehose Apache Iceberg table-এও লিখতে পারে, S3 Tables সহ, যেটা transactional data lake-এর নতুন একটা পথ; পরীক্ষার উত্তর অবশ্য এখনো S3 bucket।
• ভারী রূপান্তর: একটা S3 event বা schedule Athena query-র বদলে একটা Glue job চালু করে; Glue crawler catalog হালনাগাদ রাখে।
• প্রতি file-এ না, ঘড়ি ধরে report: EventBridge Scheduler প্রতি ঘণ্টায় Lambda function (বা একটা Step Functions workflow) চালু করে (১৯.৫); queue লাগে না।
• নিয়ন্ত্রিত access: কোন analyst lake-এর কোন column আর কোন row দেখবে, সেটা ঠিক করে Lake Formation (১৭.৬)।

#Data lake-এর zone

দুটো bucket হলো একটা চেনা বিন্যাসের সবচেয়ে ছোট রূপ: একটা raw zone (যা এসেছে, কখনো বদলানো হয় না), একটা cleaned zone (যাচাই করা, Parquet, partition করা) আর একটা curated বা reporting zone (মানুষের জন্য হিসাব)। কাঁচা data অক্ষত রাখলে bug-এর পরে বাকি সব সবসময় আবার বানানো যায়।

#নিজে করে দেখো

একটা হাতে-কলমে অনুশীলন: AWS CLI দিয়ে ap-southeast-1-এ garments pipeline-এর একটা ছোট সংস্করণ বানাও। নামগুলোর শেষে -example; account 111122223333। Firehose, IoT আর Lambda-র role (firehose-meters-role, iot-meters-role, report-starter-role) আগে থেকেই আছে, প্রতিটা ধাপের দরকারি অনুমতিসহ, ধরে নিচ্ছি।

# 1. The stream, in on-demand mode (no shard planning)
aws kinesis create-stream --stream-name meters-example \
  --stream-mode-details StreamMode=ON_DEMAND

# 2. An IoT rule: every meter topic → the stream, meter ID as partition key,
#    with an error action into a backup bucket
aws iot create-topic-rule --rule-name meters_to_kinesis --topic-rule-payload '{
  "sql": "SELECT *, topic(4) AS meter_id FROM '\''factory/+/+/+'\''",
  "actions": [{"kinesis": {"streamName": "meters-example",
    "partitionKey": "${topic(4)}",
    "roleArn": "arn:aws:iam::111122223333:role/iot-meters-role"}}],
  "errorAction": {"s3": {"bucketName": "meter-iot-errors-example",
    "key": "${timestamp()}", "roleArn": "arn:aws:iam::111122223333:role/iot-meters-role"}}
}'

# 3. Firehose: read the stream, Lambda transform, Parquet, partition by day,
#    errors to errors/ (abridged: the Glue table meters.readings already exists)
aws firehose create-delivery-stream --delivery-stream-name meters-to-s3-example \
  --delivery-stream-type KinesisStreamAsSource \
  --kinesis-stream-source-configuration \
    KinesisStreamARN=arn:aws:kinesis:ap-southeast-1:111122223333:stream/meters-example,RoleARN=arn:aws:iam::111122223333:role/firehose-meters-role \
  --extended-s3-destination-configuration file://firehose-s3.json
# firehose-s3.json sets: BucketARN meter-raw-example, BufferingHints 128 MB / 300 s,
# Prefix data/day=!{partitionKeyFromQuery:day}/, ErrorOutputPrefix errors/!{firehose:error-output-type}/,
# ProcessingConfiguration (Lambda meter-clean-example), DataFormatConversionConfiguration (Parquet),
# DynamicPartitioningConfiguration Enabled=true

# 4. New files under data/ → SQS (the queue policy must allow s3.amazonaws.com)
aws s3api put-bucket-notification-configuration --bucket meter-raw-example \
  --notification-configuration '{"QueueConfigurations": [{
    "QueueArn": "arn:aws:sqs:ap-southeast-1:111122223333:new-files-example",
    "Events": ["s3:ObjectCreated:*"],
    "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": "data/"}]}}}]}'

# 5. Lambda polls the queue: at most 5 copies, partial batch failures
aws lambda create-event-source-mapping --function-name start-report-example \
  --event-source-arn arn:aws:sqs:ap-southeast-1:111122223333:new-files-example \
  --batch-size 10 --scaling-config MaximumConcurrency=5 \
  --function-response-types ReportBatchItemFailures

Starter function শুধু file-এর দিনের জন্য SQL বানায় আর query চালু করে:

import boto3, hashlib, json, urllib.parse
athena = boto3.client("athena")

SQL = """UNLOAD (SELECT floor, date_trunc('hour', ts) AS hour, sum(kwh) AS kwh
                 FROM meters.readings WHERE day = '{day}' GROUP BY 1, 2)
TO 's3://meter-reports-example/hourly/day={day}/run={run}/' WITH (format = 'PARQUET')"""

def handler(event, context):
    failures = []
    for msg in event["Records"]:
        try:
            for rec in json.loads(msg["body"]).get("Records", []):
                key = urllib.parse.unquote_plus(rec["s3"]["object"]["key"])
                day = key.split("day=")[1].split("/")[0]          # data/day=2026-10-29/...
                token = hashlib.sha256((key + rec["s3"]["object"]["eTag"]).encode()).hexdigest()[:64]
                athena.start_query_execution(
                    QueryString=SQL.format(day=day, run=token[:12]),
                    ClientRequestToken=token,                       # same file twice → one query
                    WorkGroup="reports-example")
        except Exception:
            failures.append({"itemIdentifier": msg["messageId"]})  # only this one is retried
    return {"batchItemFailures": failures}

কী হলো দেখে নাও:

# A query started by the function (ID from its log)
aws athena get-query-execution --query-execution-id 6f1c2b8e-1a2b-4c3d-9e8f-0a1b2c3d4e5f \
  --query 'QueryExecution.[Status.State,Statistics.DataScannedInBytes,Statistics.TotalExecutionTimeInMillis]'
# [
#     "SUCCEEDED",
#     412316860,
#     6420
# ]
# 412,316,860 bytes ≈ 0.38 GB scanned: one partition, three columns

aws s3 ls s3://meter-reports-example/hourly/day=2026-10-29/ --recursive
# 2026-10-30 00:07:12      18734 hourly/day=2026-10-29/run=9c1e5a7b2d40/20261030_000706_00042_abcde_0a1b2c3d.parquet

# Messages that failed three times
aws sqs get-queue-attributes --queue-url https://sqs.ap-southeast-1.amazonaws.com/111122223333/new-files-dlq-example \
  --attribute-names ApproximateNumberOfMessages
# { "Attributes": { "ApproximateNumberOfMessages": "0" } }

Finance-এর জন্য report Redshift Serverless-এ load করো:

COPY energy.hourly_floor
FROM 's3://meter-reports-example/hourly/day=2026-10-29/'
IAM_ROLE 'arn:aws:iam::111122223333:role/redshift-reports-role'
FORMAT AS PARQUET;

অনুশীলন: নতুন চাহিদা এল, “floor engineer-কে এক সেকেন্ডের মধ্যে alert দিতে হবে”। কোন ধাপ যোগ করবে, আর কোথায়? (একটা Lambda function বা Flink application, যেটা Kinesis stream পড়ে; Firehose-এর পরের কোনো কিছু না।) আর মালিকরা পরে যদি বলেন “এক ঘণ্টার চেয়ে দ্রুত কিছুই লাগবে না”? (Stream বাদ দাও: IoT Core → Firehose; SQS trigger-এর জায়গায় প্রতি ঘণ্টার schedule।)

পরিষ্কার করা: event source mapping, bucket notification, Firehose stream, IoT rule আর Kinesis stream delete করো, তারপর bucket-গুলো খালি করে delete করো।

#সারসংক্ষেপ

মূল কথাগুলো

• Classic serverless pipeline: IoT Core → Kinesis Data Streams → Firehose (+ Lambda) → S3 ingestion bucket → (SQS) → Lambda → Athena → S3 reporting bucket → QuickSight / Redshift। প্রতিটা ধাপ একটাই কাজ করে।
• Real time থাকে stream-এ (consumer পর্যন্ত প্রায় ২০০ ms)। Firehose থেকে সামনে সবকিছু near real time: S3 buffer ০–৯০০ সেকেন্ড, default ৩০০ সেকেন্ড; “প্রতি ১ মিনিটে” হলো ২০২৩-এর আগের সর্বনিম্ন মান।
• ভারী কাজ Firehose code ছাড়াই করে: Lambda transformation, Glue schema দিয়ে Parquet রূপান্তর, dynamic partitioning, আর ফেল করা record-এর জন্য error prefix।
• S3 আর Lambda-র মাঝে SQS ঐচ্ছিক, কিন্তু এনে দেয় ঢল সামলানো, maximum concurrency, retry আর DLQ। Lambda queue poll করে; subscribe করে না।
• Athena asynchronous, দাম scan হওয়া TB ধরে। Parquet + partition + কাঁচা data-র বদলে report বিল কমায় বহু গুণ। লম্বা কাজ → Step Functions।
• নিজেকে trigger করা loop এড়াতে report রাখো আলাদা bucket-এ, আর event দুবার আসতে পারে বলে report বানাও idempotent।
• Fully serverless মানে Redshift Serverless, provisioned cluster না; MSK provisioned আর EC2-এর ওপর EMR-ও নিয়মটা ভাঙে।
• চুপচাপ গোলমালগুলোর দিকে নজর রাখো: data freshness, error prefix, DLQ-তে কত message, দেরিতে আসা data আর ছোট file।

নিজেকে যাচাই করো: serverless big data ingestion pipeline৩১টি প্রশ্ন