Managed Flink আর Amazon MSK

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

• বোঝাতে, শুধু stream যা পারে না, একটা stream processor তা কী করে: filter, window, join আর state, কয়েক সেকেন্ডের মধ্যে ফল।
• Amazon Managed Service for Apache Flink চেনাতে: এর পুরোনো নাম, এর source (Kinesis Data Streams আর MSK, Data Firehose কখনো না), sink, ভাষা, KPU, automatic scaling, checkpoint আর snapshot।
• পরীক্ষার জন্য যতটুকু লাগে, ততটুকু Apache Kafka বোঝাতে: broker, topic, partition, offset, producer, consumer group, replication আর retention।
• Amazon MSK তোমার হয়ে কী কী সামলায় (broker, ZooKeeper বা KRaft controller, Multi-AZ, recovery, EBS storage) তা বলতে, আর Standard broker, Express broker আর MSK Serverless-এর মধ্যে বাছতে।
• MSK-এর consumer-দের চিনতে (Flink, Glue Streaming ETL, Lambda, Firehose, MSK Connect, EC2, ECS আর EKS-এর app), আর কোন কাজে কোনটা, তা ঠিক করতে।
• Kinesis Data Streams আর MSK পাশাপাশি তুলনা করতে (message-এর মাপ, shard বনাম partition, scaling, encryption, retention, চালানোর ঝামেলা), আর “Kinesis না Kafka?” ধরনের প্রশ্নের উত্তর দিতে।

#Stream জমা রাখা আর stream নিয়ে ভাবা এক জিনিস না

#Stream processor বাড়তি কী করে

অধ্যায় ১৩-তে তুমি Kinesis-এর সেই দুটো service দেখেছ, যারা streaming data বয়ে নিয়ে যায়। Kinesis Data Streams (লেসন ১৩.৫) record-গুলো shard-এ জমা রাখে, যাতে অনেক consumer সেগুলো পড়তে আর আবার পড়তে পারে। Amazon Data Firehose (লেসন ১৩.৬) record জড়ো করে S3, Redshift বা OpenSearch-এ পৌঁছে দেয়। কিন্তু এদের কেউই ডেটা নিয়ে ভাবে না। “গত এক মিনিটে মিরপুর থেকে কতগুলো ride request এসেছে, আর সেখানে খালি driver কতজন?”, এই প্রশ্নের উত্তর এদের কেউ দিতে পারে না।

Stream processor হলো এমন একটা program, যেটা stream-এর ওপর সারাক্ষণ চলে আর এই ভাবনার কাজটা করে:

• Filter আর রূপান্তর: test event বাদ দেওয়া, phone number ঢেকে দেওয়া, কাঁচা ping থেকে পরিষ্কার সারি বানানো।
• Window ধরে হিসাব: প্রতি মিনিটে, প্রতি পাঁচ মিনিটে, প্রতি session-এ count, যোগফল, গড়।
• Join: একটা payment-কে ঠিক আগের login-এর সাথে মেলানো, কিংবা এক stream-এর ping-কে আরেক stream-এর trip-এর সাথে।
• ধরন খোঁজা: দুই মিনিটের মধ্যে তিনটা জেলা থেকে পাঁচবার ভুল PIN।
• State রাখা: চলতি যোগফল আর সাম্প্রতিক ইতিহাস, যা এক event থেকে পরের event পর্যন্ত মনে থাকে।

উত্তর আসে কয়েক সেকেন্ডে, যখন event-টা এখনো কাজের। Stream-এর ওপর একটা Lambda function batch ধরে ধরে সহজ কাজ করতে পারে, কিন্তু এক invocation থেকে আরেক invocation-এ সে সব ভুলে যায়; অনেকগুলো event জুড়ে window আর state লাগলে দরকার stream processor। AWS-এ সেটা হলো Amazon Managed Service for Apache Flink।

#যেসব নাম চোখে পড়বে

#Amazon Managed Service for Apache Flink

#Service-টা আসলে কী

Apache Flink হলো data stream process করার একটা open-source framework: তুমি একটা job লেখো, যেটা এক বা একাধিক source পড়ে, event-গুলো বদলায়, আর এক বা একাধিক sink-এ লেখে; Flink সেটা অনেক worker-এ parallel-এ চালায়, আর state নিরাপদে রাখে। নিজে Flink চালাতে গেলে লাগে job manager আর task manager-এর একটা cluster, state backend, upgrade, বিকল হলে সামলানো, সব নিজের হাতে।

