Amazon EMR

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

• Amazon EMR (Elastic MapReduce) কী, big data-র জন্য আদৌ cluster কেন লাগে, আর এটা কোন কোন open-source framework চালায় (Spark, Hadoop, Hive, HBase, Presto/Trino, Flink আর আরও কিছু), তা বোঝাতে।
• তিন রকম node, primary, core আর task, কে কী চালায়, আর প্রতিটা হারালে cluster আর তার ডেটার কী হয়, তা বলতে।
• প্রতিটা node type-এর জন্য purchasing option (On-Demand, Reserved, Savings Plans, Spot) বাছতে, long-running আর transient দুই রকম cluster-এর জন্যই, আর Spot task node নিয়ে পরীক্ষার চিরচেনা প্রশ্নের উত্তর দিতে।
• HDFS কেন ক্ষণস্থায়ী, আর input-output কেন EMRFS দিয়ে S3-এ রাখা উচিত, তা বোঝাতে, আর replication factor ধরে HDFS-এর মাপ হিসাব করতে।
• Uniform instance group আর instance fleet, এবং managed scaling আর custom automatic scaling-এর তুলনা করতে।
• EMR on EC2, EMR on EKS আর EMR Serverless-এর মধ্যে বাছতে, আর একটা design-এ Athena, Redshift আর Glue-এর পাশে EMR-কে ঠিক জায়গায় বসাতে।

#Big data-র জন্য cluster কেন লাগে

অধ্যায় ১৬-তে আমরা database বেছেছিলাম সে কোন প্রশ্নের উত্তর দেয় তা দেখে। কিছু প্রশ্ন এত বড় যে কোনো একটা machine-এর পক্ষে সামলানো সম্ভব না। বাংলাদেশের একটা mobile operator প্রতিটা call, SMS আর data session-এর জন্য একটা call-detail record (CDR) লেখে: এক রাতেই শত কোটি row, কয়েক terabyte। Finance সকাল ৬টার মধ্যে চায় প্রতিটা tower আর প্রতিটা package-এ কত মিনিট কথা হলো; network team চায় কোন এলাকায় কত call কেটে গেল; fraud team খোঁজে সিম বক্সের মতো অবৈধ call-এর ধরন। একটা server যত বড়ই হোক, ওই কয়েক terabyte পড়তেই তার অনেক ঘণ্টা লেগে যাবে।

২০০০-এর দশকে Google-এর গবেষণাপত্রগুলো থেকে উত্তরটা এসেছে: কাজটা অনেকগুলো সাধারণ machine-এ ভাগ করে দাও। এর programming model-এর নাম MapReduce: input-কে টুকরো করো, প্রতিটা টুকরো আলাদা machine-এ একসাথে process করো (map), তারপর আংশিক ফলগুলো জোড়া দাও (reduce)। Apache Hadoop হলো এর open-source রূপ, নিজস্ব distributed file system (HDFS) আর resource manager (YARN) সহ। পরে Apache Spark একই কাজ বেশিরভাগটা memory-তে করে অনেক দ্রুত করতে শুরু করল, আর আজ বেশিরভাগ team সেটাতেই লেখে।

এমন cluster নিজে চালানো ঝামেলার কাজ: কয়েক ডজন server কেনো বা ভাড়া নাও, Hadoop, Spark আর Hive-এর মিলে যাওয়া version install করো, একে অপরের সাথে কথা বলার মতো configure করো, নষ্ট disk বদলাও, patch দাও, আর কিছু না চললেও টাকা দিতে থাকো।

#Amazon EMR আসলে কী

Amazon EMR (আগের নাম Elastic MapReduce) হলো AWS-এর managed big-data platform। তুমি বলো কোন framework আর কতটা capacity চাও; EMR EC2 instance চালু করে, framework install আর configure করে, সবগুলোকে জুড়ে দেয়, node-এর স্বাস্থ্যের দিকে নজর রাখে, আর কাজ শেষে সব গুটিয়ে ফেলে। একটা cluster হতে পারে হাতে-গোনা কয়েকটা instance, আবার শত শত, এমনকি হাজার হাজার EC2 instance-ও।

EMR-এর সাথে প্রধান open-source big-data framework-গুলো একসাথে বাঁধা থাকে, প্রতিটা EMR release-এ (যেমন emr-7.x) AWS-এর একসাথে পরীক্ষা করা version-এ:

EMR সাথে আনে auto scaling (managed scaling), Spot Instance-এর সাথে গভীর মিল, আর Spark ও Hive-এর জন্য নিজের tuned runtime। Job তুমিই লেখো; platform চালায় EMR।

ব্যবহারের ক্ষেত্র (slide-এর তালিকা, প্রতিটা দেখতে কেমন):

• Data processing আর ETL: কয়েক terabyte log, clickstream বা CDR পরিষ্কার করে Athena বা Redshift-এর জন্য Parquet বানানো।
• Machine learning: এক machine-এ না আঁটা ডেটার ওপর Spark-এর ML library দিয়ে feature engineering আর training (managed ML platform অধ্যায় ১৮-তে)।
• Web indexing: শত কোটি page crawl করে search index বানানো, যে কাজের জন্য Hadoop-এর জন্মই হয়েছিল।
• Big data analytics: বড় join আর aggregation, risk model, genomics, Trino দিয়ে interactive SQL।

পরীক্ষার keyword সোজা: “Hadoop”, “Spark”, “Hive”, “HBase”, “big data framework”, “on-premises Hadoop cluster migrate করো” → EMR।

#Cluster-এর ভেতরে: primary, core আর task node

EC2-এর ওপর চলা প্রতিটা EMR cluster-এ সর্বোচ্চ তিন রকম node থাকে, প্রতিটা একটা EC2 instance। নোটে এটাকে পরীক্ষার জন্য অত্যন্ত গুরুত্বপূর্ণ বলা আছে।

#Primary node (আগের নাম master node)

Primary node cluster চালায়। Framework-গুলোর সমন্বয়ের অংশগুলো এখানে চলে: YARN ResourceManager, যে ঠিক করে কোন job কতটা CPU আর memory পাবে, আর HDFS NameNode, যে জানে কোন file-এর কোন block কোন machine-এ আছে। জমা দেওয়া job-এর অবস্থা আর বাকি node-গুলোর স্বাস্থ্যও সে দেখে। Log পড়তে আর framework-গুলোর web interface খুলতে এতে SSH করা যায়।

AWS “master node”-এর নাম বদলে primary node রেখেছে; পুরোনো প্রশ্ন আর নোটে এখনো master লেখা। একটা cluster-এ primary node থাকে একটা, অথবা high availability-র জন্য তিনটা (EMR 5.23 থেকে): active primary নষ্ট হলে EMR একটা standby-তে চলে যায় আর নষ্টটার জায়গায় নতুন একটা বসায়। Primary-র খুব বেশি শক্তি লাগে না; মোটামুটি পঞ্চাশটা node পর্যন্ত cluster-এ m5.xlarge-ই যথেষ্ট।

নোটে আছে primary job “broadcast করে”। আরেকটু নিখুঁতভাবে বললে, সে সমন্বয় করে: কাজ আর ডেটা কোথায় যাবে ঠিক করে, আর আসল processing হয় core আর task node-এ।

#Core node

Core node দুটো কাজ করে: task চালায় (YARN NodeManager, Spark executor, MapReduce task) আর ডেটা রাখে HDFS-এ, DataNode service দিয়ে, নিজের instance store আর EBS volume-এ। দ্বিতীয় কাজের কারণেই এরা পুরো cluster-এর জীবনকাল ধরে long-running: একটা সরাতে হলে আগে HDFS-কে তার block অন্য জায়গায় সরাতে হয়, আর হঠাৎ একটা হারালে ডেটাও হারাতে পারে। একটা cluster-এ core instance group (বা fleet) থাকে একটাই, তবে তার ভেতরে instance থাকতে পারে অনেক।

#Task node (ঐচ্ছিক)

Task node শুধু task চালায়। এদের কাছে কোনো HDFS ডেটা থাকে না, শুধু temporary file। এদের কাজ দরকারের সময় হিসাবের শক্তি বাড়ানো: ভারী রাতের জন্য বিশটা task node যোগ করো, সকালে সরিয়ে দাও। Uniform instance group-এ সর্বোচ্চ ৪৮টা task instance group রাখা যায়, প্রত্যেকটার নিজের instance type আর purchasing option। একটা হারালে যায় শুধু সময়, তাই task node সাধারণত Spot।

নিচের scenario-গুলো চালাও: একটা স্বাভাবিক job, তারপর একটা Spot task node হারানো, একটা core node হারানো, একমাত্র primary হারানো, আর তিনটা primary-র একটা হারানো।

