MCP Server AWS EMR — RunJobFlow, cluster lifecycle, AddJobFlowSteps

Cluster state machine

An EMR cluster moves through these states in order. Your MCP tool must handle the full sequence before reporting success or failure.

The StateChangeReason sub-object on cluster status contains a Code (e.g. STEP_FAILURE, INSTANCE_FAILURE, BOOTSTRAP_FAILURE) and a human-readable Message. Always log both when the cluster reaches TERMINATED_WITH_ERRORS.

Launching a cluster with RunJobFlowCommand

RunJobFlowCommand creates the cluster and optionally submits the first batch of steps. For transient job clusters, include steps here and set KeepJobFlowAliveWhenNoSteps: false — the cluster auto-terminates when all steps finish, eliminating idle cost.

import {
  EMRClient,
  RunJobFlowCommand,
  type RunJobFlowCommandInput,
} from '@aws-sdk/client-emr';

const emr = new EMRClient({ region: process.env.AWS_REGION ?? 'us-east-1' });

export async function launchSparkCluster(params: {
  name: string;
  s3LogBucket: string;
  subnetId: string;
  releaseLabel?: string;
}): Promise<string> {
  const input: RunJobFlowCommandInput = {
    Name: params.name,
    ReleaseLabel: params.releaseLabel ?? 'emr-7.2.0',
    // Applications to install on the cluster
    Applications: [
      { Name: 'Spark' },
      { Name: 'Hadoop' },
    ],
    // Ship all logs to S3 for post-mortem debugging
    LogUri: `s3://${params.s3LogBucket}/emr-logs/`,
    // Instance configuration — master + 2 core nodes
    Instances: {
      MasterInstanceType: 'm5.xlarge',
      SlaveInstanceType: 'm5.xlarge',
      InstanceCount: 3,  // 1 master + 2 core
      Ec2SubnetId: params.subnetId,
      KeepJobFlowAliveWhenNoSteps: false,  // auto-terminate when done
      TerminationProtected: false,
    },
    // IAM roles — both required
    JobFlowRole: 'EMR_EC2_DefaultRole',      // EC2 instance profile
    ServiceRole: 'EMR_DefaultRole',           // EMR service role
    // Cost guardrail: auto-terminate removes idle billing risk
    AutoTerminationPolicy: {
      IdleTimeout: 3600,  // also terminate if idle for 1h (belt-and-suspenders)
    },
    Tags: [
      { Key: 'ManagedBy', Value: 'alivemcp' },
      { Key: 'ClusterName', Value: params.name },
    ],
  };

  const response = await emr.send(new RunJobFlowCommand(input));
  return response.JobFlowId!;  // cluster ID, e.g. 'j-XXXXXXXXXX'
}

The ReleaseLabel pinpoints the EMR release version, which determines the Spark, Hadoop, and Hive versions available. Use emr-7.2.0 for Spark 3.5 (current as of 2026). The JobFlowRole is an EC2 instance profile; the ServiceRole is assumed by the EMR control plane. Both must exist in your account — the defaults (EMR_EC2_DefaultRole and EMR_DefaultRole) are created when you use the EMR console for the first time, or you can create them with aws emr create-default-roles.

Polling cluster state

Use DescribeClusterCommand to poll until the cluster reaches a terminal or ready state. Avoid polling more than once every 15 seconds — EMR is an eventually-consistent system and rapid polling wastes API quota without returning fresher data.

import {
  DescribeClusterCommand,
  type ClusterState,
} from '@aws-sdk/client-emr';

const TERMINAL_STATES = new Set<ClusterState>([
  'TERMINATED',
  'TERMINATED_WITH_ERRORS',
]);
const READY_STATES = new Set<ClusterState>(['WAITING', 'RUNNING']);

export async function waitForCluster(
  clusterId: string,
  timeoutMs = 20 * 60 * 1000,  // 20 min default
): Promise<ClusterState> {
  const deadline = Date.now() + timeoutMs;
  let lastState: ClusterState = 'STARTING';

  while (Date.now() < deadline) {
    const response = await emr.send(
      new DescribeClusterCommand({ ClusterId: clusterId })
    );
    const cluster = response.Cluster!;
    const state = cluster.Status?.State as ClusterState;
    const reason = cluster.Status?.StateChangeReason;

    if (state !== lastState) {
      console.log(`Cluster ${clusterId}: ${lastState} → ${state}`, reason?.Message ?? '');
      lastState = state;
    }

    if (TERMINAL_STATES.has(state)) {
      if (state === 'TERMINATED_WITH_ERRORS') {
        throw new Error(
          `Cluster ${clusterId} failed: ${reason?.Code} — ${reason?.Message}`
        );
      }
      return state;
    }

    if (READY_STATES.has(state)) return state;

    await new Promise(r => setTimeout(r, 15_000));  // poll every 15s
  }

  throw new Error(`Cluster ${clusterId} did not reach ready/terminal state within ${timeoutMs}ms`);
}

