Kinesis Data Streams

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

• Streaming data কী, আর stream কীভাবে queue থেকে আলাদা, তা বোঝাতে: record একটা retention period ধরে থেকে যায় (২৪ ঘণ্টা থেকে ৩৬৫ দিন), অনেক consumer একই record নিজের নিজের গতিতে পড়ে, আর যে কেউ আবার পড়তে (replay) পারে।
• Kinesis Data Streams-এর producer (SDK, KPL, Kinesis Agent, AWS service) আর consumer (নিজের app, KCL, Lambda, Data Firehose, Managed Service for Apache Flink) চিনতে।
• MB/s, record/s আর consumer-এর সংখ্যা থেকে একটা provisioned stream-এর মাপ হিসাব করতে, shard-এর সীমা ধরে: ঢুকতে 1 MB/s বা ১,০০০ record/s, বেরোতে 2 MB/s।
• Partition key দেখে বলে দিতে record কোন shard-এ যাবে, আর ক্রম ও duplicate নিয়ে Kinesis কী কথা দেয় আর কী দেয় না।
• ProvisionedThroughputExceededException আর hot shard-এর কারণ ধরতে, আর ঠিক সমাধান বাছতে: ভালো key, backoff দিয়ে retry, split, নাকি enhanced fan-out।
• Split, merge আর UpdateShardCount দিয়ে reshard করতে, provisioned আর on-demand capacity-র মধ্যে বাছতে, আর IAM, KMS ও VPC endpoint দিয়ে stream সুরক্ষিত করতে।

#Streaming data আর Kinesis

#Streaming data জিনিসটা কী

কিছু ডেটা কয়েকটা বড় file হয়ে আসে না; আসে ছোট ছোট event-এর এক অবিরাম স্রোত হয়ে, সেকেন্ডে হাজার হাজার, প্রতিটা কয়েকশো byte থেকে কয়েক kilobyte:

• Clickstream: website, app বা game-এ প্রতিটা page দেখা, প্রতিটা search, প্রতিটা tap।
• IoT telemetry: সিলেটের চা-বাগানের sensor থেকে তাপমাত্রা আর মাটির আর্দ্রতা, কিংবা delivery bike-গুলোর অবস্থান।
• হাজার হাজার server-এর metric আর log।
• ব্যবসার event: mobile wallet-এর প্রতিটা payment, food-delivery app-এর প্রতিটা order, চলতি cricket match-এর প্রতিটা বল।

এই ডেটার দাম সবচেয়ে বেশি প্রথম কয়েক সেকেন্ডে: payment settle হওয়ার আগেই fraud check চালাতে হবে, টেলিভিশনের আগেই live score ফোনে পৌঁছাতে হবে। ঘণ্টায় একবার batch file-এর জন্য বসে থাকলে দেরি হয়ে যায়। এটাকেই বলে real-time big data streaming, আর AWS-এ এর মূল service হলো Amazon Kinesis Data Streams (KDS): একটা পুরোপুরি managed service, যেটা streaming data real-time-এ সংগ্রহ করে জমা রাখে, আর অনেকগুলো application-কে সেটা process করতে দেয়।

Kinesis পরিবারে একসময় চারজন সদস্য ছিল। আজ তারা: Kinesis Data Streams (এই লেসন), Amazon Data Firehose (আগের নাম Kinesis Data Firehose, লেসন ১৩.৬), Amazon Managed Service for Apache Flink (আগের নাম Kinesis Data Analytics, অধ্যায় ১৭) আর Kinesis Video Streams (video-র জন্য, এখানে আলোচনা নেই)।

#Queue ভুলে যায়, stream মনে রাখে

SQS-এর সাথে তোমার পরিচয় হয়েছে লেসন ১৩.১-এ। Queue হলো কাজের তালিকা: প্রতিটা message একটা কাজ, একজন consumer সেটা নেয়, করে, তারপর মুছে দেয়। Stream তিনভাবে আলাদা, আর পরীক্ষা এই তিনটাই ঘুরেফিরে ধরে:

• পড়লে record মুছে যায় না। Record stream-এ থাকে retention period ধরে: default ২৪ ঘণ্টা, সর্বোচ্চ ৩৬৫ দিন। আলাদা করে একটা record তুমি নিজে মুছতে পারো না; মেয়াদ শেষ হলে Kinesis নিজেই সরিয়ে দেয়। (একবার লেখা ডেটা আর বদলানো যায় না।)
• অনেক consumer একই record পড়ে, প্রত্যেকে নিজের গতিতে। Live-score app, analytics job আর archiver সবাই প্রতিটা বল পড়ে; কেউ কারো কাছ থেকে record "নিয়ে" যায় না।
• Consumer আবার পড়তে পারে (replay)। প্রত্যেক consumer নিজের অবস্থান (একটা sequence number) নিজে মনে রাখে। Bug ঠিক করার পরে সে retention period-এর ভেতরের যেকোনো জায়গায় ফিরে গিয়ে আবার পড়তে পারে।