Managed Service for Apache Flink যেকোনো Apache Flink application তোমার হয়ে একটা managed cluster-এ চালায়। তুমি code (Java-র জন্য JAR, Python-এর জন্য zip) S3-এ upload করো, সেটার দিকে দেখিয়ে একটা application বানাও, একটা IAM role দাও, তারপর চালু করো। Compute জোগানো, application parallel-এ চালানো, scale করা, backup নেওয়া আর বিকল হলে আবার চালু করা, সব AWS-এর কাজ। তোমার দেখার মতো কোনো server নেই। Apache Flink-এর যেকোনো programming feature ব্যবহার করা যায়: নিচু স্তরের stream API, Table API আর Flink SQL, window, timer, state, join।

#কোথা থেকে পড়ে, কোথায় লেখে

AWS-এ একটা Flink application পড়ে:

• Kinesis Data Streams থেকে, Flink-এর Kinesis connector দিয়ে।
• Amazon MSK (বা যেকোনো Apache Kafka cluster) থেকে, Flink-এর Kafka connector দিয়ে।

আর লেখে অনেক রকম sink-এ: Kinesis Data Streams, Data Firehose, S3, OpenSearch, DynamoDB, MSK, আর Flink connector দিয়ে আরও অনেক কিছু।

#Window, সময় আর state

Real-time-এর বেশিরভাগ প্রশ্নই window নিয়ে: অফুরন্ত stream থেকে কাটা একটা সীমিত টুকরো।

সময় Flink মাপতে পারে machine-এর ঘড়ি দিয়ে (processing time), অথবা প্রতিটা event-এর ভেতরের timestamp দিয়ে (event time)। কোনো rider-এর ফোন সুড়ঙ্গে signal হারিয়ে দুই মিনিট পরে ping পাঠালেও event time ঠিক উত্তর দেয়; দেরিতে আসা event-এর জন্য window বন্ধ করার আগে কতক্ষণ অপেক্ষা করবে, তা Flink ঠিক করে watermark দিয়ে।

এই সবকিছুর জন্য লাগে state: খোলা window-এর ভেতরের হিসাব, প্রতিটা driver-এর শেষ অবস্থান, প্রতিটা wallet-এর আগের login। Flink state রাখে হিসাবের ঠিক পাশে (দ্রুত), আর checkpoint-এ তার backup রাখে (নিরাপদ)।

#Capacity, scaling আর খরচ

Capacity মাপা হয় Kinesis Processing Unit (KPU) দিয়ে। এক KPU মানে 1 vCPU আর 4 GB memory, সাথে চলমান application-এর জন্য 50 GB storage। তুমি ঠিক করো:

• Parallelism: প্রতিটা operator-এর কয়টা parallel কপি চলবে (default ১)।
• ParallelismPerKPU: তার কয়টা এক KPU-তে ধরবে (default ১, সর্বোচ্চ ৮; যে job বেশিরভাগ সময় I/O-র জন্য বসে থাকে, তার জন্য বাড়াও)।

বরাদ্দ KPU = Parallelism ÷ ParallelismPerKPU। Default-এ একটা application সর্বোচ্চ ৬৪টা KPU নিতে পারে (quota, বাড়ানো যায়)।

Automatic scaling default-এ চালু, আর চলে CPU দেখে:

• CPU টানা ১৫ মিনিট ৭৫% বা তার বেশি থাকলে service parallelism দ্বিগুণ করে, ফলে KPU-ও দ্বিগুণ।
• CPU টানা ছয় ঘণ্টা ১০%-এর নিচে থাকলে অর্ধেক করে, তবে তোমার ঠিক করা parallelism-এর নিচে কখনো না।
• Scale করার সময় application একটা snapshot থেকে আবার চালু হয়, তাই output অল্প সময়ের জন্য থেমে যায়। এই বিরতি মেনে নেওয়া না গেলে automatic scaling বন্ধ রাখো, আর parallelism নিজে ঠিক করো (হাতে, জানা ভিড়ের আগে schedule করে, বা নিজের metric-ভিত্তিক automation দিয়ে)।