Spot task node উধাও হলে চলতি কাজের টুকরোগুলো শুধু আবার schedule হয়। যেটা কিছুতেই হারানো চলবে না সেটা হলো application master: প্রতিটা job-এর নিজস্ব সমন্বয়কারী, যাকে YARN cluster-এর কোথাও চালু করে। EMR 5.x default-এ application master core node-এ রাখত; EMR 6.x থেকে YARN-এর এই node-label setting default-এ বন্ধ, আর EMR 7.x-এ market type ধরে node-এ label দেওয়া যায়, যাতে application master থাকে On-Demand (বা core) node-এ। অনেক Spot task node চালালে এটা চালু করে নাও।

#HDFS replication আর capacity

HDFS প্রতিটা block-এর কয়েকটা কপি আলাদা আলাদা core node-এ রাখে, যাতে একটা disk নষ্ট হলে ডেটা না হারায়। Core node-এর সংখ্যা দেখে EMR default replication factor ঠিক করে:

ব্যবহারযোগ্য HDFS capacity = (প্রতি core node-এর storage × core node-এর সংখ্যা) ÷ replication factor। দশটা core node, প্রত্যেকটায় ৮০০ GB local storage হলে ডেটা ধরে প্রায় ৮,০০০ ÷ ৩ ≈ ২,৬৬৬ GB। মনে রেখো, job চলার সময় input, মাঝের ডেটা আর output, তিনটাকেই একসাথে আঁটতে হয়। HDFS ছোট পড়লে: আরও বা আরও বড় core node, আরও EBS volume, compression, কিংবা (সবচেয়ে ভালো) ডেটা S3-এ রাখা।

HDFS rack চেনে, এজন্যই লেসন ৪.৫-এ Hadoop-কে partition placement group-এর সাথে জোড়া হয়েছে।

#প্রতিটা node type-এর purchasing option

EMR-এর node সাধারণ EC2 instance, তাই লেসন ৪.৪-এর EC2 purchasing option এখানেও খাটে। EC2-এর দামের ওপর প্রতি instance-ঘণ্টায় একটা EMR fee দিতে হয় (us-east-1-এ একটা m5.xlarge-এর জন্য EC2 On-Demand $0.192, সাথে EMR $0.048), bill হয় প্রতি সেকেন্ড ধরে, সর্বনিম্ন এক মিনিট। Instance On-Demand হোক বা Spot, EMR fee একই।

• On-Demand: নির্ভরযোগ্য আর আগে থেকে জানা দাম; AWS instance ফেরত নেয় না। যা কিছু হারানো চলবে না, তার default।
• Reserved Instance (সর্বনিম্ন ১ বছর) আর Savings Plans: সারা বছর চলা capacity-তে বড় ছাড়। Reserved Instance হলো bill-এর ছাড়, তাই মিলে যাওয়া instance চালু করলে EMR নিজে থেকেই সেটা ব্যবহার করে; EMR-এ আলাদা কিছু সেট করতে হয় না। নির্দিষ্ট AZ-এ জায়গা নিশ্চিত করতে EMR instance fleet On-Demand Capacity Reservation-ও ব্যবহার করতে পারে।
• Spot Instance: প্রায় ৯০% পর্যন্ত সস্তা, কিন্তু EC2 দুই মিনিটের notice দিয়ে ফেরত নিতে পারে। সস্তা আর কম নির্ভরযোগ্য: আবার করা যায় এমন কাজের জন্য একদম ঠিক।

চলতি cluster-এ কোনো group-এর purchasing option বদলানো যায় না: primary আর core-এর জন্য নতুন cluster চালু করতে হয়; task node-এর জন্য নতুন task group বা fleet যোগ করে পুরোনোটা সরাতে হয়।

এবার নিজে cluster বানাও। প্রতিটা node group-এর purchasing option আর ডেটা কোথায় থাকবে বেছে নাও, দুই ঘণ্টা পরে Spot ফেরত নেওয়া হবে কি না ঠিক করো, তারপর মিলিয়ে দেখো কী টিকে থাকে আর রাতের খরচ কত।

#Long-running আর transient cluster

Slide শেষ হয় এমন একটা সিদ্ধান্ত দিয়ে, যেটা বাকি সবকিছু বদলে দেয়:

• Transient (অস্থায়ী) cluster বানানো হয় এক দফা কাজের জন্য: নিজের step-গুলো চালায়, আর শেষ step শেষ হলে নিজেই terminate হয়। টাকা দাও শুধু কাজের ঘণ্টাগুলোর। ডেটা থাকতে হবে S3-এ, কারণ cluster-এর disk উধাও হয়ে যায়। রাতের ETL, মাসিক report আর এককালীন backfill-এর জন্য এটাই ঠিক model।
• Long-running (persistent) cluster অনেক ব্যবহারকারী আর job-এর জন্য চালু থাকে: EMR Studio-তে interactive notebook, হঠাৎ দরকারের Hive আর Trino query, সারাদিন app যে HBase table পড়ে। Termination protection চালু করো, স্থির অংশটা Reserved Instance বা Savings Plans দিয়ে ঢাকো, আর চাপের সাথে scaling-কে চলতে দাও।

Transient cluster চালু হতে কয়েক মিনিট লাগে (instance, bootstrap action, framework install), তাই এটা মানায় এক ঘণ্টা বা তার বেশি সময়ের job-এ। কয়েক সেকেন্ডে শুরু হতে হবে এমন কিছুর জন্য long-running cluster, অথবা pre-initialized capacity সহ EMR Serverless (নিচে) নাও।

Step হলো cluster-এ জমা দেওয়া এক টুকরো কাজ: একটা Spark application, একটা Hive script, একটা custom JAR। প্রতিটা step-এর একটা fail করলে কী হবে setting থাকে: চালিয়ে যাও, বাকি step বাতিল করে অপেক্ষা করো, নাকি cluster terminate করো। Cluster-এর log একটা S3 log URI-তে পাঠানো যায়, তাই transient cluster চলে যাওয়ার পরেও log পড়া যায়। Auto-termination policy তোমার ঠিক করা অলস সময়ের পর cluster বন্ধ করে দেয়: ঈদের ছুটিতে কেউ ভুলে যাওয়া test cluster-এর ওষুধ।

#Storage: HDFS নাকি EMRFS দিয়ে S3

একটা EMR cluster দুটো একদম আলাদা জায়গায় ডেটা রাখতে পারে।

• HDFS থাকে core node-গুলোর instance store আর EBS volume-এ। এটা দ্রুত, local, আর core node-গুলোর মধ্যে replicate করা, কিন্তু ক্ষণস্থায়ী: cluster terminate হলে মুছে যায় (লেসন ৫.৩-এর instance store-এর মতো)। এর মাপও তোমার core node দিয়ে বাঁধা।
• EMRFS (EMR File System) Hadoop, Spark আর Hive-কে s3:// path দিয়ে S3-কে file system-এর মতো ব্যবহার করতে দেয়। ওখানে লেখা ডেটা স্থায়ী: cluster-এর পরেও থাকে, পরের cluster, Athena বা Redshift Spectrum পড়তে পারে, আর S3-এর এগারোটা নয়-এর durability নিয়ে থাকে। ২০২০-র ডিসেম্বর থেকে S3 strongly consistent, তাই পুরোনো “EMRFS consistent view” কৌশল (S3 listing-এর হিসাব রাখা একটা DynamoDB table) আর লাগে না; ২০২৩-এ এর standard support শেষ হয়েছে। EMR 7.10 থেকে default S3 connector হলো open-source S3A; ধারণাটা একই, S3-ই cluster-এর টেকসই storage।

আধুনিক pattern: input আর output S3-এ, HDFS (বা local disk) শুধু মাঝের ডেটার জন্য, যেমন Spark-এর shuffle file। এতে পাওয়া যায় storage আর compute-এর আলাদা হওয়া: cluster হয়ে যায় ফেলে দেওয়ার জিনিস, যেকোনো মাপ বা EMR release একই ডেটা পড়তে পারে, transient cluster অলস থাকলে কোনো খরচ নেই, আর AZ বিপর্যয় বা ফেরত নেওয়া core node-এ হারায় শুধু খসড়া ডেটা। HBase-ও নিজের ডেটা S3-এ রাখতে পারে (HBase on S3), তাই database cluster-ও বদলে ফেলা যায়।

#Instance group আর instance fleet

Cluster বানানোর সময় পুরো cluster-এর instance বর্ণনার দুটো উপায়ের একটা বাছতে হয়:

Spot-নির্ভর cluster-এ fleet-ই ভালো default: কয়েকটা instance type আর pool-এ ছড়িয়ে দেওয়াই Spot capacity পাওয়ার আর কম interruption-এর সবচেয়ে কার্যকর উপায় (লেসন ৪.৪-এ Spot Fleet-এর বেলাতেও একই পরামর্শ)।