#Producer, record আর consumer

#Record দেখতে কেমন

Stream-এ যা লেখা হয়, তার প্রতিটা একটা record, তিন অংশে গড়া:

Record-এর মাপ। বহু বছর ধরে data blob-এর সর্বোচ্চ মাপ ছিল 1 MB, পরীক্ষার প্রশ্নে এখনো এই সংখ্যাই, আর প্রতিটা stream-এর default-ও এটা। ২০২৫-এর অক্টোবর থেকে একটা stream-এর সর্বোচ্চ record-এর মাপ 10 MiB পর্যন্ত বাড়ানো যায় (UpdateMaxRecordSize দিয়ে), মাঝে মাঝে আসা বড় record-এর জন্য, যেমন একটা বড় IoT snapshot; shard-এর সীমা কিন্তু বদলায় না, তাই একটানা বড় record পাঠানোর জন্য এটা না। সাধারণ ব্যবহার এখনো অনেক অনেক ছোট record।

#Producer

• AWS SDK: PutRecord একটা record লেখে; PutRecords এক call-এ ৫০০টা record পর্যন্ত লেখে (এক request-এ সর্বোচ্চ 10 MiB)। PutRecords সব-অথবা-কিছুই-না ধরনের না: কিছু record ব্যর্থ হতে পারে, বাকিগুলো সফল, তাই FailedRecordCount দেখে শুধু ব্যর্থগুলো আবার পাঠাও।
• Kinesis Producer Library (KPL): বেশি throughput-এর producer-এর জন্য একটা library। সে record জড়ো করে PutRecords call-এ পাঠায় (collection), অনেক ছোট user record একটা Kinesis record-এ ভরে দেয় (aggregation, এতে ১,০০০ record/s-এর সীমা বাঁচে), আর নিজেই retry করে। বিনিময়ে batch জমাতে একটু বাড়তি দেরি হয়, আর consumer-কে aggregation খুলতে হয় (KCL আর Lambda এটা নিজেরাই করে)।
• Kinesis Agent: Linux server-এ চলা একটা আলাদা Java program, যেটা log file-এর দিকে নজর রাখে আর নতুন লাইন stream-এ (বা Firehose-এ) পাঠায়।
• AWS service, যারা তোমার হয়ে stream-এ লেখে: CloudWatch Logs subscription filter, AWS IoT Core rule, DMS (database-এর change data capture, CDC), DynamoDB (change data capture একটা stream-এ), সামনের দরজা হিসেবে API Gateway, আরও অনেকে।

#Consumer

• SDK দিয়ে নিজের application (GetShardIterator, তারপর loop-এ GetRecords)।
• EC2, container বা অফিসের server-এ চলা Kinesis Client Library (KCL) application: scale করতে পারে এমন consumer লেখার প্রচলিত উপায় (বিস্তারিত নিচে)।
• Event source mapping দিয়ে AWS Lambda: কোনো server নেই, shard ধরে scale হয়।
• Stream-কে source বানিয়ে Amazon Data Firehose: কোনো consumer code ছাড়াই record S3, Redshift, OpenSearch, Splunk আর HTTP endpoint-এ পৌঁছে দেয় (লেসন ১৩.৬)।
• Amazon Managed Service for Apache Flink: stream-এর ওপর SQL বা Java/Python দিয়ে processing (window, join, অস্বাভাবিক কিছু ধরা) (অধ্যায় ১৭)।

#Shard: capacity-র একক

একটা stream এক বা একাধিক shard দিয়ে তৈরি। Shard হলো record-এর একটা ক্রমিক খাতা, যার capacity নির্দিষ্ট, আর stream-এর capacity তার সব shard-এর যোগফল।

#প্রতি shard-এর সীমা

দুটো খুঁটিনাটিতে মানুষ ধরা খায়। প্রথমত, লেখার সীমার দুটো অংশ: অনেক ছোট ছোট record এলে 1 MB/s-এর অনেক আগেই ১,০০০ record/s-এ পৌঁছে যায়। দ্বিতীয়ত, 2 MB/s shared পড়ার সীমা ওই shard-এর সব সাধারণ consumer ভাগ করে নেয়: তিনটা app প্রত্যেকে সব পড়লে তারা এই 2 MB/s ভাগাভাগি করে।

#কয়টা shard লাগবে: মাপার সূত্র

Provisioned stream-এর জন্য প্রতিটা সীমার হিসাবে কয়টা shard লাগে বের করো, তারপর সবচেয়ে বড়টা নাও:

write MB/s      = records/s × average record size
shards (MB)     = ceil(write MB/s ÷ 1)
shards (count)  = ceil(records/s ÷ 1,000)
shards (reads)  = ceil(write MB/s × number of shared consumers ÷ 2)
shards needed   = max of the three

হাতে-কলমে উদাহরণ:

• Note-এর উদাহরণ: একটা app সেকেন্ডে 3 MB ডেটা বানায়। প্রতিটা shard 1 MB/s নেয়, তাই কমপক্ষে ৩টা shard লাগবে।
• ছোট record: সেকেন্ডে ৫,০০০ sensor reading, প্রতিটা ২০০ byte, মোটে মাত্র প্রায় 1 MB/s, কিন্তু ৫,০০০ record/s-এর জন্য লাগবে ৫টা shard। KPL দিয়ে দশটা reading একটা record-এ ভরলে হয় ৫০০ record/s, আর লাগে ১টা shard।
• অনেক পাঠক: 2 MB/s লেখা হচ্ছে, আর তিনটা shared consumer সব পড়ছে, মানে পড়া 6 MB/s: 6 ÷ 2 = ৩টা shard, যদিও শুধু লেখার জন্য লাগত ২টা। অথবা ২টা shard রেখে consumer-দের enhanced fan-out দাও।

হিসাব করো সর্বোচ্চ চাপ (peak) ধরে, গড় ধরে না, সাথে কিছুটা বাড়তি জায়গা রাখো, আর মনে রেখো একটা partition key কখনো এক shard-এর বেশি ব্যবহার করতে পারে না।

#Partition key আর ক্রম

#Key কীভাবে shard বেছে নেয়

Producer কখনো shard বাছে না। সে পাঠায় একটা partition key, আর Kinesis key-টাকে MD5 hash-এর ভেতর দিয়ে চালিয়ে একটা ১২৮-bit সংখ্যা পায়, এটাই hash key। প্রতিটা shard hash key-এর একটা টানা range-এর মালিক, আর record যায় সেই shard-এ, যার range-এর মধ্যে তার hash পড়েছে। (Producer চাইলে একটা explicit hash key দিয়ে hash-টা নিজে ঠিক করে দিতে পারে, তবে সেটা কালেভদ্রে।)

মনে রাখার নিয়ম: একই partition key-এর record সবসময় একই shard-এ যায় (যতক্ষণ না reshard করো)। অনেক আলাদা key সব shard-এ ছড়িয়ে পড়ে, আর এই ছড়ানোর জন্যই একটা stream তার পুরো capacity কাজে লাগাতে পারে।

#Kinesis কোন ক্রমের কথা দেয়

এক shard-এর ভেতরে record জমা হয় আর পড়া হয় ঠিক যে ক্রমে লেখা হয়েছে (sequence number ধরে)। Shard-এর মাঝে কোনো ক্রম নেই: দুটো shard মানে দুটো আলাদা খাতা, পড়া হয় একসাথে।

তাই Kinesis ক্রমের নিশ্চয়তা দেয় প্রতি shard-এ, বাস্তবে মানে প্রতি partition key-তে, পুরো stream জুড়ে কখনো না। Key এমনভাবে বাছো, যেন ঠিক যেসব record-এর ক্রম রাখতেই হবে, তারাই একই key পায়: একটা wallet-এর সব লেনদেন, একটা sensor-এর সব reading, একটা order-এর সব event। যদি সত্যিই পুরো system জুড়ে একটাই ক্রম লাগে, তাহলে লাগবে একটা key, একটা shard, আর তাই এক shard-এর সীমা (1 MB/s, ১,০০০ record/s)।

#Retry, ক্রম ওলটপালট আর duplicate

Kinesis পৌঁছে দেয় অন্তত একবার (at least once), আর দুই দিক থেকেই চমক আসতে পারে:

• Producer-এর retry ক্রম বদলে দেয়। একই key-এর A তারপর B নিয়ে একটা PutRecords batch-এ যদি শুধু A ব্যর্থ হয়, A আবার পাঠালে সে বসে B-এর পরে। ক্রম জরুরি হলে একটা key-এর record একটার পর একটা পাঠাও, আর পরেরটা পাঠানোর আগে ব্যর্থটার retry শেষ করো (এক producer থেকে কড়া ক্রমের জন্য PutRecord SequenceNumberForOrdering-ও নেয়)।
• Producer-এর retry duplicate বানায়। Write সফল হলো কিন্তু response হারিয়ে গেল (mobile network কেটে গেল), producer আবার পাঠাল, record জমা হলো দুবার।
• Consumer-এর retry duplicate বানায়। Process করার পরে কিন্তু checkpoint করার আগে crash করা consumer ওই record-গুলো আবার পড়ে।

তাই প্রতিটা record-এ একটা অনন্য ID রাখো (transaction ID, event ID), আর consumer-কে idempotent বানাও: একই record দুবার process করলেও ফল যেন একবারের মতোই হয়।

#Hot shard আর ProvisionedThroughputExceededException

#এই error-এর মানে