দাম (us-east-1): প্রতি KPU-ঘণ্টা $0.11, সেকেন্ড ধরে bill, সাথে orchestration-এর জন্য প্রতি application-এ একটা বাড়তি KPU, চলমান storage প্রতি GB-মাসে $0.10, আর টেকসই backup-এর জন্য প্রতি GB-মাস। Parallelism ২ আর ParallelismPerKPU ১ দিয়ে একটা ছোট application চালালে লাগে ২ + ১ = ৩টা KPU: মোটামুটি ৩ × $0.11 × ৭৩০ ঘণ্টা ≈ মাসে $241, দিনে ২৪ ঘণ্টা, event আসুক বা না আসুক। প্রতি event ধরে দামের কোনো ব্যবস্থা নেই; stream processor সবসময় চালু থাকে।

#Checkpoint আর snapshot

Slide-এ এদের বলা হয়েছে “application backup”। এরা দুই রকম:

Checkpoint রাখে state, আর সাথে প্রতিটা source-এ কতদূর পড়া হয়েছে (প্রতিটা Kinesis shard-এর sequence number, প্রতিটা Kafka partition-এর offset)। Crash-এর পরে Flink শেষ checkpoint লোড করে, আর ওই অবস্থান থেকে source আবার পড়ে। ফলে প্রতিটা event state-এ ঠিক একবার গোনা হয়, যদিও কিছু event দুবার পড়া হয়েছে। এটা সম্ভব শুধু কারণ source আবার পড়ানো (replay) যায়: Flink কেন Kinesis Data Streams বা Kafka পড়ে, Firehose না, তার আরেকটা কারণ।

#এক নজরে Apache Kafka

#Kafka-র অংশগুলো

Apache Kafka হলো event streaming-এর সবচেয়ে বেশি ব্যবহৃত open-source platform, LinkedIn-এ তৈরি। বাংলাদেশে আর বাইরে অনেক company নিজেদের server-এ আগে থেকেই এটা চালায়। এর অংশগুলো:

• Broker: একটা Kafka server, যে ডেটা রাখে আর client-দের সেবা দেয়। কয়েকটা broker মিলে একটা cluster।
• Topic: record-এর একটা নামওয়ালা stream, যেমন orders বা payments।
• Partition: একটা topic ভাগ হয় কয়েকটা partition-এ, প্রতিটা একটা broker-এ রাখা ক্রমানুসারী, শুধু-শেষে-জোড়া log। Kinesis-এর কাছে shard যা, Kafka-র কাছে partition তা-ই: ক্রম আর parallelism-এর একক।
• Offset: partition-এর ভেতরে record-এর অবস্থান (০, ১, ২…)।
• Producer: তোমার code, যে topic-এ record লেখে। Record-এর একটা key থাকতে পারে; একই key-এর record একই partition-এ যায়, তাই ক্রম ঠিক থাকে।
• Consumer আর consumer group: তোমার code, যে record পড়ে (poll করে)। একই group ID-র consumer-রা partition নিজেদের মধ্যে ভাগ করে নেয়: প্রতিটা partition group-এর ঠিক একজন consumer পড়ে। আলাদা আলাদা group প্রত্যেকে সবকিছু পড়ে, নিজের নিজের offset নিয়ে।
• Replication: প্রতিটা partition-এর একটা leader থাকে এক broker-এ, আর follower থাকে অন্যগুলোতে (সাধারণত replication factor ৩)। Producer লেখে leader-এ; follower-রা কপি করে। Leader-এর broker বিকল হলে একজন in-sync follower দায়িত্ব নেয়।
• Retention: পড়লে কিছু মুছে যায় না। Record থাকে topic-এর retention সময় পর্যন্ত (Kafka-য় default ৭ দিন, চাইলে অনেক বেশি বা চিরকাল), তাই consumer আবার পড়তে পারে।

Slide যে প্রবাহ দেখায়: ডেটা আসে নানা জায়গা থেকে (Kinesis, IoT device, RDS…), producer (তোমার code) topic-এ লেখে, broker-রা সেটা replicate করে, consumer (তোমার code) topic থেকে poll করে, আর ফল পাঠায় EMR, S3, SageMaker, Kinesis, RDS ইত্যাদিতে।

#Partition-ই ঠিক করে কতটা parallel