#Scaling: managed scaling আর custom automatic scaling

• EMR managed scaling (EMR 5.30 আর তার পরে) cluster-এর কাজের চাপ দেখে নিজেই core আর task capacity যোগ করে বা সরায়। তুমি দাও একটা সর্বনিম্ন আর একটা সর্বোচ্চ, চাইলে একটা On-Demand সীমা (এর ওপরের capacity Spot) আর একটা সর্বোচ্চ core সীমা (এর ওপরের capacity task node)। Primary কখনো scale হয় না। Instance group আর fleet দুটোতেই চলে, সাবধানে ছোট হয় (চলতি Spark shuffle ডেটা বা application master আছে এমন node এড়িয়ে যায়), কিন্তু বোঝে শুধু YARN application, যেমন Spark, Hive, Hadoop আর Flink; Presto/Trino বা HBase না।
• Custom automatic scaling policy হলো CloudWatch metric-এর ওপর rule (যেমন YARN-এর খালি memory ১৫%-এর নিচে → দুটো task node যোগ করো)। এগুলো চলে শুধু instance group-এ, আর YARN-এর বাইরের workload-এর জন্য এটাই উপায়।

বাড়া-কমা করো task node দিয়ে, core group স্থির রাখো: core node কমানো মানে HDFS-কে আগে সেটা decommission করে তার block অন্যত্র copy করতে হয়।

#Deployment option: EMR on EC2, EMR on EKS, EMR Serverless

ওপরের সবকিছু EMR on EC2, মূল আর এখনো সবচেয়ে নমনীয় model। নতুন দুটো option একই EMR runtime চালায়, কিন্তু সামলানোর মতো EC2 node group ছাড়া।

হাতে-কলমে EMR Serverless: একটা application বানাও (একটা release label আর একটা type, Spark বা Hive), তারপর একটা IAM runtime role দিয়ে job run জমা দাও, যেটা S3-এর অনুমতি দেয়। প্রথম job আসতেই application নিজে চালু হয়, আর default-এ ১৫ মিনিট অলস থাকলে বন্ধ হয়। Pre-initialized capacity worker-এর একটা গরম pool রাখে যাতে job কয়েক সেকেন্ডে শুরু হয় (application চালু থাকা পর্যন্ত তার দাম দিতে হয়), আর maximum capacity vCPU, memory আর disk-এর ছাদ বেঁধে দেয়, যাতে লাগামছাড়া job bill বাড়াতে না পারে।

EMR on EKS মানায় এমন company-কে যারা আগে থেকেই সবকিছু Amazon EKS-এ চালায় (১৪.৪): Spark job হয়ে যায় microservice-এর পাশের pod, EC2 (Spot সহ) বা Fargate node-এ, আর খালি capacity ভাগাভাগি হয়।

#এক পাতায় নিরাপত্তা

• Network: cluster থাকে একটা VPC subnet-এ, সবচেয়ে ভালো private subnet-এ, সাথে NAT gateway বা S3 আর অন্য service-এর জন্য VPC endpoint। EMR primary আর বাকি node-গুলোর জন্য EMR-managed security group বানায়; SSH বা notebook-এর জন্য নিজের group যোগ করো।
• IAM: তিনটা role। Service role EMR-কে EC2 instance আর অন্য resource বানানোর অনুমতি দেয়; EC2 instance profile দিয়ে cluster-এর job S3 আর অন্য service পড়ে (EMR Serverless আর EMR on EKS-এ প্রতি job-এর runtime বা execution role); আর IAM policy ঠিক করে কোন user cluster বানাতে বা কাজ জমা দিতে পারবে।
• Encryption একবারেই ঠিক করা হয় একটা security configuration-এ: EMRFS দিয়ে S3-এ লেখা ডেটার at-rest encryption (SSE-S3, SSE-KMS বা client-side), local disk আর EBS volume-এর জন্য, আর node-গুলোর মধ্যে TLS দিয়ে in-transit। KMS key আছে নিরাপত্তা অধ্যায়ের KMS লেসনে।
• সূক্ষ্ম অনুমতি: user authentication-এর জন্য Kerberos, আর S3-এর ডেটায় table, column আর row পর্যায়ের অনুমতির জন্য Lake Formation (লেসন ১৭.৬)।
• Catalog: Hive আর Spark-এর metastore হিসেবে Glue Data Catalog দাও, যাতে table-এর সংজ্ঞা transient cluster-এর পরেও থাকে, আর Athena ও Redshift Spectrum-এর সাথে ভাগ হয়।

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