Producer কোনো shard-এ 1 MB/s বা ১,০০০ record/s-এর বেশি লিখলে, কিংবা consumer-রা 2 MB/s বা সেকেন্ডে ৫টা GetRecords call-এর বেশি পড়লে, Kinesis বাড়তি request-গুলো ফিরিয়ে দেয় ProvisionedThroughputExceededException দিয়ে। Shard throttle হয়েছে, নষ্ট হয়নি: সে সীমা পর্যন্ত traffic নিতেই থাকে, আর ফিরিয়ে দেওয়া request-গুলো আবার পাঠাতে হয়। CloudWatch-এ এটা দেখা যায় WriteProvisionedThroughputExceeded আর ReadProvisionedThroughputExceeded হিসেবে।

পুরো stream-ই যদি ছোট হয়, সমাধান আরও shard (split, বা UpdateShardCount)। কিন্তু প্রায়ই দেখা যায় পুরো stream-এ জায়গা খালি আছে, শুধু একটা shard-এর ওপর চাপ বেশি।

#Hot shard

সমাধানগুলো, সবচেয়ে ভালো থেকে শেষ উপায় পর্যন্ত:

1. ভালো partition key বাছো: অনেক আলাদা মান, কোনোটাই বাকিদের চেয়ে অনেক বেশি ব্যস্ত না।
2. যে key রাখতেই হবে, তার পেছনে একটা random suffix জুড়ে দাও (Bangladesh-0 … Bangladesh-9): চাপ দশটা hash মানে ছড়িয়ে যায়, তবে ওই key-এর ক্রম আর থাকে না।
3. Exponential backoff আর jitter দিয়ে retry করো (SDK আর KPL নিজেরাই করে), যাতে ছোটখাটো চাপের ঢেউ হারিয়ে না গিয়ে সামলে যায়।
4. সমস্যা যদি record/s-এর সীমা হয়, PutRecords আর KPL দিয়ে batch আর aggregate করো।
5. Hot shard-টা split করো, তবে এতে কাজ হয় শুধু যখন shard-টায় অনেকগুলো ব্যস্ত key আছে। একটা বিশাল key হলো hash range-এর একটা বিন্দু: split করলে সে এক child-এ যায়, আর সেই child-ও ঠিক ততটাই hot।
6. Read throttle হলে enhanced fan-out নাও, অথবা একই shard poll করা consumer কমাও।

চেনার উপায়: পুরো stream-এর IncomingBytes capacity-র অনেক নিচে, অথচ throttle হচ্ছে। Enhanced (shard-level) monitoring চালু করো, তাহলে প্রতি shard-এর metric দেখে hot-টাকে খুঁজে পাবে।

#Resharding: split আর merge

Traffic সবসময় এক থাকে না। একটা provisioned stream-এর shard সংখ্যা বদলানোকে বলে resharding:

• Shard split (scale out)। একটা parent shard নতুন একটা starting hash key-তে ভেঙে দুটো child হয়; প্রত্যেক child parent-এর hash range-এর একটা অংশ পায়। ওই key-গুলোর capacity দ্বিগুণ হয়, আর stream-এর ওই অংশের ঘণ্টার খরচও।
• Shard merge (scale in)। পাশাপাশি দুটো shard (যাদের hash range গায়ে গায়ে লাগানো) মিলে একটা child হয়। Capacity আর খরচ দুটোই কমে। যাদের range পাশাপাশি না, এমন দুটো shard merge করা যায় না।

Parent-দের কী হয়: তারা open থেকে closed হয় (নতুন write নেয় না; নতুন record যায় child-এ), আর retention period পেরোলে শেষে expired হয়। ততদিন তাদের পুরোনো record পড়া যায়। এজন্যই ক্রম নিয়ে সাবধান হতে হয়: split-এর পরে একটা key-এর পুরোনো record থাকে parent-এ, আর নতুনগুলো child-এ। Consumer-কে child পড়ার আগে parent পুরো শেষ করতে হবে; KCL এটা তোমার হয়েই করে।

একটা একটা split বা merge না করে তুমি একটা লক্ষ্য সংখ্যা দিয়ে UpdateShardCount call করতে পারো, আর কাজটা Kinesis-কে করতে দিতে পারো (uniform scaling)। প্রতি stream-এ এর default সীমা: এক call-এ বর্তমান সংখ্যার সর্বোচ্চ দ্বিগুণ বা সর্বনিম্ন অর্ধেক, যেকোনো ২৪ ঘণ্টায় সর্বোচ্চ ১০ বার scaling, আর প্রতি stream-এ ১০,০০০ shard পর্যন্ত। Stream UPDATING থাকার সময়ও producer আর consumer কাজ চালিয়ে যায়।

Provisioned mode-এ resharding তোমার দায়িত্ব: console থেকে হাতে, script দিয়ে, বা নিজের বানানো automation দিয়ে (একটা CloudWatch alarm, যেটা একটা Lambda function চালায়, আর সেটা UpdateShardCount call করে)। On-demand mode কাজটা নিজেই করে।

#Capacity mode: provisioned আর on-demand

Note-এ resharding-কে হাতের কাজ বলা হয়েছে। এটা সত্যি শুধু provisioned mode-এর জন্য; ২০২১ থেকে on-demand mode-ও আছে।