একটা partition যেহেতু group-এর ঠিক একজন consumer-এর কাছে যায়, তাই partition-এর সংখ্যাই হলো এক group-এ একসাথে কাজ করতে পারা consumer-এর সর্বোচ্চ সংখ্যা। চার partition-এর topic-এ ছয়জন consumer দিলে দুজন বসে থাকে। কোনো consumer crash করলে বা নতুন কেউ যোগ দিলে group rebalance করে: partition নতুন করে ভাগ হয়, আর নতুন মালিক শুরু করে শেষ commit করা offset থেকে। যেসব record process হয়েছে কিন্তু commit হয়নি, সেগুলো আবার process হয়, তাই Kinesis consumer-এর মতোই Kafka consumer default-এ at-least-once, আর তাকে idempotent হতে হয়।

পরে topic-এ partition বাড়ানো যায়, কিন্তু কমানো কখনো যায় না। Partition বাড়ালে কোন key কোন partition-এ যাবে তা বদলে যায়, তাই কিছু সময় একটা key-এর event পুরোনো আর নতুন partition-এ ভাগ হয়ে ক্রম হারাতে পারে। সবচেয়ে ব্যস্ত দিনের কথা ভেবে partition-এর সংখ্যা শুরুতেই ঠিক করো।

#Amazon MSK: managed Kafka

#MSK তোমার হয়ে কী সামলায়

Amazon Managed Streaming for Apache Kafka (MSK) হলো AWS-এ পুরোপুরি managed Apache Kafka, আর slide একে দেখিয়েছে Amazon Kinesis-এর বিকল্প হিসেবে। এটা আসল, open-source Kafka চালায়, তাই তোমার আগের producer, consumer, tool আর দক্ষতা কোনো বদল ছাড়াই কাজ করে। MSK:

• একটা API call বা কয়েকটা click-এ cluster বানায়, বদলায়, মোছে।
• তোমার হয়ে Kafka broker node আর metadata node বানায় ও সামলায়: পুরোনো cluster-এ ZooKeeper node, নতুনগুলোতে KRaft controller।
• Cluster বসায় তোমার VPC-তে, high availability-র জন্য দুই বা তিনটা Availability Zone জুড়ে (তিনটাই পরামর্শ; Express broker-এ তিনটা লাগবেই)।
• Kafka-র সাধারণ বিকল হওয়া থেকে নিজে থেকেই সেরে ওঠে: বিকল broker ধরে ফেলে, বদলে দেয়, আর তার storage নতুনটার সাথে জুড়ে দেয়।
• ডেটা রাখে EBS volume-এ, প্রতি broker-এ সর্বোচ্চ 16 TiB, যতদিন চাও ততদিন: retention তোমার পছন্দ, service-এর কোনো সীমা না। Storage নিজে থেকে বাড়তে পারে, আর tiered storage পুরোনো ডেটা সস্তা স্তরে সরিয়ে দেয়।
• At rest ডেটা সবসময় KMS দিয়ে encrypt করে, আর পথে default-এ TLS দিয়ে।
• তুমি চাইলে Kafka version patch আর upgrade করে, আর metric পাঠায় CloudWatch-এ (বা Prometheus-এ)।

Provisioned cluster-এ যা তোমাকেই ঠিক করতে হয়: broker-এর instance type আর সংখ্যা, storage-এর মাপ, Kafka version, topic-এর settings (partition, replication factor, retention), আর client authentication।

#MSK চালানোর তিন উপায়

Slide যেটা বলে, MSK Serverless তা-ই: capacity না সামলিয়ে MSK-এ Kafka চালানো; compute আর storage MSK নিজে জোগায় আর বাড়ায়-কমায়। এটা Kinesis on-demand mode-এর Kafka সংস্করণ।

#MSK-এর নিরাপত্তা

• Network: broker-এর network interface থাকে তোমার private subnet-এ, security group দিয়ে সুরক্ষিত। Provisioned cluster-এ public access চালু করা যায়, তবে বেশিরভাগ cluster private-ই থাকে।
• Authentication: IAM access control (কে কোন topic-এ লিখবে বা পড়বে, IAM policy ঠিক করে), SASL/SCRAM (username আর password থাকে Secrets Manager-এ), mutual TLS (AWS Private CA থেকে client certificate), কিংবা তুমি অনুমতি দিলে authentication ছাড়াই। MSK Serverless শুধু IAM সমর্থন করে।
• Authorization: IAM auth-এ IAM policy; SASL/SCRAM বা mTLS-এ Kafka ACL।
• পথে encryption: client আর broker-এর মধ্যে default-এ TLS; PLAINTEXT, বা দুটোই, শুধু তুমি বেছে নিলে। Broker-দের নিজেদের মধ্যে default-এ TLS।
• At rest encryption: সবসময় চালু, AWS managed KMS key বা তোমার নিজের customer managed key দিয়ে।