• Transient cluster শেষ হতেই ফল উধাও। ফল লেখা হয়েছিল HDFS-এ। Output S3-এ লেখো, নয়তো terminate-এর আগে copy করার একটা step (s3-dist-cp) যোগ করো।
• পুরো cluster হঠাৎ হঠাৎ মরে যায়। একমাত্র primary node Spot-এ ছিল, ফেরত নিয়ে নিয়েছে। On-Demand রাখো, অথবা long-running cluster-এ তিনটা primary নাও।
• HDFS block না পেয়ে job fail। Core node ছিল Spot-এ, নয়তো তিন বা তার কম core node আর replication factor ১। Core On-Demand, আরও core node, ডেটা S3-এ।
• Spot task node গেলেই job fail। Application master চলছিল Spot node-এ। YARN node label দিয়ে application master core বা On-Demand node-এ রাখো।
• Spot core group কখনো শুরুই হয় না। Uniform group চাওয়া প্রতিটা Spot instance পাওয়া পর্যন্ত অপেক্ষা করে। কয়েকটা instance type আর On-Demand-এ চলে যাওয়ার timeout সহ instance fleet নাও।
• Job-এর মাঝখানে HDFS ভরে যায়। Replication factor দিয়ে ভাগ করার পর input, মাঝের ডেটা আর output, তিনটাকেই আঁটতে হবে। Core capacity বাড়াও, নয়তো input-output S3-এ সরাও।
• Trino বা HBase-এর জন্য managed scaling কিছুই করে না। এটা শুধু YARN application scale করে; instance group-এ custom automatic scaling নাও।
• AZ বিপর্যয়ে রাতের job থেমে গেল। Cluster থাকে একটা AZ-এ। ডেটা S3-এ রাখো আর অন্য AZ-এ আবার চালু করো (কয়েকটা subnet সহ fleet launch-এর সময় এটা নিজে করে)।
• Bill চলতেই থাকে। কেউ ভুলে যাওয়া test cluster, কিংবা fail করেও cluster বন্ধ না করা step। Auto-termination policy, step fail করলে “cluster terminate”, budget আর alarm (অধ্যায় ১৯, CloudWatch)।
• EC2 সীমার কারণে cluster বানানো fail। প্রতি Region-এর On-Demand vCPU quota EMR-এর বেলাতেও খাটে; বড় রাতের আগেই বাড়ানোর অনুরোধ করো।

#System design-এ EMR

চেনা pattern:

• প্রতিবার চালানোর জন্য একটা transient ETL cluster, Step Functions বা Amazon Managed Workflows for Apache Airflow দিয়ে সাজানো, ডেটা S3-এ, catalog Glue-তে। Batch-এর default।
• ভাগাভাগি করা long-running cluster, interactive analytics আর notebook-এর জন্য, Spot task node সহ managed scaling, স্থির অংশের জন্য Reserved বা Savings Plans, তিনটা primary।
• HBase on S3: বড়, বেশি পড়া হয় এমন key-value store, যেটা cluster বদলালেও টিকে থাকতে হবে।
• Hadoop হুবহু তুলে আনা: on-premises Hadoop cluster একই Hive, Spark আর HBase job নিয়ে EMR-এ আসে; HDFS-এর ডেটা যায় S3-এ (অনেক বড় হলে DataSync বা Snowball দিয়ে, অধ্যায় ১২)।
• Serverless Spark: যে team শুধু job জমা দিতে চায় তার জন্য EMR Serverless; crawler, catalog আর job bookmark-ও চাইলে Glue।
• Streaming: EMR-এ Spark Structured Streaming বা Flink, Kinesis বা MSK থেকে পড়ে; পুরোপুরি managed বিকল্প হলো Managed Service for Apache Flink (১৭.৭), আর পুরো serverless pipeline আছে লেসন ১৭.৮-এ।

দ্রুত বাছার উপায়:

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

এই command-গুলো সত্যিকারের resource বানায়, টাকা খরচ হয়; একটা sandbox account-এ চালাও আর শেষে সব terminate করো।

১. Default role বানাও (account-এ একবার):

aws emr create-default-roles

[
  { "RoleName": "EMR_DefaultRole", "Arn": "arn:aws:iam::111122223333:role/EMR_DefaultRole", ... },
  { "RoleName": "EMR_EC2_DefaultRole", "Arn": "arn:aws:iam::111122223333:role/EMR_EC2_DefaultRole", ... }
]