Submitting Spark steps with AddJobFlowStepsCommand

AddJobFlowStepsCommand queues one or more steps on a running or waiting cluster. Steps execute sequentially by default — a step does not start until the previous one finishes. Use the ActionOnFailure field to control whether a step failure should terminate the cluster (TERMINATE_CLUSTER), cancel remaining steps (CANCEL_AND_WAIT), or continue to the next step (CONTINUE).

import {
  AddJobFlowStepsCommand,
  DescribeStepCommand,
  type StepState,
} from '@aws-sdk/client-emr';

export async function submitSparkStep(
  clusterId: string,
  params: {
    name: string;
    jarPath: string;        // s3://bucket/app.jar or 'command-runner.jar'
    args: string[];
    actionOnFailure?: 'TERMINATE_CLUSTER' | 'CANCEL_AND_WAIT' | 'CONTINUE';
  },
): Promise<string> {
  const response = await emr.send(new AddJobFlowStepsCommand({
    JobFlowId: clusterId,
    Steps: [
      {
        Name: params.name,
        ActionOnFailure: params.actionOnFailure ?? 'CANCEL_AND_WAIT',
        HadoopJarStep: {
          Jar: params.jarPath,
          Args: params.args,
        },
      },
    ],
  }));

  return response.StepIds![0];  // step ID, e.g. 's-XXXXXXXXXX'
}

// Convenience wrapper for Spark-submit steps via command-runner.jar
export async function submitSparkSubmitStep(
  clusterId: string,
  name: string,
  sparkArgs: string[],  // everything after 'spark-submit'
): Promise<string> {
  return submitSparkStep(clusterId, {
    name,
    jarPath: 'command-runner.jar',
    args: ['spark-submit', ...sparkArgs],
    actionOnFailure: 'CANCEL_AND_WAIT',
  });
}

// Poll a step to completion
export async function waitForStep(
  clusterId: string,
  stepId: string,
  timeoutMs = 60 * 60 * 1000,  // 1h
): Promise<StepState> {
  const deadline = Date.now() + timeoutMs;
  const TERMINAL: StepState[] = ['COMPLETED', 'FAILED', 'CANCELLED', 'INTERRUPTED'];

  while (Date.now() < deadline) {
    const response = await emr.send(
      new DescribeStepCommand({ ClusterId: clusterId, StepId: stepId })
    );
    const step = response.Step!;
    const state = step.Status?.State as StepState;

    if (TERMINAL.includes(state)) {
      if (state !== 'COMPLETED') {
        throw new Error(
          `Step ${stepId} ${state}: ${step.Status?.FailureDetails?.Message}`
        );
      }
      return state;
    }

    await new Promise(r => setTimeout(r, 15_000));
  }

  throw new Error(`Step ${stepId} timed out after ${timeoutMs}ms`);
}

When Jar is set to command-runner.jar, EMR interprets the first argument as a command available on the cluster PATH — including spark-submit, hadoop, hive, and pig. This is the standard pattern for running Spark jobs without bundling them into a custom JAR.

Instance fleet vs instance group

EMR offers two capacity management models. Instance fleets (recommended for cost optimization) allow mixing multiple instance families to satisfy a vCPU/memory target, enabling aggressive Spot usage. Instance groups use a fixed instance type at a fixed count, suitable when you need deterministic capacity.

import { RunJobFlowCommand } from '@aws-sdk/client-emr';