#MSK Connect আর MSK Replicator

• MSK Connect তোমার হয়ে Kafka Connect worker চালায়। Consumer code না লিখে Kafka-য় ডেটা ঢোকানো-বের করার প্রচলিত উপায় হলো Kafka Connect: source connector ডেটা ভেতরে আনে (যেমন Debezium একটা database-এর change log পড়ে, এটাই change data capture), sink connector বাইরে নেয় (S3, OpenSearch, কোনো database-এ)। Worker-দের সংখ্যা MSK Connect নিজে বাড়ায়-কমায়।
• MSK Replicator এক MSK cluster থেকে আরেকটায় topic অবিরাম কপি করে, একই Region-এ বা অন্য Region-এ, disaster recovery-র জন্য বা ব্যবহারকারীর কাছাকাছি ডেটা রাখতে।

#MSK থেকে কারা পড়ে

Slide চার রকম consumer দেখায়; Firehose আর MSK Connect মিলিয়ে ছবিটা পুরো হয়:

Lambda consumer প্রতি partition-এ ক্রম রাখে: কোনো batch বারবার ব্যর্থ হলে সেই partition তার পেছনে অপেক্ষা করে। Event source mapping-এ একটা retry সীমা আর একটা on-failure destination (SQS, SNS বা S3) ঠিক করে দাও, যাতে খারাপ record সরিয়ে রেখে partition এগিয়ে যেতে পারে। Event source mapping বিস্তারিত আছে লেসন ১৫.১-এ।

#Kinesis Data Streams নাকি Amazon MSK

#পাশাপাশি তুলনা

#পরীক্ষার কৌশল

নোট যে নিয়ম দিয়েছে, সেটাই খাটে:

• AWS-এর নিজস্ব, serverless সমাধান, সামলানোর কাজ সবচেয়ে কম → Kinesis Data Streams + Managed Service for Apache Flink।
• Open-source, বা আগে থেকে চলা Kafka-ভিত্তিক ecosystem (“company on-premises-এ Apache Kafka চালায়, সবচেয়ে কম code বদলে migrate করতে চায়”, “Kafka Connect ব্যবহার করে”, “Kafka API লাগবে”) → Amazon MSK।

আরও কিছু সূত্র: ১ MB-এর চেয়ে বড় message (MSK-এর পুরোনো একটা কারণ, যদিও Kinesis এখন প্রতি stream-এ 10 MiB পর্যন্ত নিতে পারে), এক বছরের বেশি retention (MSK), “আমাদের VPC-র ভেতরেই চলতে হবে” (MSK, বা VPC endpoint-সহ Kinesis), “capacity-র হিসাব চাই না, traffic ওঠানামা করে” (Kinesis on-demand বা MSK Serverless; Kafka-র কথা না থাকলে AWS-এর নিজস্বটাই জেতে)।

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