একটা stream দুই mode-এর মধ্যে ২৪ ঘণ্টায় দুবার বদলানো যায়, কোনো downtime ছাড়াই। On-demand mode-এও একটা partition key এক shard-এর 1 MB/s আর ১,০০০ record/s-এর বেশি পায় না, তাই hot key তখনও throttle হয়। Traffic বাড়লে on-demand mode shard-গুলো সমান ভাগে split করে, কিন্তু একটা ব্যস্ত hash range আলাদা করে খুঁজে বের করে না। তাই key খুব অসমান হলে AWS নিজেই provisioned mode নিতে বলে, যেখানে ঠিক যে shard-টা hot, শুধু সেটাকেই তুমি split করতে পারো। আরেকটা কথা: default-এ একটা account-এ সর্বোচ্চ ৫০টা on-demand stream রাখা যায় (বাড়ানো যায়); provisioned stream-এর সংখ্যায় এমন কোনো সীমা নেই।

বিনা পয়সার ২৪ ঘণ্টার পরে retention-এর জন্য বাড়তি টাকা লাগে: ৭ দিন পর্যন্ত extended retention-এর বিল হয় প্রতি shard-hour-এ (provisioned) বা প্রতি GB-তে, আর ৭ দিনের পরে ৩৬৫ দিন পর্যন্ত long-term retention-এর বিল প্রতি GB-month-এ, সাথে long-term ডেটা GetRecords দিয়ে পড়লে প্রতি GB-র একটা charge।

#Consumer-দের ভেতরের কথা

#Shared throughput আর enhanced fan-out

মোটা দাগের নিয়ম: এক-দুজন consumer, যাদের একটু দেরিতে সমস্যা নেই → shared; তিন বা তার বেশি consumer, কিংবা সবচেয়ে কম latency লাগে এমন consumer → enhanced fan-out।

#Kinesis Client Library (KCL)

Raw GetRecords call দিয়ে stream পড়া মানে shard, অবস্থান, বিকল হওয়া আর resharding সব নিজের হাতে সামলানো। KCL এর সবটা করে দেয়:

• প্রতিটা shard একসময়ে ঠিক একজন worker-কে দেয় একটা lease-এর মাধ্যমে; একজন worker কয়েকটা lease ধরে রাখতে পারে।
• Lease আর checkpoint (কোন shard কোন sequence number পর্যন্ত process হয়েছে) থাকে একটা DynamoDB lease table-এ, প্রতি KCL application-এর জন্য একটা। এজন্যই KCL app-এর DynamoDB permission লাগে, আর DynamoDB-র সামান্য খরচ যোগ হয়।
• কোনো worker মারা গেলে আরেকজন worker তার lease নিয়ে নেয়, আর চালিয়ে যায় শেষ checkpoint থেকে। ওই checkpoint-এর পরে যা process হয়েছিল, সেগুলো আবার process হয়, তাই processing idempotent হতে হবে।
• Resharding-এর পরে সে child shard-এর আগে parent shard পড়ে, তাতে প্রতি key-এর ক্রম ঠিক থাকে।
• Version 2 থেকে সে default-এ enhanced fan-out ব্যবহার করে, আর KCL 3 lease ভাগ করে worker-এর CPU ব্যবহার দেখে।
• এটা একটা Java library; অন্য ভাষা (Python, Node.js, .NET, Ruby) এটা ব্যবহার করে MultiLangDaemon দিয়ে।

যেহেতু এক shard একজন worker-ই পড়ে, shard-এর চেয়ে বেশি worker বৃথা: ৪টা shard থাকলে পঞ্চম worker-এর হাতে কোনো lease নেই। দ্রুত process করতে চাইলে shard বাড়াও (অথবা Lambda হলে parallelization factor বাড়াও)।

#Consumer হিসেবে Lambda

Lambda-র একটা event source mapping প্রতিটা shard poll করে (অথবা enhanced fan-out দিয়ে subscribe করে), আর record-এর একটা batch দিয়ে তোমার function চালায়: default-এ সর্বোচ্চ ১০০টা record, বাড়িয়ে ১০,০০০ পর্যন্ত, চাইলে ৫ মিনিট পর্যন্ত একটা batching window সহ।

• এক shard-এর batch process হয় একটার পর একটা, তাই Lambda প্রতি shard-এর ক্রম ঠিক রাখে। Default-এ প্রতি shard-এ একটা invocation; parallelization factor ১০ পর্যন্ত বাড়ালে এক shard থেকে একসাথে ১০টা পর্যন্ত batch চলে, তবু Lambda প্রতিটা partition key-এর ক্রম ঠিক রাখে।
• Function error ছুড়লে Lambda default-এ পুরো batch আবার চেষ্টা করে, যতক্ষণ না সফল হয় বা record-এর মেয়াদ শেষ হয়, আর ততক্ষণ ওই shard আটকে থাকে। একটা মাত্র খারাপ record (poison pill) একটা shard-কে পুরো একদিন আটকে রাখতে পারে। সমাধান: bisect batch on error, একটা maximum retry attempts মান, একটা maximum record age, আর একটা on-failure destination (SQS, SNS বা S3), যেখানে ব্যর্থ batch-এর বিবরণ যায়। Lambda-র event source mapping বিস্তারিত আছে অধ্যায় ১৫-এ।

