{"id":4635,"date":"2026-07-23T14:15:48","date_gmt":"2026-07-23T14:15:48","guid":{"rendered":"https:\/\/rssfeedtelegrambot.bnaya.co.il\/index.php\/2026\/07\/23\/building-reliable-emr-pipelines-with-custom-amis-and-step-functions\/"},"modified":"2026-07-23T14:15:48","modified_gmt":"2026-07-23T14:15:48","slug":"building-reliable-emr-pipelines-with-custom-amis-and-step-functions","status":"publish","type":"post","link":"https:\/\/rssfeedtelegrambot.bnaya.co.il\/index.php\/2026\/07\/23\/building-reliable-emr-pipelines-with-custom-amis-and-step-functions\/","title":{"rendered":"Building Reliable EMR Pipelines With Custom AMIs and Step Functions"},"content":{"rendered":"
\"OWASP<\/div>\n

\"OWASP<\/p>\n

A practical pattern for replacing runtime bootstrap fragility with image-based dependencies, orchestration and retry-aware recovery.<\/p>\n

The Problem: Dependency Drift in EMR Pipelines<\/h3>\n

Amazon EMR remains a strong option for running Apache Spark workloads, but many production-style pipelines still rely on bootstrap actions to install dependencies when a cluster starts. That works for early experiments. It becomes fragile when pipelines need predictable startup behavior, repeatable dependency versions and cleaner recovery from transient infrastructure failures.<\/p>\n

The common failure mode is simple: Every cluster launch repeats package installation, network calls and setup scripts. If a package repository is slow, a dependency resolver changes behavior or a security patch needs to be rolled across many jobs, the pipeline team is debugging environment creation instead of data logic.<\/p>\n

One way to reduce that operational noise is to treat the EMR runtime as immutable infrastructure: Custom AMIs with pre-baked dependencies, automated image builds, versioned metadata, Step Functions orchestration and failure handling that understands retryable conditions.<\/p>\n

Why Custom AMIs Beat Bootstrap Actions<\/h3>\n

Architecture Overview<\/h3>\n

\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\n\u2502 CodeCommit \u2502\u2500\u2500\u2500\u2500\"\u25b6\"\u2502 Lambda Builder \u2502\u2500\u2500\u2500\u2500\"\u25b6\"\u2502 EC2 (Packer) \u2502
\n\u2502 (requirements) \u2502 \u2502 (trigger AMI \u2502 \u2502 (Custom AMI) \u2502
\n\u2502 \u2502 \u2502 build) \u2502 \u2502 \u2502
\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n\u2502
\n\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n\u25bc
\n\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\n\u2502 Golden AMI \u2502
\n\u2502 (EMR + Python) \u2502
\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n\u2502
\n\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u2502 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\n\u2502 Step Functions \u2502\"\u25c0\"\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\"\u25b6\"\u2502 EMR Cluster \u2502
\n\u2502 Orchestrator \u2502 \u2502 (Custom AMI) \u2502
\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n\u2502 \u2502
\n\u2502 Error? \u2502
\n\u2502\"\u25c0\"\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500 Retry logic \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\n\u2502
\n\u250c\u2500\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2510
\n\u2502 DLQ \u2502 (CloudWatch analysis)
\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518<\/p>\n

Part 1: Building the Custom AMI<\/h3>\n

Step 1: Create the Packer Configuration<\/h3>\n

Use HashiCorp Packer to automate AMI builds. This creates a reproducible, version-controlled image pipeline.<\/p>\n

# emr-custom-ami.pkr.hcl
\npacker {
\nrequired_plugins {
\namazon = {
\nsource = \u201cgithub.com\/hashicorp\/amazon\u201d
\nversion = \u201c~> 1\u201d
\n}
\n}
\n}<\/p>\n

variable \u201cemr_release\u201d {
\ndefault = \u201cemr-6.15.0\u201d
\n}<\/p>\n

variable \u201cpython_packages\u201d {
\ndefault = [
\n\u201cpandas==2.0.3\u201d,
\n\u201cnumpy==1.24.3\u201d,
\n\u201cboto3==1.28.25\u201d,
\n\u201cpyspark==3.4.1\u201d,
\n\u201cgreat-expectations==0.17.0\u201d
\n]
\n}<\/p>\n

source \u201camazon-ebs\u201d \u201cemr-custom\u201d {
\nami_name = \u201cemr-${var.emr_release}-analytics-${formatdate(\u201dYYYYMMDD-hhmm\u201d, timestamp())}\u201d
\ninstance_type = \u201cm5.xlarge\u201d
\nregion = \u201cus-east-1\u201d<\/p>\n

# Start from official EMR AMI
\nsource_ami_filter {
\nfilters = {
\nname = \u201camzn2-ami-emr-${var.emr_release}*\u201d
\nroot-device-type = \u201cebs\u201d
\nvirtualization-type = \u201chvm\u201d
\n}
\nowners = [\u201camazon\u201d]
\nmost_recent = true
\n}<\/p>\n

ssh_username = \u201cec2-user\u201d<\/p>\n

# EBS optimization for Spark
\nlaunch_block_device_mappings {
\ndevice_name = \u201c\/dev\/xvda\u201d
\nvolume_size = 100
\nvolume_type = \u201cgp3\u201d
\ndelete_on_termination = true
\n}<\/p>\n

tags = {
\nName = \u201cEMR Analytics AMI\u201d
\nEMRRelease = var.emr_release
\nBuiltBy = \u201cPacker\u201d
\nRepository = \u201cgithub.com\/example-org\/emr-ami-pipeline\u201d
\n}
\n}<\/p>\n

build {
\nsources = [\u201csource.amazon-ebs.emr-custom\u201d]\n<\/p>\n

# Install Python 3.9 (EMR default is often 3.7)
\nprovisioner \u201cshell\u201d {
\ninline = [
\n\u201csudo amazon-linux-extras install python3.9 -y\u201d,
\n\u201csudo alternatives \u2013install \/usr\/bin\/python3 python3 \/usr\/bin\/python3.9 1\u201d,
\n\u201csudo python3.9 -m pip install \u2013upgrade pip setuptools wheel\u201d
\n]
\n}<\/p>\n

# Install Python dependencies
\nprovisioner \u201cshell\u201d {
\ninline = concat(
\n[\u201csudo python3.9 -m pip install \u201c],
\n[join(\u201d \u201c, var.python_packages)]
\n)
\n}<\/p>\n

# Verify installations
\nprovisioner \u201cshell\u201d {
\ninline = [
\n\u201cpython3 \u2013version\u201d,
\n\u201cpip3 list | grep pandas\u201d,
\n\u201cpip3 list | grep numpy\u201d
\n]
\n}<\/p>\n

# Clean up for AMI optimization
\nprovisioner \u201cshell\u201d {
\ninline = [
\n\u201csudo yum clean all\u201d,
\n\u201csudo rm -rf \/var\/cache\/yum\u201d,
\n\u201csudo rm -f \/home\/ec2-user\/.ssh\/authorized_keys\u201d,
\n\u201csudo rm -f \/root\/.ssh\/authorized_keys\u201d
\n]
\n}
\n}<\/p>\n

Step 2: Lambda-Powered AMI Builder<\/h3>\n

Trigger AMI builds automatically when requirements.txt changes:<\/p>\n

# lambda\/ami_builder.py
\nimport json
\nimport boto3
\nimport os<\/p>\n

ec2 = boto3.client(\u2018ec2\u2019)
\nssm = boto3.client(\u2018ssm\u2019)<\/p>\n

def lambda_handler(event, context):
\n\u201c\u201d\u201d
\nTriggered by CodeCommit push to requirements.txt
\nBuilds new EMR custom AMI using Packer
\n\u201c\u201d\u201d<\/p>\n

# Get latest EMR release from SSM Parameter Store
\nemr_release = ssm.get_parameter(
\nName=\u2019\/emr-pipeline\/emr-release\u2019
\n)[\u2018Parameter\u2019][\u2018Value\u2019]\n<\/p>\n

# Start Packer build on EC2 instance (builder pattern)
\nuser_data = f\u201d\u2019#!\/bin\/bash
\nyum update -y
\nyum install -y packer git<\/p>\n

# Clone repo with Packer templates
\ncd \/tmp
\ngit clone https:\/\/github.com\/example-org\/emr-ami-pipeline.git
\ncd emr-ami-pipeline<\/p>\n

# Run Packer build
\npacker init emr-custom-ami.pkr.hcl
\npacker build
\n-var \u2019emr_release={emr_release}\u2019
\n-var \u2018python_packages_file=requirements.txt\u2019
\nemr-custom-ami.pkr.hcl<\/p>\n

# Notify Step Functions\/SNS on completion
\naws sns publish
\n\u2013topic-arn {os.environ[\u2018SNS_TOPIC_ARN\u2019]}
\n\u2013message \u201cAMI build completed\u201d
\n\u201d\u2019<\/p>\n

# Launch builder instance
\nresponse = ec2.run_instances(
\nImageId=\u2019ami-0c55b159cbfafe1f0\u2032, # Amazon Linux 2
\nInstanceType=\u2019m5.2xlarge\u2019,
\nMinCount=1,
\nMaxCount=1,
\nIamInstanceProfile={\u2018Name\u2019: \u2018EMR-AMI-Builder\u2019},
\nUserData=user_data,
\nTagSpecifications=[{
\n\u2018ResourceType\u2019: \u2018instance\u2019,
\n\u2018Tags\u2019: [
\n{\u2018Key\u2019: \u2018Name\u2019, \u2018Value\u2019: \u2019emr-ami-builder\u2019},
\n{\u2018Key\u2019: \u2018AutoTerminate\u2019, \u2018Value\u2019: \u2018true\u2019}
\n]
\n}]
\n)<\/p>\n

return {
\n\u2018statusCode\u2019: 200,
\n\u2018body\u2019: json.dumps({
\n\u2018message\u2019: \u2018AMI build initiated\u2019,
\n\u2018instanceId\u2019: response[\u2018Instances\u2019][0][\u2018InstanceId\u2019]
\n})
\n}<\/p>\n

Step 3: Store Golden AMI Metadata<\/h3>\n

Track AMI versions in DynamoDB for pipeline lookups:<\/p>\n

# lambda\/ami_registry.py
\nimport boto3
\nfrom datetime import datetime<\/p>\n

dynamodb = boto3.resource(\u2018dynamodb\u2019)
\ntable = dynamodb.Table(\u2019emr-golden-amis\u2019)<\/p>\n

def register_ami(ami_id, emr_release, packages, commit_hash):
\n\u201c\u201d\u201dRegister newly built AMI as \u2018active\u2019 for Step Functions.\u201d\u201d\u201d<\/p>\n

# Mark previous AMI as deprecated
\ntable.update_item(
\nKey={\u2019emr_release\u2019: emr_release, \u2018status\u2019: \u2018active\u2019},
\nUpdateExpression=\u2019SET #status = :deprecated\u2019,
\nExpressionAttributeNames={\u2018#status\u2019: \u2018status\u2019},
\nExpressionAttributeValues={\u2018:deprecated\u2019: \u2018deprecated\u2019}
\n)<\/p>\n

# Register new AMI
\ntable.put_item(Item={
\n\u2018ami_id\u2019: ami_id,
\n\u2019emr_release\u2019: emr_release,
\n\u2018status\u2019: \u2018active\u2019,
\n\u2018python_packages\u2019: packages,
\n\u2018commit_hash\u2019: commit_hash,
\n\u2018created_at\u2019: datetime.utcnow().isoformat(),
\n\u2018ttl\u2019: int((datetime.utcnow().timestamp()) + (90 * 86400)) # 90-day retention
\n})<\/p>\n

Part 2: Step Functions Orchestration<\/h3>\n

State Machine Definition<\/h3>\n

{
\n\u201cComment\u201d: \u201cEMR Data Pipeline with Custom AMI and Error Handling\u201d,
\n\u201cStartAt\u201d: \u201cGetGoldenAMI\u201d,
\n\u201cStates\u201d: {
\n\u201cGetGoldenAMI\u201d: {
\n\u201cType\u201d: \u201cTask\u201d,
\n\u201cResource\u201d: \u201carn:aws:states:::lambda:invoke\u201d,
\n\u201cParameters\u201d: {
\n\u201cFunctionName\u201d: \u201cget-golden-ami\u201d,
\n\u201cPayload\u201d: {
\n\u201cemr_release\u201d: \u201cemr-6.15.0\u201d
\n}
\n},
\n\u201cResultPath\u201d: \u201c$.ami\u201d,
\n\u201cNext\u201d: \u201cCreateEMRCluster\u201d
\n},<\/p>\n

\u201cCreateEMRCluster\u201d: {
\n\u201cType\u201d: \u201cTask\u201d,
\n\u201cResource\u201d: \u201carn:aws:states:::elasticmapreduce:createCluster.sync\u201d,
\n\u201cParameters\u201d: {
\n\u201cName\u201d: \u201cpipeline-${$.executionId}\u201d,
\n\u201cReleaseLabel\u201d: \u201cemr-6.15.0\u201d,
\n\u201cCustomAmiId.$\u201d: \u201c$.ami.Payload.ami_id\u201d,
\n\u201cLogUri\u201d: \u201cs3:\/\/emr-logs-bucket\/pipelines\/\u201d,
\n\u201cInstances\u201d: {
\n\u201cKeepJobFlowAliveWhenNoSteps\u201d: false,
\n\u201cInstanceFleets\u201d: [
\n{
\n\u201cName\u201d: \u201cMaster\u201d,
\n\u201cInstanceFleetType\u201d: \u201cMASTER\u201d,
\n\u201cTargetOnDemandCapacity\u201d: 1,
\n\u201cInstanceTypeConfigs\u201d: [
\n{\u201cInstanceType\u201d: \u201cm5.xlarge\u201d}
\n]
\n},
\n{
\n\u201cName\u201d: \u201cCore\u201d,
\n\u201cInstanceFleetType\u201d: \u201cCORE\u201d,
\n\u201cTargetSpotCapacity\u201d: 4,
\n\u201cInstanceTypeConfigs\u201d: [
\n{\u201cInstanceType\u201d: \u201cm5.2xlarge\u201d, \u201cWeightedCapacity\u201d: 2},
\n{\u201cInstanceType\u201d: \u201cm5.4xlarge\u201d, \u201cWeightedCapacity\u201d: 4}
\n]
\n}
\n],
\n\u201cEc2SubnetIds\u201d: [\u201csubnet-12345abcde\u201d],
\n\u201cEmrManagedMasterSecurityGroup\u201d: \u201csg-master\u201d,
\n\u201cEmrManagedSlaveSecurityGroup\u201d: \u201csg-core\u201d
\n},
\n\u201cApplications\u201d: [
\n{\u201cName\u201d: \u201cSpark\u201d},
\n{\u201cName\u201d: \u201cHadoop\u201d}
\n],
\n\u201cServiceRole\u201d: \u201cEMR_DefaultRole\u201d,
\n\u201cJobFlowRole\u201d: \u201cEMR_EC2_DefaultRole\u201d
\n},
\n\u201cResultPath\u201d: \u201c$.cluster\u201d,
\n\u201cNext\u201d: \u201cSubmitSparkJob\u201d
\n},<\/p>\n

\u201cSubmitSparkJob\u201d: {
\n\u201cType\u201d: \u201cTask\u201d,
\n\u201cResource\u201d: \u201carn:aws:states:::elasticmapreduce:addStep.sync\u201d,
\n\u201cParameters\u201d: {
\n\u201cClusterId.$\u201d: \u201c$.cluster.ClusterId\u201d,
\n\u201cStep\u201d: {
\n\u201cName\u201d: \u201cDataProcessing\u201d,
\n\u201cActionOnFailure\u201d: \u201cCONTINUE\u201d,
\n\u201cHadoopJarStep\u201d: {
\n\u201cJar\u201d: \u201ccommand-runner.jar\u201d,
\n\u201cArgs\u201d: [
\n\u201cspark-submit\u201d,
\n\u201c\u2013deploy-mode\u201d, \u201ccluster\u201d,
\n\u201c\u2013conf\u201d, \u201cspark.dynamicAllocation.enabled=true\u201d,
\n\u201cs3:\/\/emr-jobs\/process_data.py\u201d,
\n\u201c\u2013input\u201d, \u201cs3:\/\/data-bucket\/raw\/\u201d,
\n\u201c\u2013output\u201d, \u201cs3:\/\/data-bucket\/processed\/\u201d
\n]
\n}
\n}
\n},
\n\u201cResultPath\u201d: \u201c$.step\u201d,
\n\u201cNext\u201d: \u201cCheckStepStatus\u201d
\n},<\/p>\n

\u201cCheckStepStatus\u201d: {
\n\u201cType\u201d: \u201cChoice\u201d,
\n\u201cChoices\u201d: [
\n{
\n\u201cVariable\u201d: \u201c$.step.Step.State\u201d,
\n\u201cStringEquals\u201d: \u201cCOMPLETED\u201d,
\n\u201cNext\u201d: \u201cSuccess\u201d
\n},
\n{
\n\u201cVariable\u201d: \u201c$.step.Step.State\u201d,
\n\u201cStringEquals\u201d: \u201cFAILED\u201d,
\n\u201cNext\u201d: \u201cAnalyzeFailure\u201d
\n}
\n],
\n\u201cDefault\u201d: \u201cWaitForCompletion\u201d
\n},<\/p>\n

\u201cWaitForCompletion\u201d: {
\n\u201cType\u201d: \u201cWait\u201d,
\n\u201cSeconds\u201d: 60,
\n\u201cNext\u201d: \u201cCheckStepStatus\u201d
\n},<\/p>\n

\u201cAnalyzeFailure\u201d: {
\n\u201cType\u201d: \u201cTask\u201d,
\n\u201cResource\u201d: \u201carn:aws:states:::lambda:invoke\u201d,
\n\u201cParameters\u201d: {
\n\u201cFunctionName\u201d: \u201canalyze-emr-failure\u201d,
\n\u201cPayload\u201d: {
\n\u201ccluster_id.$\u201d: \u201c$.cluster.ClusterId\u201d,
\n\u201cstep_id.$\u201d: \u201c$.step.Step.Id\u201d,
\n\u201cattempt_count.$\u201d: \u201c$.attempt_count\u201d
\n}
\n},
\n\u201cResultPath\u201d: \u201c$.failure_analysis\u201d,
\n\u201cNext\u201d: \u201cShouldRetry\u201d
\n},<\/p>\n

\u201cShouldRetry\u201d: {
\n\u201cType\u201d: \u201cChoice\u201d,
\n\u201cChoices\u201d: [
\n{
\n\u201cAnd\u201d: [
\n{\u201cVariable\u201d: \u201c$.failure_analysis.Payload.retryable\u201d, \u201cBooleanEquals\u201d: true},
\n{\u201cVariable\u201d: \u201c$.attempt_count\u201d, \u201cNumericLessThan\u201d: 3}
\n],
\n\u201cNext\u201d: \u201cCalculateBackoff\u201d
\n}
\n],
\n\u201cDefault\u201d: \u201cSendToDLQ\u201d
\n},<\/p>\n

\u201cCalculateBackoff\u201d: {
\n\u201cType\u201d: \u201cTask\u201d,
\n\u201cResource\u201d: \u201carn:aws:states:::lambda:invoke\u201d,
\n\u201cParameters\u201d: {
\n\u201cFunctionName\u201d: \u201ccalculate-backoff\u201d,
\n\u201cPayload\u201d: {
\n\u201cattempt.$\u201d: \u201c$.attempt_count\u201d
\n}
\n},
\n\u201cResultPath\u201d: \u201c$.backoff\u201d,
\n\u201cNext\u201d: \u201cWaitBeforeRetry\u201d
\n},<\/p>\n

\u201cWaitBeforeRetry\u201d: {
\n\u201cType\u201d: \u201cWait\u201d,
\n\u201cSecondsPath\u201d: \u201c$.backoff.Payload.seconds\u201d,
\n\u201cNext\u201d: \u201cCreateEMRCluster\u201d
\n},<\/p>\n

\u201cSendToDLQ\u201d: {
\n\u201cType\u201d: \u201cTask\u201d,
\n\u201cResource\u201d: \u201carn:aws:states:::sqs:sendMessage\u201d,
\n\u201cParameters\u201d: {
\n\u201cQueueUrl\u201d: \u201chttps:\/\/sqs.us-east-1.amazonaws.com\/123456789\/emr-failures-dlq\u201d,
\n\u201cMessageBody\u201d: {
\n\u201cexecution_id.$\u201d: \u201c$$.Execution.Id\u201d,
\n\u201ccluster_id.$\u201d: \u201c$.cluster.ClusterId\u201d,
\n\u201cerror.$\u201d: \u201c$.failure_analysis.Payload.error_type\u201d,
\n\u201ctimestamp.$\u201d: \u201c$$.State.EnteredTime\u201d
\n}
\n},
\n\u201cNext\u201d: \u201cPipelineFailed\u201d
\n},<\/p>\n

\u201cSuccess\u201d: {
\n\u201cType\u201d: \u201cSucceed\u201d
\n},<\/p>\n

\u201cPipelineFailed\u201d: {
\n\u201cType\u201d: \u201cFail\u201d,
\n\u201cError\u201d: \u201cPipelineFailed\u201d,
\n\u201cCause\u201d: \u201cNon-retryable error or max retries exceeded\u201d
\n}
\n}
\n}<\/p>\n

Error Analysis Lambda<\/h3>\n

# lambda\/analyze_emr_failure.py
\nimport boto3
\nimport json<\/p>\n

emr = boto3.client(\u2019emr\u2019)
\nlogs = boto3.client(\u2018logs\u2019)<\/p>\n

def lambda_handler(event, context):
\ncluster_id = event[\u2018cluster_id\u2019]
\nstep_id = event[\u2018step_id\u2019]
\nattempt = event.get(\u2018attempt_count\u2019, 0)<\/p>\n

# Get step details
\nresponse = emr.describe_step(ClusterId=cluster_id, StepId=step_id)
\nstep_status = response[\u2018Step\u2019][\u2018Status\u2019]\n<\/p>\n

error_info = step_status.get(\u2018FailureDetails\u2019, {})
\nreason = error_info.get(\u2018Reason\u2019, \u2018Unknown\u2019)
\nlog_file = error_info.get(\u2018LogFile\u2019, \u201d)<\/p>\n

# Classify error type
\nretryable_errors = [
\n\u2018Spot instance termination\u2019,
\n\u2018Container killed on request\u2019,
\n\u2018Connection timeout\u2019,
\n\u2018ThrottlingException\u2019
\n]\n<\/p>\n

is_retryable = any(err in reason for err in retryable_errors)<\/p>\n

# If spot termination, suggest on-demand retry
\nif \u2018Spot\u2019 in reason and attempt == 2:
\nis_retryable = False # Third failure, use DLQ<\/p>\n

return {
\n\u2018retryable\u2019: is_retryable,
\n\u2018error_type\u2019: reason,
\n\u2018log_file\u2019: log_file,
\n\u2018recommendation\u2019: \u2018retry_on_demand\u2019 if \u2018Spot\u2019 in reason else \u2018retry\u2019
\n}<\/p>\n

def calculate_backoff(attempt):
\n\u201c\u201d\u201dExponential backoff: 30s, 2min, 5min\u201d\u201d\u201d
\ndelays = [30, 120, 300]
\nreturn {\u2018seconds\u2019: delays[min(attempt, 2)]}<\/p>\n

Part 3: Production Hardening<\/h3>\n

Spot Instance Handling With Checkpointing<\/h3>\n

# spark_job.py (runs on EMR)
\nfrom pyspark.sql import SparkSession
\nimport boto3
\nimport sys<\/p>\n

spark = SparkSession.builder
\n.appName(\u201cDataPipeline\u201d)
\n.config(\u201cspark.sql.adaptive.enabled\u201d, \u201ctrue\u201d)
\n.config(\u201cspark.checkpoint.dir\u201d, \u201cs3:\/\/emr-checkpoints\/job\/\u201d)
\n.getOrCreate()<\/p>\n

def main():
\n# Check for existing checkpoint
\ncheckpoint_path = \u201cs3:\/\/emr-checkpoints\/job\/progress\/\u201d
\ns3 = boto3.client(\u2018s3\u2032)<\/p>\n

try:
\ns3.head_object(Bucket=\u2019emr-checkpoints\u2019, Key=\u2019job\/progress\/_SUCCESS\u2019)
\nprint(\u201cResuming from checkpoint\u2026\u201d)
\ndf = spark.read.parquet(checkpoint_path)
\nexcept:
\nprint(\u201cStarting fresh\u2026\u201d)
\ndf = spark.read.parquet(\u201cs3:\/\/data-bucket\/raw\/\u201d)<\/p>\n

# Process with periodic checkpointing
\ndf = df.filter(df[\u2018date\u2019] >= \u20182024-01-01\u2019)
\ndf.write.mode(\u2018overwrite\u2019).parquet(checkpoint_path)<\/p>\n

# Final output
\ndf.write.mode(\u2018overwrite\u2019).parquet(\u201cs3:\/\/data-bucket\/processed\/\u201d)<\/p>\n

# Mark completion
\ns3.put_object(Bucket=\u2019emr-checkpoints\u2019, Key=\u2019job\/progress\/_SUCCESS\u2019, Body=b\u201d)<\/p>\n

if __name__ == \u201c__main__\u201d:
\ntry:
\nmain()
\nexcept Exception as e:
\nprint(f\u201dJob failed: {e}\u201d)
\nsys.exit(1)<\/p>\n

Cost Monitoring<\/h3>\n

# Track spot vs on-demand costs per job
\nemr = boto3.client(\u2019emr\u2019)
\npricing = boto3.client(\u2018pricing\u2019)<\/p>\n

def get_job_cost(cluster_id):
\n\u201c\u201d\u201dCalculate actual cost for EMR job including spot savings.\u201d\u201d\u201d<\/p>\n

instances = emr.list_instances(ClusterId=cluster_id)
\ntotal_cost = 0<\/p>\n

for inst in instances[\u2018Instances\u2019]:
\ninstance_type = inst[\u2018InstanceType\u2019]
\nmarket = inst[\u2018Market\u2019] # ON_DEMAND or SPOT
\nhours = (inst[\u2018Status\u2019][\u2018Timeline\u2019][\u2018EndDateTime\u2019] \u2013
\ninst[\u2018Status\u2019][\u2018Timeline\u2019][\u2018CreationDateTime\u2019]).seconds \/ 3600<\/p>\n

# Get pricing (simplified\u2014use CUR for actuals)
\nif market == \u2018SPOT\u2019:
\nrate = get_spot_price(instance_type)
\nelse:
\nrate = get_ondemand_price(instance_type)<\/p>\n

total_cost += rate * hours<\/p>\n

return total_cost<\/p>\n

Operational Impact to Measure<\/h3>\n

The value of this pattern should be measured with operational metrics rather than treated as an architecture preference. Useful measures include:<\/p>\n

When Custom AMIs Don\u2019t Make Sense<\/strong><\/p>\n