• Flink-কে Firehose-এর দিকে দেখানো: অসম্ভব; Firehose-এর কোনো consumer API নেই। Source হিসেবে Kinesis Data Streams বা MSK নাও।
• Flink application পিছিয়ে পড়ছে: source-এর lag বাড়তেই থাকে; parallelism কম, বা কোনো একটা operator আটকে দিচ্ছে। Source-এর lag metric (Kinesis-এ millisBehindLatest, Kafka-য় consumer lag) আর CPU দেখো, আর parallelism বাড়াও বা automatic scaling-কে কাজ করতে দাও।
• ঈদের ভিড়ে output থেমে যাওয়া: automatic scaling application আবার চালু করেছে। জানা ভিড়ের আগেই scale করো, বা automatic scaling বন্ধ রেখে নিজে মাপ ঠিক করো।
• Flink job আর ফিরতে পারছে না: checkpoint বন্ধ ছিল, বা source-এর retention বিভ্রাটের চেয়ে ছোট, তাই checkpoint-এ লেখা অবস্থান আর নেই।
• বসে থাকা Kafka consumer: group-এ partition-এর চেয়ে বেশি consumer। Partition বাড়াও (key-এর ক্রমের কথা মাথায় রেখে), বা বাড়তিদের বদলি হিসেবে মেনে নাও।
• Rebalance-এর পরে duplicate: একজন consumer process করার পরে, commit-এর আগে crash করেছে। Processing idempotent করো; commit করো process-এর পরে, আগে না।
• Partition বাড়ানোর পরে ক্রম ভাঙা: key-গুলো নতুন partition-এ সরে গেছে। Partition-এর সংখ্যা শুরুতেই ঠিক করো।
• Broker-এর disk ভরে যাওয়া: EBS-এর মাপের তুলনায় retention বেশি লম্বা। Disk ব্যবহারের ওপর alarm বসাও, storage বাড়াও, tiered storage বা Express broker নাও, বা retention কমাও।
• এক AZ-এর cluster, বা replication factor ১: AZ বিকল হলে partition পাওয়া যায় না বা ডেটা হারায়। দরকারি ডেটার জন্য তিনটা AZ, replication factor ৩, min.insync.replicas ২ আর acks=all রাখো।
• Client connect করতে পারছে না: ভুল port বা ভুল authentication পদ্ধতি (IAM, SCRAM, TLS প্রত্যেকের আলাদা bootstrap string), security group বন্ধ, বা IAM policy-তে topic-এর জন্য kafka-cluster অনুমতি নেই।
• একটা খারাপ record-এ আটকে থাকা Lambda consumer: retry সীমা আর on-failure destination ঠিক করো।

#System design-এ Flink আর MSK

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

• Real-time analytics: producer → Kinesis Data Streams বা MSK → Flink (window, join) → OpenSearch dashboard, DynamoDB, SNS দিয়ে alert; সাথে Firehose বা Glue Streaming → data lake-এর জন্য S3। IoT device থেকে dashboard পর্যন্ত পুরো serverless pipeline নিয়ে লেসন ১৭.৮।
• Fraud ধরা: payment → stream → wallet ধরে key করা Flink, প্রতিটা payment-কে state-এ রাখা wallet-এর সাম্প্রতিক ইতিহাসের সাথে মেলায় → এক সেকেন্ডের মধ্যে আটকায় বা চিহ্ন দেয়।
• Change data capture: database → MSK Connect (Debezium) → MSK → consumer, যারা search index, cache আর data warehouse হালনাগাদ রাখে।
• On-premises Kafka migrate করা: topic MSK-এ replicate করো, আগে consumer সরাও, তারপর producer; পরে অন্য Region-এ কপির জন্য MSK Replicator।
• Event দিয়ে microservice decouple করা: service-গুলো নিজেদের domain event MSK topic-এ publish করে; প্রতিটা service নিজের consumer group-এ যা দরকার পড়ে।

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

এখানে AWS CLI v2 আর Kafka-র open-source command-line tool ব্যবহার করা হয়েছে। MSK আর Flink application যতক্ষণ থাকে ততক্ষণ ঘণ্টা ধরে bill হয়, তাই শেষে মুছে ফেলো। Subnet, security group আর account-এর মানগুলো উদাহরণ।

# 1. একটা MSK Serverless cluster বানাও (শুধু IAM authentication)
aws kafka create-cluster-v2 --cluster-name rides-example \
  --serverless '{"VpcConfigs":[{"SubnetIds":["subnet-0a1b2c3d","subnet-0e5f6a7b","subnet-0c8d9e0f"],
                 "SecurityGroupIds":["sg-0123456789abcdef0"]}],
                 "ClientAuthentication":{"Sasl":{"Iam":{"Enabled":true}}}}'
# { "ClusterArn": "arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1",
#   "ClusterName": "rides-example", "State": "CREATING", "ClusterType": "SERVERLESS" }

# 2. Cluster ACTIVE হলে bootstrap server জেনে নাও
aws kafka get-bootstrap-brokers \
  --cluster-arn arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1
# { "BootstrapBrokerStringSaslIam":
#     "boot-abc12345.c1.kafka-serverless.ap-southeast-1.amazonaws.com:9098" }