#Firehose, Flink আর MSK

• Stream থেকে পড়া Data Firehose হলো code ছাড়াই একই record S3, Redshift বা OpenSearch-এ batch করে নামানোর উপায় (প্রায় real-time)। প্রচলিত একটা design: real-time consumer-দের জন্য একটা stream, আর archive-এর জন্য আরেকজন consumer হিসেবে Firehose। Firehose আর Data Streams-এর বিস্তারিত তুলনা আছে লেসন ১৩.৬-এ।
• Managed Service for Apache Flink stream-এর ওপর একটানা query আর state ধরে রাখা processing চালায়।
• Amazon MSK (managed Apache Kafka) হলো আরেকটা streaming service, যখন কোনো team আগে থেকেই Kafka-র API আর tool ব্যবহার করে; এটা আলাদা service, Kinesis-এর consumer না। দুটোই আছে অধ্যায় ১৭-এ।

#নিরাপত্তা

• IAM দিয়ে access নিয়ন্ত্রণ। Producer-এর লাগে stream ARN-এর ওপর kinesis:PutRecord/PutRecords; consumer-এর লাগে GetRecords, GetShardIterator, DescribeStream/ListShards (আর enhanced fan-out-এর জন্য SubscribeToShard)। EC2, Lambda আর container-কে IAM role দাও, code-এ কখনো access key না (অধ্যায় ২)।
• Resource-based policy (২০২৩ থেকে) একটা stream বা consumer-এর গায়ে লাগিয়ে আরেকটা account-কে সরাসরি লেখা বা পড়ার অনুমতি দেওয়া যায়, S3 bucket policy-র মতো; পুরোনো বিকল্প হলো cross-account IAM role।
• চলার পথে encryption: প্রতিটা API call চলে HTTPS (TLS) দিয়ে।
• জমা থাকা অবস্থায় encryption: KMS দিয়ে server-side encryption, AWS managed key aws/kinesis বা তোমার নিজের customer managed key দিয়ে। তখন producer আর consumer-এর key-টা ব্যবহারের অনুমতিও লাগে (KMS key-এর ব্যাখ্যা লেসন ১০.৮-এ)।
• Client-side encryption-ও করা যায়: PutRecord-এর আগে তুমি encrypt করো, consumer-এ decrypt করো।
• Private access: একটা interface VPC endpoint (PrivateLink) private subnet-এর producer আর consumer-কে internet gateway বা NAT gateway ছাড়াই Kinesis-এ পৌঁছাতে দেয় (অধ্যায় ২২)।
• Audit: control-plane call (create, delete, reshard, encryption বদল) CloudTrail-এ লেখা থাকে।

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

• জায়গা খালি থাকা সত্ত্বেও throttle: hot partition key। Shard সংখ্যা না, key ঠিক করো।
• Stream আসলেই ছোট বলে throttle: shard বাড়াও (split, UpdateShardCount), নয়তো on-demand-এ যাও।
• অনেক poll করা consumer, read throttle আর বাড়তে থাকা iterator age: তাদের enhanced fan-out-এ নাও।
• Consumer পিছিয়ে পড়ল আর ডেটার মেয়াদ শেষ: retention খুব ছোট, বা consumer তার চেয়ে বেশি সময় বন্ধ ছিল। GetRecords.IteratorAgeMilliseconds (বা Lambda-র IteratorAge metric)-এর ওপর alarm বসাও, আর retention বাড়াও।
• একটা খারাপ record-এ আটকে থাকা Lambda consumer: bisect on error, সর্বোচ্চ retry, সর্বোচ্চ record age আর on-failure destination ঠিক করো।
• Resharding-এর পরে ক্রম ওলটপালট: নিজের বানানো consumer parent শেষ না করেই child পড়ছে। KCL ব্যবহার করো।
• Retry-এর পরে ক্রম ওলটপালট বা duplicate: আংশিক ব্যর্থ PutRecords ভুল ক্রমে আবার পাঠানো, হারানো response-এর পরে আবার পাঠানো। একটা key-এর record ক্রমে পাঠাও, আর consumer idempotent বানাও।
• অনেকগুলো KCL worker কিছুই করছে না: shard সংখ্যার বাড়তি worker-দের হাতে lease নেই।
• Encrypted stream-এ AccessDeniedException: role Kinesis ব্যবহার করতে পারে, কিন্তু KMS key না।
• হঠাৎ বড় বিল: শেষ হয়ে যাওয়া peak-এর জন্য রেখে দেওয়া shard, লম্বা retention, কেউ ব্যবহার করে না এমন enhanced fan-out consumer।

#System design-এ Kinesis Data Streams

প্রচলিত ধরনগুলো:

• Real-time analytics + archive: producer → Kinesis Data Streams → (real-time-এর জন্য Lambda / KCL / Flink) + (data lake-এর জন্য Firehose → S3)।
• প্রতিটা entity ধরে event sourcing আর ক্রমিক processing: partition key = account ID, order ID বা device ID।
• Log আর metric এক জায়গায় জড়ো করা: CloudWatch Logs subscription → stream → processing → OpenSearch।
• Change data capture: DMS বা DynamoDB → stream → consumer, যারা search index আর cache মিলিয়ে রাখে।
• ধীর system-এর সামনে event-এর বন্যা সামলানো: SQS-এর মতো না, ধীর system পড়ে ফেলার পরেও stream ডেটা রেখে দেয়, তাই producer না বদলেই পরে আরেকজন consumer যোগ করা যায়।

কখন Kinesis ভুল উত্তর: স্বাধীন কাজ, যার প্রতিটা যেকোনো একজন worker একবার করলেই হলো (SQS); মানুষ আর endpoint-এ notification push করা (SNS); কোনো custom code ছাড়া S3 বা Redshift-এ ডেটা নামানো, যেখানে replay বা সেকেন্ডের কম latency লাগে না (শুধু Firehose)।

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

এই command-গুলো AWS CLI v2-এর। Kinesis প্রতি shard-hour-এ (on-demand হলে প্রতি stream-hour-এ) টাকা নেয়, তাই শেষে stream মুছে দিয়ো। CLI v2-এ --data base64 ধরে নেয়, যদি না --cli-binary-format raw-in-base64-out যোগ করো।

# ১. ২টা shard-এর একটা provisioned stream বানাও, তারপর ACTIVE হওয়া পর্যন্ত অপেক্ষা করো
aws kinesis create-stream --stream-name scores-example --shard-count 2
aws kinesis wait stream-exists --stream-name scores-example
aws kinesis describe-stream-summary --stream-name scores-example
# {
#   "StreamDescriptionSummary": {
#     "StreamName": "scores-example",
#     "StreamARN": "arn:aws:kinesis:ap-southeast-1:111122223333:stream/scores-example",
#     "StreamStatus": "ACTIVE",
#     "StreamModeDetails": { "StreamMode": "PROVISIONED" },
#     "RetentionPeriodHours": 24,
#     "OpenShardCount": 2,
#     "EncryptionType": "NONE", ...
#   }
# }

# ২. একটা record লেখো; একই partition key সবসময় একই shard-এ যায়
aws kinesis put-record --stream-name scores-example --partition-key match-77 \
  --data '{"over":41.1,"runs":4}' --cli-binary-format raw-in-base64-out
# { "ShardId": "shardId-000000000001",
#   "SequenceNumber": "49650925471223049381123456789012345678901234567890" }

# ৩. একটা batch লেখো, আর আংশিক ব্যর্থতা আছে কিনা দেখো
aws kinesis put-records --stream-name scores-example --cli-binary-format raw-in-base64-out \
  --records Data='{"over":41.2}',PartitionKey=match-77 Data='{"over":41.3}',PartitionKey=match-77
# { "FailedRecordCount": 0, "Records": [ { "SequenceNumber": "...", "ShardId": "shardId-000000000001" }, ... ] }

# ৪. Shard-গুলো আর তাদের hash key range দেখো
aws kinesis list-shards --stream-name scores-example \
  --query 'Shards[].[ShardId,HashKeyRange.StartingHashKey,HashKeyRange.EndingHashKey]' --output table

# ৫. Shard-এ এখনো থাকা সবচেয়ে পুরোনো record থেকে পড়ো (TRIM_HORIZON)
IT=$(aws kinesis get-shard-iterator --stream-name scores-example \
  --shard-id shardId-000000000001 --shard-iterator-type TRIM_HORIZON \
  --query ShardIterator --output text)
aws kinesis get-records --shard-iterator "$IT" --limit 10
# { "Records": [ { "SequenceNumber": "...", "PartitionKey": "match-77",
#                  "Data": "eyJvdmVyIjo0MS4xLCJydW5zIjo0fQ==", ... } ],
#   "NextShardIterator": "...", "MillisBehindLatest": 0 }
# Data base64-এ: echo eyJvdmVyIjo0MS4xLCJydW5zIjo0fQ== | base64 -d  →  {"over":41.1,"runs":4}

# ৬. বদলে একটা নির্দিষ্ট সময় থেকে replay করো
aws kinesis get-shard-iterator --stream-name scores-example --shard-id shardId-000000000001 \
  --shard-iterator-type AT_TIMESTAMP --timestamp 2026-10-26T09:00:00Z

# ৭. Reshard: shard দ্বিগুণ করো (uniform scaling), অথবা নিজে একটা shard split করো
aws kinesis update-shard-count --stream-name scores-example --target-shard-count 4 \
  --scaling-type UNIFORM_SCALING
# { "StreamName": "scores-example", "CurrentShardCount": 2, "TargetShardCount": 4 }

