TL;DR (Tóm tắt)
Các API của AWS EMR (Elastic MapReduce) giúp bạn quản lý các cụm dữ liệu lớn chạy Hadoop, Spark, Hive và Presto. Bạn có thể tạo cụm, gửi công việc dưới dạng các bước, tự động điều chỉnh quy mô dựa trên tải, và chấm dứt khi hoàn thành. Xác thực sử dụng AWS IAM. Để kiểm thử, sử dụng Apidog để xác thực cấu hình cụm, kiểm thử gửi công việc dựa trên API, và ghi lại quy trình xử lý dữ liệu.
Giới thiệu
AWS EMR là dịch vụ quản lý Hadoop/Spark trên AWS, giúp xử lý petabyte dữ liệu cho analytics, machine learning và ETL. Thay vì tự quản lý cụm Hadoop, bạn để AWS xử lý hạ tầng.
EMR chạy trên các phiên bản EC2 trong một cụm. Bạn cần chỉ định:
- Loại phiên bản (nút master, core, task)
- Ứng dụng (Spark, Hadoop, Hive, Presto, HBase)
- Hành động khởi động (script thiết lập)
- Các bước (công việc cần chạy)
EMR API cho phép bạn tự động hóa toàn bộ quy trình: tạo cụm, gửi job, giám sát và tích hợp với dịch vụ AWS khác.
💡 Nếu bạn xây dựng pipeline dữ liệu, Apidog giúp kiểm thử cấu hình cụm, xác thực job, và ghi lại workflow EMR trước khi chạy các job tốn kém.
Kiểm thử API AWS với Apidog - miễn phí
Đến cuối bài viết bạn sẽ biết cách:
- Tạo và cấu hình cụm EMR qua API
- Gửi job dưới dạng các bước
- Quản lý tự động điều chỉnh quy mô
- Giám sát cụm và tiến độ công việc
- Tối ưu hóa chi phí với instance fleets và spot instances
Xác thực với AWS
EMR sử dụng xác thực tiêu chuẩn AWS IAM.
Sử dụng AWS SDK (Khuyến nghị)
import { EMRClient, RunJobFlowCommand } from '@aws-sdk/client-emr'
const client = new EMRClient({
region: 'us-east-1',
credentials: {
accessKeyId: process.env.AWS_ACCESS_KEY_ID,
secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY
}
})
Gọi API trực tiếp với SigV4
EMR yêu cầu AWS Signature Version 4. Dùng SDK, boto3, AWS CLI, hoặc ký thủ công.
aws emr list-clusters --region us-east-1
Quyền IAM cơ bản
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:*",
"ec2:Describe*",
"s3:GetObject",
"s3:PutObject",
"s3:DeleteObject",
"s3:ListBucket"
],
"Resource": "*"
}
]
}
Tạo một cụm EMR
Tạo cụm cơ bản bằng CLI
aws emr create-cluster \
--name "My Spark Cluster" \
--release-label emr-7.0.0 \
--applications Name=Spark Name=Hadoop \
--instance-type m5.xlarge \
--instance-count 3 \
--service-role EMR_DefaultRole \
--job-flow-role EMR_EC2_DefaultRole
Tạo cụm qua API (RunJobFlow)
{
"Name": "Data Processing Cluster",
"ReleaseLabel": "emr-7.0.0",
"Applications": [
{ "Name": "Spark" },
{ "Name": "Hadoop" },
{ "Name": "Hive" }
],
"Instances": {
"MasterInstanceType": "m5.xlarge",
"SlaveInstanceType": "m5.xlarge",
"InstanceCount": 3,
"KeepJobFlowAliveWhenNoSteps": true,
"TerminationProtected": false
},
"Steps": [],
"ServiceRole": "EMR_DefaultRole",
"JobFlowRole": "EMR_EC2_DefaultRole",
"LogUri": "s3://my-bucket/emr-logs/",
"Tags": [
{ "Key": "Environment", "Value": "Production" }
]
}
Phản hồi:
{
"JobFlowId": "j-ABC123DEF456"
}
Instance groups vs Instance fleets
Instance groups: Loại phiên bản cố định trên từng nhóm (master, core, task).
Instance fleets: Nhiều loại phiên bản trên mỗi nhóm. EMR tự chọn dựa vào giá cả và khả dụng.
{
"Instances": {
"InstanceFleets": [
{
"Name": "MasterFleet",
"InstanceFleetType": "MASTER",
"TargetOnDemandCapacity": 1,
"InstanceTypeConfigs": [
{ "InstanceType": "m5.xlarge" },
{ "InstanceType": "m4.xlarge" }
]
},
{
"Name": "CoreFleet",
"InstanceFleetType": "CORE",
"TargetOnDemandCapacity": 2,
"TargetSpotCapacity": 4,
"InstanceTypeConfigs": [
{ "InstanceType": "m5.2xlarge" },
{ "InstanceType": "m4.2xlarge" }
],
"LaunchSpecifications": {
"SpotSpecification": {
"TimeoutDurationMinutes": 60,
"TimeoutAction": "SWITCH_TO_ON_DEMAND"
}
}
}
]
}
}
Gửi công việc dưới dạng các bước
EMR xử lý công việc qua các "bước" (steps) theo thứ tự.
Thêm bước Spark bằng AWS CLI
aws emr add-steps \
--cluster-id j-ABC123DEF456 \
--steps '[
{
"Name": "Process Data",
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"spark-submit",
"--deploy-mode",
"cluster",
"--class",
"com.example.DataProcessor",
"s3://my-bucket/jars/processor.jar",
"s3://my-bucket/input/",
"s3://my-bucket/output/"
]
}
}
]'
Thêm bước qua API (AddJobFlowSteps)
{
"JobFlowId": "j-ABC123DEF456",
"Steps": [
{
"Name": "Spark ETL Job",
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"spark-submit",
"--executor-memory",
"4g",
"--executor-cores",
"2",
"s3://my-bucket/scripts/process.py",
"--input",
"s3://my-bucket/input/",
"--output",
"s3://my-bucket/output/"
]
}
}
]
}
ActionOnFailure options
-
TERMINATE_CLUSTER: Dừng cụm khi step fail -
CANCEL_AND_WAIT: Hủy các bước còn lại, giữ cụm chạy -
CONTINUE: Tiếp tục với bước tiếp theo
Bước Hive
{
"Name": "Hive Query",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"hive-script",
"--run-hive-script",
"--args",
"-f",
"s3://my-bucket/scripts/transform.q"
]
}
}
Tự động điều chỉnh quy mô (Auto-scaling)
EMR tự động thêm/xóa nút task dựa trên tải.
Tạo policy auto-scaling
aws emr put-auto-scaling-policy \
--cluster-id j-ABC123DEF456 \
--instance-group-id ig-ABC123 \
--auto-scaling-policy '{
"Constraints": {
"MinCapacity": 2,
"MaxCapacity": 10
},
"Rules": [
{
"Name": "ScaleOut",
"Description": "Add nodes when memory is high",
"Action": {
"SimpleScalingPolicyConfiguration": {
"AdjustmentType": "CHANGE_IN_CAPACITY",
"ScalingAdjustment": 2,
"CoolDown": 300
}
},
"Trigger": {
"CloudWatchAlarmDefinition": {
"ComparisonOperator": "GREATER_THAN",
"EvaluationPeriods": 3,
"MetricName": "MemoryAvailableMB",
"Namespace": "AWS/ElasticMapReduce",
"Period": 300,
"Threshold": 2000,
"Statistic": "AVERAGE"
}
}
}
]
}'
Các chỉ số phổ biến để auto-scale
-
MemoryAvailableMB: Bộ nhớ còn lại -
MemoryTotalMB: Tổng bộ nhớ -
HDFSUtilization: % HDFS đã dùng -
AppsRunning: Ứng dụng YARN đang chạy -
AppsPending: Ứng dụng YARN chờ xử lý
Giám sát và ghi nhật ký
Liệt kê các cụm
aws emr list-clusters --states RUNNING
Mô tả cụm
aws emr describe-cluster --cluster-id j-ABC123DEF456
Phản hồi mẫu:
{
"Cluster": {
"Id": "j-ABC123DEF456",
"Name": "My Cluster",
"Status": {
"State": "RUNNING",
"StateChangeReason": {},
"Timeline": {
"CreationDateTime": "2026-03-24T10:00:00.000Z"
}
},
"Applications": [
{ "Name": "Spark", "Version": "3.5.0" }
],
"InstanceCollectionType": "INSTANCE_GROUP",
"LogUri": "s3://my-bucket/emr-logs/",
"MasterPublicDnsName": "ec2-12-34-56-78.compute-1.amazonaws.com"
}
}
Liệt kê các bước
aws emr list-steps --cluster-id j-ABC123DEF456
Trạng thái bước
{
"Id": "s-ABC123",
"Name": "Process Data",
"Status": {
"State": "COMPLETED",
"Timeline": {
"StartDateTime": "2026-03-24T10:05:00.000Z",
"EndDateTime": "2026-03-24T11:30:00.000Z"
}
}
}
Tích hợp CloudWatch
EMR xuất các chỉ số lên CloudWatch:
-
JobsFailed: Công việc thất bại -
JobsRunning: Công việc đang chạy -
MemoryAvailableMB: Bộ nhớ khả dụng -
MemoryTotalMB: Tổng bộ nhớ -
HDFSUtilization: % sử dụng HDFS
Tối ưu hóa chi phí
Sử dụng Spot Instances
Nút task phù hợp để chạy Spot. Nếu bị chấm dứt, các job vẫn chạy trên các nút còn lại.
{
"Name": "TaskGroup",
"InstanceRole": "TASK",
"InstanceType": "m5.2xlarge",
"InstanceCount": 4,
"Market": "SPOT",
"BidPrice": "0.10"
}
Cụm tạm thời (Transient clusters)
Tự động chấm dứt khi xong việc:
{
"KeepJobFlowAliveWhenNoSteps": false,
"Steps": [
{ ... step 1 ... },
{ ... step 2 ... }
]
}
Instance fleet với nhiều lựa chọn
Để EMR chọn loại phiên bản rẻ nhất sẵn có:
{
"InstanceTypeConfigs": [
{
"InstanceType": "m5.2xlarge",
"BidPrice": "0.15"
},
{
"InstanceType": "m4.2xlarge",
"BidPrice": "0.12"
},
{
"InstanceType": "c5.2xlarge",
"BidPrice": "0.10"
}
]
}
Kiểm thử với Apidog
Các cụm EMR rất tốn kém – kiểm thử cấu hình kỹ càng là cần thiết.
1. Xác thực cấu hình cụm
Lưu mẫu cluster trong Apidog và kiểm thử:
pm.test('Cluster has required applications', () => {
const config = pm.request.body.toJSON()
const apps = config.Applications.map(a => a.Name)
pm.expect(apps).to.include('Spark')
})
pm.test('Instance types are valid', () => {
const config = pm.request.body.toJSON()
const types = ['m5.xlarge', 'm5.2xlarge', 'm4.xlarge']
pm.expect(types).to.include(config.Instances.MasterInstanceType)
})
2. Kiểm thử định nghĩa bước
pm.test('Spark step has valid args', () => {
const step = pm.request.body.toJSON().Steps[0]
const args = step.HadoopJarStep.Args
pm.expect(args[0]).to.eql('spark-submit')
pm.expect(args).to.include('--deploy-mode')
})
3. Biến môi trường
AWS_REGION: us-east-1
EMR_SERVICE_ROLE: EMR_DefaultRole
EMR_EC2_ROLE: EMR_EC2_DefaultRole
S3_LOG_BUCKET: my-emr-logs
S3_SCRIPTS_BUCKET: my-emr-scripts
Kiểm thử API AWS với Apidog - miễn phí
Các lỗi thường gặp và cách khắc phục
ValidationError: ServiceRole không hợp lệ
Nguyên nhân: Vai trò IAM chưa tồn tại hoặc chưa cấu hình cho EMR.
Khắc phục: Tạo vai trò dịch vụ trong IAM hoặc dùng vai trò mặc định: EMR_DefaultRole_V2.
Không thể cung cấp các phiên bản EC2
Nguyên nhân: Loại phiên bản không có sẵn trong AZ hoặc vượt giới hạn dịch vụ.
Khắc phục:
- Dùng instance fleet với nhiều loại phiên bản
- Yêu cầu tăng giới hạn
- Thử loại phiên bản khác
Bước thất bại với mã thoát ứng dụng 1
Nguyên nhân: Job Spark/Hadoop thực tế bị lỗi.
Khắc phục: Kiểm tra log S3 (LogUri). Xem stderr và stdout của step.
Cụm bị kẹt ở trạng thái STARTING
Nguyên nhân: Lỗi ở bootstrap actions hoặc quyền truy cập.
Khắc phục: Kiểm tra console output của EC2. Đảm bảo quyền truy cập S3 cho script khởi động.
Các giải pháp thay thế và so sánh
| Tính năng | AWS EMR | Google Dataproc | Azure HDInsight | Databricks |
|---|---|---|---|---|
| Hadoop/Spark được quản lý | ✓ | ✓ | ✓ | Chỉ Spark |
| Tích hợp AWS | Tuyệt vời | Hạn chế | Hạn chế | Tốt |
| Tùy chọn Serverless | EMR Serverless | Dataproc Serverless | Hạn chế | ✓ |
| Chi phí | Hỗ trợ Spot | VM có thể bị ngắt (Preemptible VMs) | Phiên bản Spot | Tốt |
| Hỗ trợ học máy (ML) | EMR Studio | Vertex AI | Synapse | Tích hợp MLflow |
EMR tích hợp sâu với AWS. Databricks mạnh về Spark. Dataproc tiết kiệm hơn cho khách hàng GCP.
Các trường hợp sử dụng thực tế
ETL hồ dữ liệu: Công ty bán lẻ xử lý dữ liệu bán hàng hàng ngày. EMR nhập file CSV từ S3, biến đổi với Spark, ghi Parquet vào data lake. Cụm chỉ chạy 2 giờ/ngày và tự động chấm dứt.
Phân tích log: SaaS xử lý log ứng dụng. Spark đọc log từ S3, tổng hợp, ghi về kho dữ liệu. Auto-scaling tự động thêm task node khi tải tăng.
Quy trình Machine Learning: Nhóm data science huấn luyện model trên EMR. Spark đọc feature từ S3, huấn luyện với MLlib, xuất model lên SageMaker để triển khai.
Tổng kết
Bạn đã biết:
- Tạo cụm bằng API RunJobFlow
- Gửi job dưới dạng các bước
- Sử dụng auto-scaling tối ưu chi phí
- Giám sát với CloudWatch
- Tối ưu chi phí nhờ Spot instance và cụm tạm thời
Các bước tiếp theo:
- Thiết lập IAM role cho EMR
- Tạo một cụm kiểm thử
- Gửi job Spark đơn giản
- Xem lại log trên S3
- Thực hiện các chiến lược tiết kiệm chi phí
Kiểm thử API AWS với Apidog - miễn phí
Câu hỏi thường gặp
Sự khác biệt giữa master, core, task node?
- Master: Quản lý cluster (YARN ResourceManager, HDFS NameNode)
- Core: Xử lý và lưu trữ dữ liệu HDFS
- Task: Chỉ xử lý dữ liệu, không lưu HDFS (tốt cho Spot)
Làm sao SSH vào master node?
aws emr ssh --cluster-id j-ABC123DEF456 --key-pair-file my-key.pem
Có chạy Jupyter notebook trên EMR được không?
Có. Dùng EMR Studio, bật JupyterHub hoặc EMR Notebooks (Jupyter managed).
EMR Serverless là gì?
Tùy chọn serverless: gửi job Spark/Hive mà không phải quản lý cụm, trả tiền theo job. Phù hợp workload không liên tục.
Làm sao đọc từ DynamoDB?
spark-submit --conf spark.hadoop.dynamodb.servicename=dynamodb \
--conf spark.hadoop.dynamodb.input.tableName=MyTable \
--conf spark.hadoop.dynamodb.output.tableName=MyTable \
--conf spark.hadoop.dynamodb.region=us-east-1 \
my-job.jar
Nên dùng release label nào?
Dùng bản ổn định mới nhất (emr-7.x cho Spark 3.x). Đảm bảo đồng nhất version giữa môi trường. Kiểm tra compatibility trong release notes.
Khắc phục bước bị lỗi thế nào?
- Kiểm tra trạng thái bước:
aws emr describe-step - Xem log trong S3:
s3://your-log-bucket/logs/j-ABC123/steps/s-DEF123/ - SSH vào master, kiểm tra
/mnt/var/log/
Top comments (0)