// Instance fleet example — master on-demand, core fleet mixing m5/m6i Spot
export async function launchFleetCluster(subnetIds: string[]): Promise<string> {
  const response = await emr.send(new RunJobFlowCommand({
    Name: 'spark-fleet-cluster',
    ReleaseLabel: 'emr-7.2.0',
    Applications: [{ Name: 'Spark' }],
    LogUri: 's3://my-bucket/emr-logs/',
    InstanceFleets: [
      {
        Name: 'Master',
        InstanceFleetType: 'MASTER',
        TargetOnDemandCapacity: 1,
        InstanceTypeConfigs: [{ InstanceType: 'm5.xlarge' }],
      },
      {
        Name: 'Core',
        InstanceFleetType: 'CORE',
        TargetSpotCapacity: 8,       // 8 Spot units (vCPU weight)
        TargetOnDemandCapacity: 2,   // 2 on-demand units as fallback
        InstanceTypeConfigs: [
          { InstanceType: 'm5.xlarge',  WeightedCapacity: 4 },  // 4vCPU = 1 unit
          { InstanceType: 'm6i.xlarge', WeightedCapacity: 4 },
          { InstanceType: 'm5.2xlarge', WeightedCapacity: 8 },
        ],
        LaunchSpecifications: {
          SpotSpecification: {
            TimeoutDurationMinutes: 10,
            TimeoutAction: 'SWITCH_TO_ON_DEMAND',  // fallback if Spot unavailable
          },
        },
      },
    ],
    Ec2KeyName: undefined,  // no SSH access needed for automated clusters
    Ec2SubnetIds: subnetIds,  // multiple AZs for better Spot availability
    JobFlowRole: 'EMR_EC2_DefaultRole',
    ServiceRole: 'EMR_DefaultRole',
    KeepJobFlowAliveWhenNoSteps: false,
  }));

  return response.JobFlowId!;
}

Termination protection

Termination protection prevents TerminateJobFlowsCommand from succeeding unless you first explicitly disable it. Use it on long-running clusters where losing the cluster mid-job would be costly. For transient clusters, leave protection off.

import {
  SetTerminationProtectionCommand,
  TerminateJobFlowsCommand,
} from '@aws-sdk/client-emr';

export async function enableTerminationProtection(clusterId: string): Promise<void> {
  await emr.send(new SetTerminationProtectionCommand({
    JobFlowIds: [clusterId],
    TerminationProtected: true,
  }));
}

export async function safelyTerminateCluster(clusterId: string): Promise<void> {
  // Must disable protection first
  await emr.send(new SetTerminationProtectionCommand({
    JobFlowIds: [clusterId],
    TerminationProtected: false,
  }));

  await emr.send(new TerminateJobFlowsCommand({
    JobFlowIds: [clusterId],
  }));
}

Termination protection also prevents the cluster from being terminated when a step fails with ActionOnFailure: TERMINATE_CLUSTER. That failure will instead put the cluster in WAITING state with a step state of FAILED, requiring you to decide the next action. For automated pipelines, prefer CANCEL_AND_WAIT and handle termination explicitly in your MCP tool.

Log aggregation to S3

Setting LogUri in RunJobFlowCommand causes EMR to push cluster logs, step logs, and application logs to S3 automatically. The layout under LogUri is:

s3://your-bucket/emr-logs/
  <cluster-id>/
    node/
      <ec2-instance-id>/
        applications/      # Spark/YARN application logs
        bootstrap-actions/ # Bootstrap action stdout/stderr
        daemons/           # ResourceManager, NodeManager logs
    steps/
      <step-id>/
        controller         # Step lifecycle log
        stderr             # Step stderr (spark-submit output goes here)
        stdout             # Step stdout
        syslog             # Main step log — start here for failures

For failed steps, the syslog file is the first place to check. For failed bootstrap actions, look in bootstrap-actions/1/stderr (numbered by bootstrap action index). To fetch logs programmatically from an MCP tool, use GetStepCommand to find the log path prefix, then use GetObjectCommand from S3 to read the files.

import { S3Client, GetObjectCommand } from '@aws-sdk/client-s3';

const s3 = new S3Client({ region: process.env.AWS_REGION ?? 'us-east-1' });

export async function getStepSyslog(
  logBucket: string,
  clusterId: string,
  stepId: string,
): Promise<string> {
  const key = `emr-logs/${clusterId}/steps/${stepId}/syslog`;
  const response = await s3.send(new GetObjectCommand({
    Bucket: logBucket,
    Key: key,
  }));

  return response.Body?.transformToString() ?? '';
}

Logs are typically available in S3 within 5 minutes of the step completing. During step execution they are only available on the master node — the S3 aggregation happens at step completion.

Error handling reference

Related pages

• EMR Serverless — CreateApplication, StartJobRun, pre-initialized capacity — run Spark/Hive without managing clusters
• EMR Spark job patterns — spark-submit, S3A, Glue metastore, dynamic allocation — Spark-specific configuration for EMR
• EMR Studio — CreateStudio, CreateEditor, SparkMagic notebooks — interactive notebook environment
• AWS Glue ETL jobs for MCP servers
• Athena workgroups for MCP servers