# 3. একই VPC-র একটা EC2 instance-এ: IAM auth-এর client settings (aws-msk-iam-auth jar লাগবে)
cat > client.properties <<'EOF'
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
EOF
BS=boot-abc12345.c1.kafka-serverless.ap-southeast-1.amazonaws.com:9098

# 4. ৬টা partition-এর একটা topic বানাও (replication Serverless নিজেই ঠিক করে)
kafka-topics.sh --bootstrap-server $BS --command-config client.properties \
  --create --topic ride-pings --partitions 6
# Created topic ride-pings.

# 5. Key দিয়ে record লেখো, তারপর একটা consumer group-এ পড়ো
kafka-console-producer.sh --bootstrap-server $BS --producer.config client.properties \
  --topic ride-pings --property parse.key=true --property key.separator=:
# > driver-17:{"area":"Mirpur","lat":23.80,"lon":90.36}
kafka-console-consumer.sh --bootstrap-server $BS --consumer.config client.properties \
  --topic ride-pings --group surge --from-beginning

# 6. Group-এর প্রতিটা consumer-এর partition, offset আর lag দেখো
kafka-consumer-groups.sh --bootstrap-server $BS --command-config client.properties \
  --describe --group surge
# GROUP  TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# surge  ride-pings  0          1204            1204            0    console-consumer-5f1...
# surge  ride-pings  1          977             981             4    console-consumer-5f1...

# 7. Partition বাড়াও (করা যায়) - কমানো সম্ভব না
kafka-topics.sh --bootstrap-server $BS --command-config client.properties \
  --alter --topic ride-pings --partitions 12

# 8. MSK consumer হিসেবে একটা Lambda function
aws lambda create-event-source-mapping --function-name update-ride-status \
  --event-source-arn arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1 \
  --topics ride-pings --starting-position LATEST --batch-size 100

Provisioned cluster-এ এর বদলে একটা --provisioned অংশ দিতে হয়, যেখানে থাকে broker type, দুই বা তিনটা AZ-এর client subnet আর EBS-এর মাপ:

aws kafka create-cluster-v2 --cluster-name payments-example --provisioned '{
  "KafkaVersion": "3.7.x.kraft", "NumberOfBrokerNodes": 3,
  "BrokerNodeGroupInfo": { "InstanceType": "kafka.m7g.large",
    "ClientSubnets": ["subnet-0a1b2c3d","subnet-0e5f6a7b","subnet-0c8d9e0f"],
    "SecurityGroups": ["sg-0123456789abcdef0"],
    "StorageInfo": { "EbsStorageInfo": { "VolumeSize": 1000 } } },
  "EncryptionInfo": { "EncryptionInTransit": { "ClientBroker": "TLS", "InCluster": true } },
  "ClientAuthentication": { "Sasl": { "Iam": { "Enabled": true } } } }'

এবার একটা Flink application। Job-টা একটা JAR-এ build করে S3-এ upload করো, তারপর:

# 9. Application বানাও: S3-এ code, একটা IAM role, parallelism আর checkpoint
aws kinesisanalyticsv2 create-application --application-name surge-example \
  --runtime-environment FLINK-1_20 \
  --service-execution-role arn:aws:iam::111122223333:role/flink-surge-role \
  --application-configuration '{
    "ApplicationCodeConfiguration": { "CodeContentType": "ZIPFILE",
      "CodeContent": { "S3ContentLocation": {
        "BucketARN": "arn:aws:s3:::flink-code-example", "FileKey": "surge-1.0.jar" } } },
    "FlinkApplicationConfiguration": {
      "ParallelismConfiguration": { "ConfigurationType": "CUSTOM", "Parallelism": 2,
        "ParallelismPerKPU": 1, "AutoScalingEnabled": true },
      "CheckpointConfiguration": { "ConfigurationType": "DEFAULT" } },
    "ApplicationSnapshotConfiguration": { "SnapshotsEnabled": true } }'
# { "ApplicationDetail": { "ApplicationName": "surge-example", "ApplicationStatus": "READY",
#     "RuntimeEnvironment": "FLINK-1_20", "ApplicationVersionId": 1, ... } }

# 10. চালু করো, snapshot থাকলে সর্বশেষটা থেকে
aws kinesisanalyticsv2 start-application --application-name surge-example \
  --run-configuration '{"ApplicationRestoreConfiguration":{"ApplicationRestoreType":"RESTORE_FROM_LATEST_SNAPSHOT"}}'