# ৮. ডেটা ৭ দিন রাখো, KMS দিয়ে encrypt করো, একটা enhanced fan-out consumer register করো
aws kinesis increase-stream-retention-period --stream-name scores-example --retention-period-hours 168
aws kinesis start-stream-encryption --stream-name scores-example \
  --encryption-type KMS --key-id alias/aws/kinesis
aws kinesis register-stream-consumer --consumer-name live-app \
  --stream-arn arn:aws:kinesis:ap-southeast-1:111122223333:stream/scores-example

# ৯. On-demand-এ বদলাও (২৪ ঘণ্টায় দুবার পর্যন্ত), তারপর সব মুছে দাও
aws kinesis update-stream-mode --stream-mode-details StreamMode=ON_DEMAND \
  --stream-arn arn:aws:kinesis:ap-southeast-1:111122223333:stream/scores-example
aws kinesis delete-stream --stream-name scores-example --enforce-consumer-deletion

Console-এও চেষ্টা করো: একটাই partition key দিয়ে loop-এ record পাঠাতে পাঠাতে stream-এর Monitoring tab খোলো, আর দেখো একটা shard-এর incoming bytes বাড়ছে (আগে enhanced shard-level metric চালু করে নিয়ো)। Data viewer tab code ছাড়াই প্রতি shard-এর record দেখায়।

#সারাংশ

মূল কথাগুলো

• Kinesis Data Streams real-time streaming data (clickstream, IoT, log, event) সংগ্রহ করে জমা রাখে। SQS-এর মতো না, পড়লে record মোছে না: থাকে retention period ধরে (default ২৪ ঘণ্টা, সর্বোচ্চ ৩৬৫ দিন), অনেক consumer একই ডেটা নিজের গতিতে পড়ে, আর replay করতে পারে। মেয়াদ শেষের আগে record মোছা যায় না।
• একটা record = partition key + data blob + sequence number। ডেটার সর্বোচ্চ মাপ 1 MB (পরীক্ষার সংখ্যা আর default; ২০২৫ থেকে 10 MiB পর্যন্ত বাড়ানো যায়)।
• Producer: SDK (PutRecord, ৫০০ পর্যন্ত PutRecords), KPL (batching, aggregation, retry), Kinesis Agent, AWS service। Consumer: SDK, KCL, Lambda, Data Firehose, Managed Service for Apache Flink।
• Shard-এর সীমা: ঢুকতে 1 MB/s বা ১,০০০ record/s, বেরোতে 2 MB/s (সাধারণ consumer-দের ভাগাভাগি, সেকেন্ডে ৫টা GetRecords) অথবা enhanced fan-out-এ প্রতি consumer-এর 2 MB/s (push, প্রায় ৭০ ms)। দরকারি shard = max(MB/s ÷ ১, record/s ÷ ১,০০০, shared পড়া ÷ ২)। 3 MB/s → ৩টা shard।
• Partition key → MD5 → hash range → shard। একই key, একই shard। ক্রমের নিশ্চয়তা শুধু প্রতি shard-এ, পুরো stream জুড়ে কখনো না। পৌঁছানো অন্তত একবার: consumer idempotent বানাও।
• ProvisionedThroughputExceededException = shard-এর সীমা পার; shard throttle হয়েছে, crash না। পুরো stream ছোট → shard বাড়াও; hot shard (দেশের নাম key, ৯৫% বাংলাদেশ) → অনেক মানের key নাও (user ID, transaction ID); backoff দিয়ে retry। একটা বিশাল key-কে split দিয়ে সারানো যায় না।
• Resharding: split (১ → ২, বেশি capacity আর খরচ), merge (পাশাপাশি ২ → ১)। Parent closed হয়, তারপর expired; child-এর আগে parent পড়ো (KCL পড়ে)। UpdateShardCount: এক call-এ ×২ বা ÷২, ২৪ ঘণ্টায় ১০ বার।
• Capacity mode: provisioned (shard তুমি বাছো, shard-hour ধরে দাম, হাতে scale) বনাম on-demand (planning নেই, শুরু 4 MB/s, ৩০ দিনের peak-এর দ্বিগুণ পর্যন্ত scale, stream-hour আর GB ধরে দাম)। ২৪ ঘণ্টায় দুবার বদলানো যায়।
• KCL: lease দিয়ে প্রতি shard-এ একজন worker, checkpoint থাকে DynamoDB-তে, shard-এর বেশি worker কাজে লাগে না। Lambda: প্রতি shard-এ ক্রম, parallelization factor ১০ পর্যন্ত, poison record-এর বিরুদ্ধে bisect on error আর on-failure destination।
• নিরাপত্তা: IAM (role), cross-account-এর জন্য resource-based policy, চলার পথে HTTPS, জমা অবস্থায় KMS, private access-এর জন্য interface VPC endpoint।

নিজেকে যাচাই করো: Kinesis Data Streams৩২টি প্রশ্ন