২. একটা transient cluster চালু করো, uniform instance group দিয়ে: primary আর core On-Demand, task Spot, একটা Spark step, log S3-এ, কাজ শেষে terminate।

aws emr create-cluster \
  --name "cdr-nightly-2026-10-29" \
  --release-label emr-7.12.0 \
  --applications Name=Spark Name=Hadoop \
  --service-role EMR_DefaultRole \
  --ec2-attributes InstanceProfile=EMR_EC2_DefaultRole,SubnetId=subnet-0a1b2c3d4e5f67890 \
  --instance-groups \
    InstanceGroupType=MASTER,InstanceCount=1,InstanceType=m5.xlarge \
    InstanceGroupType=CORE,InstanceCount=4,InstanceType=m5.xlarge \
    InstanceGroupType=TASK,InstanceCount=8,InstanceType=m5.xlarge,Market=SPOT \
  --log-uri s3://cdr-logs-example/emr/ \
  --steps Type=Spark,Name=aggregate,ActionOnFailure=TERMINATE_CLUSTER,\
Args=[s3://cdr-code-example/aggregate.py,--date,2026-10-29] \
  --auto-terminate

{
    "ClusterId": "j-2AXXXXXXGAPLF",
    "ClusterArn": "arn:aws:elasticmapreduce:ap-southeast-1:111122223333:cluster/j-2AXXXXXXGAPLF"
}

(তোমার Region-এ সবচেয়ে নতুন যে release আছে সেটাই নাও; aws emr list-release-labels তালিকা দেখায়।)

৩. নজর রাখো আর instance-গুলোর তালিকা দেখো:

aws emr describe-cluster --cluster-id j-2AXXXXXXGAPLF \
  --query 'Cluster.{State:Status.State,AZ:Ec2InstanceAttributes.Ec2AvailabilityZone}'
aws emr list-instances --cluster-id j-2AXXXXXXGAPLF --instance-group-types TASK \
  --query 'Instances[].{Id:Ec2InstanceId,Market:Market,State:Status.State}' --output table

{ "State": "RUNNING", "AZ": "ap-southeast-1a" }
-------------------------------------------------
|                 ListInstances                 |
+----------------------+---------+--------------+
|          Id          | Market  |    State     |
+----------------------+---------+--------------+
|  i-0f1e2d3c4b5a69788 |  SPOT   |  RUNNING     |
|  i-0a9b8c7d6e5f43210 |  SPOT   |  RUNNING     |
|  ...                 |  ...    |  ...         |
+----------------------+---------+--------------+

খেয়াল করো, Availability Zone একটাই।

৪. একটা long-running cluster-এ managed scaling policy (সর্বনিম্ন ২, সর্বোচ্চ ২০, On-Demand সর্বোচ্চ ৪, core সর্বোচ্চ ৪) আর একটা step যোগ করো:

aws emr put-managed-scaling-policy --cluster-id j-3BYYYYYYHBQMG \
  --managed-scaling-policy ComputeLimits='{UnitType=Instances,MinimumCapacityUnits=2,MaximumCapacityUnits=20,MaximumOnDemandCapacityUnits=4,MaximumCoreCapacityUnits=4}'

aws emr add-steps --cluster-id j-3BYYYYYYHBQMG \
  --steps Type=Hive,Name=daily-towers,ActionOnFailure=CONTINUE,\
Args=[-f,s3://cdr-code-example/towers.q]

{ "StepIds": [ "s-1K2L3M4N5O6P7" ] }

৫. Terminate করো (termination protection চালু থাকা long-running cluster-এ আগে সেটা বন্ধ করতে হয়):

aws emr modify-cluster-attributes --cluster-id j-3BYYYYYYHBQMG --no-termination-protected
aws emr terminate-clusters --cluster-ids j-3BYYYYYYHBQMG

৬. একই job EMR Serverless-এ: maximum capacity সহ application বানাও, job চালাও, খোঁজ নাও।

aws emr-serverless create-application \
  --name cdr-spark --type SPARK --release-label emr-7.12.0 \
  --maximum-capacity '{"cpu":"100vCPU","memory":"800GB"}'

{ "applicationId": "00fabcd1234efgh5", "name": "cdr-spark",
  "arn": "arn:aws:emr-serverless:ap-southeast-1:111122223333:/applications/00fabcd1234efgh5" }

aws emr-serverless start-job-run --application-id 00fabcd1234efgh5 \
  --execution-role-arn arn:aws:iam::111122223333:role/cdr-job-runtime-role \
  --job-driver '{"sparkSubmit":{"entryPoint":"s3://cdr-code-example/aggregate.py","entryPointArguments":["--date","2026-10-29"]}}'

aws emr-serverless get-job-run --application-id 00fabcd1234efgh5 --job-run-id 00fabcd9876xyz12 \
  --query 'jobRun.{state:state,vCPUHours:totalResourceUtilization.vCPUHour}'

{ "applicationId": "00fabcd1234efgh5", "jobRunId": "00fabcd9876xyz12", ... }
{ "state": "SUCCESS", "vCPUHours": 41.3 }

Console-এ, Spot task group চলার সময় cluster-এর Instances tab খোলো, আর প্রতিটা step-এর log দেখতে Steps tab। EMR Serverless-এ শেষ হওয়া job run-এর Spark UI link দেখায় প্রতিটা stage কয়টা executor ব্যবহার করেছে।

#সারাংশ

মূল কথাগুলো

• Amazon EMR (Elastic MapReduce) একটা managed big-data platform: EC2 instance-এর cluster চালু করে (শত শত বা হাজার হাজার পর্যন্ত), Spark, Hadoop, Hive, HBase, Presto/Trino, Flink আর আরও কিছু install করে, আর scale করে। ব্যবহার: data processing আর ETL, machine learning, web indexing, big data analytics। Keyword: Hadoop, Spark, Hive, HBase, Hadoop migrate।
• Primary node (আগের নাম master): YARN ResourceManager, HDFS NameNode, job আর স্বাস্থ্যের হিসাব। একটা, অথবা HA-র জন্য তিনটা। একমাত্র primary হারালে cluster terminate।
• Core node: task চালায় আর HDFS ডেটা রাখে। একটা হারালে block হারাতে পারে; ১–৩ / ৪–৯ / ১০+ core node-এ HDFS replication factor ১ / ২ / ৩। Task node (ঐচ্ছিক): শুধু compute, HDFS নেই, সাধারণত Spot।
• Purchasing: On-Demand (নির্ভরযোগ্য), Reserved (সর্বনিম্ন ১ বছর, নিজে থেকেই খাটে) আর Savings Plans (স্থির অংশ), Spot (প্রায় ৯০% পর্যন্ত সস্তা, দুই মিনিটের notice)। ওপরে EMR fee, প্রতি সেকেন্ডে। পরীক্ষার উত্তর: primary আর core On-Demand/Reserved, task Spot। Primary-তে Spot চলে, কিন্তু পরামর্শ দেওয়া হয় না, শুধু test বা খরচ-কেন্দ্রিক পুরো-Spot cluster ছাড়া।
• Transient cluster নিজের step চালিয়ে terminate হয় (batch, প্রতি run-এ দাম); long-running cluster অনেক user-কে সেবা দেয় (termination protection, Reserved দিয়ে স্থির অংশ, scaling)। Auto-termination অলস cluster থামায়।
• HDFS ক্ষণস্থায়ী (core node-এর disk, cluster-এর সাথেই শেষ)। Input আর output রাখো EMRFS দিয়ে S3-এ (EMR 7.10 থেকে S3A); S3 strongly consistent, তাই “consistent view” আর লাগে না।
• একটা cluster চলে একটা AZ-এ; টিকে থাকা = ডেটা S3-এ + অন্য জায়গায় আবার চালু।
• Instance fleet: ৫টা পর্যন্ত instance type (allocation strategy সহ ৩০), On-Demand আর Spot-এর লক্ষ্য, কয়েকটা subnet, Spot timeout → On-Demand। Uniform instance group: প্রতি group-এ একটা type আর একটা purchasing option, custom automatic scaling।
• Managed scaling: সর্বনিম্ন, সর্বোচ্চ, On-Demand সীমা, সর্বোচ্চ core; শুধু YARN application। বাড়ো task node দিয়ে।
• EMR on EC2 (পুরো নিয়ন্ত্রণ) বনাম EMR on EKS (তোমার Kubernetes-এ Spark) বনাম EMR Serverless (Spark/Hive job run, প্রতি সেকেন্ডে bill, pre-initialized capacity, maximum capacity, ১৫ মিনিট অলস থাকলে নিজে বন্ধ)।

নিজেকে যাচাই করো: Amazon EMR৩২টি প্রশ্ন