# 11. ঝুঁকির বদলের আগে snapshot নাও, আর snapshot-এর তালিকা দেখো
aws kinesisanalyticsv2 create-application-snapshot --application-name surge-example \
  --snapshot-name before-eid-release
aws kinesisanalyticsv2 list-application-snapshots --application-name surge-example

# 12. গুছিয়ে ফেলো
aws kinesisanalyticsv2 stop-application --application-name surge-example
aws kafka delete-cluster \
  --cluster-arn arn:aws:kafka:ap-southeast-1:111122223333:cluster/rides-example/1a2b3c4d-...-s1

Console-এ Flink application-এর পাতা থেকে Apache Flink dashboard খোলে, সেখানে প্রতিটা operator, তার parallelism আর back pressure দেখা যায়। MSK-এর দিকে Monitoring tab-এ প্রতি broker-এ ঢোকা-বেরোনো byte আর provisioned cluster-এ disk ব্যবহার দেখা যায়।

#সারসংক্ষেপ

মূল কথা

• Stream processor live stream-এর ওপর filter করে, window ধরে হিসাব করে, join করে আর state রাখে, ফল দেয় কয়েক সেকেন্ডে। AWS-এ সেটা Amazon Managed Service for Apache Flink, আগের নাম Kinesis Data Analytics for Apache Flink। শুধু-SQL Kinesis Data Analytics বন্ধ (১৫ অক্টোবর ২০২৫ থেকে নতুন app নেই, ২৭ জানুয়ারি ২০২৬ থেকে মুছে ফেলা)।
• Flink পড়ে Kinesis Data Streams আর Amazon MSK (Kafka) থেকে। Data Firehose থেকে পড়ে না; Firehose sink হতে পারে। Code লেখা যায় Java, Python, SQL-এ (Flink 2.x-এ Scala deprecated); interactive SQL-এর জন্য Studio notebook।
• Capacity KPU-তে (1 vCPU, 4 GB)। Automatic scaling ১৫ মিনিট ৭৫%-এর ওপরে CPU থাকলে parallelism দ্বিগুণ করে, ৬ ঘণ্টা ১০%-এর নিচে থাকলে অর্ধেক, অল্প সময়ের restart-সহ। Checkpoint (স্বয়ংক্রিয়, প্রতি ৬০ সেকেন্ডে) crash থেকে ফেরায়, state থাকে ঠিক একবার গোনা; snapshot (হাতে বা update-এর সময়) বেছে নেওয়া state-এ ফেরায়।
• Kafka: broker রাখে topic, topic ভাগ হয় partition-এ (offset-ওয়ালা ক্রমানুসারী log); producer key ধরে লেখে; consumer group প্রতিটা partition একজন consumer-কে দেয়; leader আর follower নিয়ে replication factor ৩; retention আছে, তাই পড়লে কিছু মোছে না, আবার পড়া যায়।
• Amazon MSK = managed Kafka, Kinesis-এর বিকল্প: broker আর ZooKeeper / KRaft node বানায় ও সামলায়, চলে তোমার VPC-তে ২ বা ৩টা AZ জুড়ে, বিকল হলে নিজেই সারে, ডেটা রাখে EBS-এ যতদিন চাও। উপায়: Standard broker, Express broker (বেশি throughput, storage সামলানোর কিছু নেই), MSK Serverless (capacity সামলানো নেই, শুধু IAM)।
• MSK-এর consumer: Flink, Glue Streaming ETL, Lambda, EC2 / ECS / EKS-এর app, Firehose, MSK Connect। MSK Connect চালায় Kafka Connect; MSK Replicator cluster থেকে cluster-এ কপি করে।
• Kinesis বনাম MSK: 1 MB বনাম default 1 MB (বাড়ানো যায়); shard (split/merge) বনাম partition (শুধু বাড়ানো); সবসময় TLS বনাম TLS বা plaintext; দুটোতেই at rest KMS।
• নোটের পরীক্ষার নিয়ম: AWS-এর নিজস্ব → Kinesis Data Streams + Managed Flink; open-source বা আগের Kafka → Amazon MSK।

নিজেকে যাচাই করো: Managed Flink আর Amazon MSK৩২টি প্রশ্ন