AWS EMR for MCP Servers: Cluster Lifecycle, Serverless Spark, S3A Pattern, and EMR Studio

Pattern 1: EMR cluster lifecycle — RunJobFlowCommand, state machine polling, and instance fleet Spot diversification

When an MCP server tool needs to run batch Spark, Hive, or Hadoop workloads, EMR cluster lifecycle management is the starting point. The core model is: launch a cluster with RunJobFlowCommand, poll through the full state machine until the cluster reaches WAITING (long-running) or TERMINATED (auto-terminate), and submit Spark jobs as EMR steps that execute sequentially on the cluster.

Installing the SDK and creating the client

EMR uses a dedicated package separate from the core AWS SDK. Install it alongside the S3 client if your MCP server retrieves step logs from S3.

npm install @aws-sdk/client-emr @aws-sdk/client-s3

Construct the client once at module initialization — it is stateless and safe to share across concurrent tool calls. EMR is a regional service; specify the region where your data lives to minimize cross-region data transfer costs for Spark jobs reading from S3.

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

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

The seven-state cluster state machine

An EMR cluster moves through up to seven states between launch and termination. Your MCP tool must handle the full sequence before reporting success or failure — returning the cluster ID as a success indicator before the cluster reaches WAITING is the most common EMR bug in MCP servers.

The StateChangeReason sub-object on cluster status carries the failure code: STEP_FAILURE (a Spark step exited non-zero), BOOTSTRAP_FAILURE (a bootstrap action failed), or INSTANCE_FAILURE (EC2 capacity issue or Spot interruption). Log both .Code and .Message on TERMINATED_WITH_ERRORS — the message often contains the step ID or instance ID that failed, which you need to look up the S3 log path.

Launching a cluster with RunJobFlowCommand

The two launch-time decisions that determine everything else are cluster persistence (auto-terminate or long-running) and IAM roles (both are required — omitting either causes the API to return InvalidRequestException with a confusing message).

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

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

// Launch a transient cluster that auto-terminates when all steps complete
export async function launchTransientCluster(params: {
  name: string;
  s3LogBucket: string;
  subnetId: string;
}): Promise<string> {
  const response = await emr.send(new RunJobFlowCommand({
    Name: params.name,
    ReleaseLabel: 'emr-7.2.0',          // Spark 3.5 + Hadoop 3.3
    Applications: [{ Name: 'Spark' }, { Name: 'Hadoop' }],
    LogUri: `s3://${params.s3LogBucket}/emr-logs/`,
    Instances: {
      MasterInstanceType: 'm5.xlarge',
      SlaveInstanceType: 'm5.xlarge',
      InstanceCount: 3,                   // 1 master + 2 core nodes
      Ec2SubnetId: params.subnetId,
      KeepJobFlowAliveWhenNoSteps: false, // auto-terminate on completion
      TerminationProtected: false,
    },
    // Both IAM roles required — created by 'aws emr create-default-roles'
    JobFlowRole: 'EMR_EC2_DefaultRole',   // EC2 instance profile
    ServiceRole: 'EMR_DefaultRole',       // EMR service role
    // Belt-and-suspenders: auto-terminate if idle for 1h (catches stuck clusters)
    AutoTerminationPolicy: { IdleTimeout: 3600 },
    Tags: [{ Key: 'ManagedBy', Value: 'alivemcp' }],
  }));

  return response.JobFlowId!;  // 'j-XXXXXXXXXX'
}

// Poll cluster to completion — call after RunJobFlowCommand
export async function waitForCluster(
  clusterId: string,
  timeoutMs = 20 * 60 * 1000,  // 20 min is usually sufficient for 3-node cluster
): Promise<ClusterState> {
  const deadline = Date.now() + timeoutMs;
  const TERMINAL = new Set<ClusterState>(['TERMINATED', 'TERMINATED_WITH_ERRORS']);
  const READY = new Set<ClusterState>(['WAITING', 'RUNNING']);
  let lastState: ClusterState = 'STARTING';

  while (Date.now() < deadline) {
    const r = await emr.send(new DescribeClusterCommand({ ClusterId: clusterId }));
    const cluster = r.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.has(state)) {
      if (state === 'TERMINATED_WITH_ERRORS') {
        throw new Error(`Cluster ${clusterId} failed: ${reason?.Code} — ${reason?.Message}`);
      }
      return state;
    }
    if (READY.has(state)) return state;

    await new Promise(r => setTimeout(r, 15_000));  // 15s — EMR is eventually consistent
  }
  throw new Error(`Cluster ${clusterId} timed out after ${timeoutMs}ms`);
}

Submitting steps and polling step state

AddJobFlowStepsCommand queues steps on a running or waiting cluster. Steps execute sequentially — a step does not start until the previous one finishes. Use CANCEL_AND_WAIT as the default ActionOnFailure for automated pipelines: it stops the step queue but leaves the cluster running so you can inspect logs and add remediated steps without terminating the cluster.

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

export async function submitSparkSubmitStep(
  clusterId: string,
  name: string,
  scriptS3Path: string,
  sparkArgs: string[],
): Promise<string> {
  const r = await emr.send(new AddJobFlowStepsCommand({
    JobFlowId: clusterId,
    Steps: [{
      Name: name,
      ActionOnFailure: 'CANCEL_AND_WAIT',
      HadoopJarStep: {
        Jar: 'command-runner.jar',         // EMR pre-installs this on every cluster
        Args: ['spark-submit', ...sparkArgs],  // first arg = command on PATH
      },
    }],
  }));
  return r.StepIds![0];  // 's-XXXXXXXXXX'
}

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

  while (Date.now() < deadline) {
    const r = await emr.send(new DescribeStepCommand({ ClusterId: clusterId, StepId: stepId }));
    const step = r.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`);
}

Instance fleet Spot diversification

Instance fleets are the recommended EMR capacity model for cost-sensitive batch jobs. Rather than specifying a single instance type at a fixed count, you specify a vCPU target and up to 5 instance type configurations — EMR fills the target using whatever Spot capacity is available across those types, dramatically reducing Spot interruption risk compared to single-type instance groups.

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

export async function launchFleetCluster(subnetIds: string[]): Promise<string> {
  const r = await emr.send(new RunJobFlowCommand({
    Name: 'spark-fleet-cluster',
    ReleaseLabel: 'emr-7.2.0',
    Applications: [{ Name: 'Spark' }],
    LogUri: 's3://my-bucket/emr-logs/',
    // InstanceFleets replaces Instances — cannot mix the two
    InstanceFleets: [
      {
        Name: 'Master',
        InstanceFleetType: 'MASTER',
        TargetOnDemandCapacity: 1,
        InstanceTypeConfigs: [{ InstanceType: 'm5.xlarge' }],
      },
      {
        Name: 'Core',
        InstanceFleetType: 'CORE',
        TargetSpotCapacity: 8,       // 8 Spot units (each m5.xlarge = 4 units = 4 vCPUs)
        TargetOnDemandCapacity: 2,   // On-Demand fallback for reliability
        InstanceTypeConfigs: [
          { InstanceType: 'm5.xlarge',  WeightedCapacity: 4 },
          { InstanceType: 'm6i.xlarge', WeightedCapacity: 4 },
          { InstanceType: 'm5.2xlarge', WeightedCapacity: 8 },  // counts as 2 units
        ],
        LaunchSpecifications: {
          SpotSpecification: {
            TimeoutDurationMinutes: 10,
            TimeoutAction: 'SWITCH_TO_ON_DEMAND',  // fallback if Spot unavailable at launch
          },
        },
      },
    ],
    Ec2SubnetIds: subnetIds,  // multiple AZs → more Spot capacity pools to draw from
    JobFlowRole: 'EMR_EC2_DefaultRole',
    ServiceRole: 'EMR_DefaultRole',
    KeepJobFlowAliveWhenNoSteps: false,
  }));

  return r.JobFlowId!;
}

Termination protection and S3 log retrieval

Termination protection prevents TerminateJobFlowsCommand from succeeding without first disabling it. Enable it on long-running clusters where losing the cluster mid-job would require hours of restart. For step failure diagnosis, read the syslog from S3 — it is the first place to check for any STEP_FAILURE.

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

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

// Enable protection before starting long-running work
export async function enableTerminationProtection(clusterId: string): Promise<void> {
  await emr.send(new SetTerminationProtectionCommand({
    JobFlowIds: [clusterId],
    TerminationProtected: true,
  }));
}

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

// Read failed step syslog — available in S3 ~5min after step completes
export async function getStepSyslog(
  logBucket: string, clusterId: string, stepId: string,
): Promise<string> {
  const r = await s3.send(new GetObjectCommand({
    Bucket: logBucket,
    Key: `emr-logs/${clusterId}/steps/${stepId}/syslog`,
  }));
  return r.Body?.transformToString() ?? '';
}

S3 log layout under LogUri: <cluster-id>/steps/<step-id>/syslog (main step log), stderr (Spark driver stderr — spark-submit progress goes here), stdout. Bootstrap failures: <cluster-id>/node/<instance-id>/bootstrap-actions/1/stderr (numbered by bootstrap action index). During step execution, logs are only on the master node — S3 aggregation happens at step completion, so reading logs during execution returns empty objects.

Pattern 2: EMR Serverless — CreateApplicationCommand, StartJobRunCommand, and pre-initialized capacity

When an MCP server tool needs to run Spark or Hive jobs without managing EC2 clusters, EMR Serverless eliminates the cluster lifecycle entirely. The model is: create an application once per workload type (the billing and configuration boundary), then submit job runs against it. Workers are provisioned on-demand per job run and released on completion.

Installing the SDK and creating the client

EMR Serverless is a separate package and service from EMR on EC2. The APIs and SDK client are distinct — do not confuse EMRServerlessClient with EMRClient.

npm install @aws-sdk/client-emr-serverless

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

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

Application lifecycle and auto-start

An application is a reusable container for job runs. Create it once per workload type and reuse it across many job submissions. With autoStartConfiguration.enabled: true (the default), submitting a job against a stopped application starts it automatically — you do not need to call StartApplicationCommand explicitly.

import {
  EMRServerlessClient,
  CreateApplicationCommand,
  StartApplicationCommand,
  GetApplicationCommand,
  type ApplicationState,
} from '@aws-sdk/client-emr-serverless';

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

export async function createSparkApplication(params: {
  name: string;
  releaseLabel?: string;
}): Promise<string> {
  const r = await emrServerless.send(new CreateApplicationCommand({
    name: params.name,
    type: 'SPARK',
    releaseLabel: params.releaseLabel ?? 'emr-7.2.0',
    workerTypeSpecifications: {
      Driver: {
        workerConfiguration: { cpu: '4 vCPU', memory: '16 GB', disk: '20 GB' },
      },
      Executor: {
        workerConfiguration: { cpu: '4 vCPU', memory: '16 GB', disk: '20 GB' },
      },
    },
    autoStopConfiguration: { enabled: true, idleTimeoutMinutes: 15 },
    autoStartConfiguration: { enabled: true },
    tags: { ManagedBy: 'alivemcp' },
  }));
  return r.applicationId!;
}

Submitting a job run with StartJobRunCommand

The critical parameters on StartJobRunCommand are executionRoleArn (must have S3 read/write, Glue, and CloudWatch Logs permissions for the job), jobDriver.sparkSubmit.entryPoint (S3 path to your Python script or JAR), and configurationOverrides.monitoringConfiguration (required to get logs after job completion — without this, the only debugging channel is the console).

import {
  StartJobRunCommand,
  GetJobRunCommand,
  CancelJobRunCommand,
  type JobRunState,
} from '@aws-sdk/client-emr-serverless';

export async function submitSparkJob(params: {
  applicationId: string;
  executionRoleArn: string;
  name: string;
  entryPoint: string;         // s3://bucket/scripts/my-job.py
  entryPointArgs?: string[];
  sparkSubmitParams?: string; // '--conf spark.executor.memory=8g'
  logBucket?: string;
  executionTimeoutMinutes?: number;
}): Promise<string> {
  const r = await emrServerless.send(new StartJobRunCommand({
    applicationId: params.applicationId,
    name: params.name,
    executionRoleArn: params.executionRoleArn,
    executionTimeoutMinutes: params.executionTimeoutMinutes ?? 60,
    jobDriver: {
      sparkSubmit: {
        entryPoint: params.entryPoint,
        entryPointArguments: params.entryPointArgs ?? [],
        sparkSubmitParameters: params.sparkSubmitParams ?? '',
      },
    },
    configurationOverrides: params.logBucket ? {
      monitoringConfiguration: {
        s3MonitoringConfiguration: {
          logUri: `s3://${params.logBucket}/emr-serverless-logs/`,
        },
        managedPersistenceMonitoringConfiguration: { enabled: true },
      },
    } : undefined,
    tags: { ManagedBy: 'alivemcp' },
  }));
  return r.jobRunId!;
}

export async function waitForJobRun(
  applicationId: string,
  jobRunId: string,
  timeoutMs = 60 * 60 * 1000,
): Promise<JobRunState> {
  const deadline = Date.now() + timeoutMs;
  const TERMINAL: JobRunState[] = ['SUCCESS', 'FAILED', 'CANCELLED'];

  while (Date.now() < deadline) {
    const r = await emrServerless.send(new GetJobRunCommand({ applicationId, jobRunId }));
    const jobRun = r.jobRun!;
    const state = jobRun.state as JobRunState;

    if (TERMINAL.includes(state)) {
      if (state !== 'SUCCESS') {
        throw new Error(`Job run ${jobRunId} ${state}: ${jobRun.stateDetails}`);
      }
      return state;
    }
    await new Promise(res => setTimeout(res, 15_000));
  }

  // Cancel before giving up — prevents lingering charges
  await emrServerless.send(new CancelJobRunCommand({ applicationId, jobRunId }));
  throw new Error(`Job run ${jobRunId} timed out`);
}

Pre-initialized capacity for warm starts

The default EMR Serverless behavior starts workers from scratch per job run, taking 60–120 seconds for the first job. Pre-initialized capacity keeps a warm worker pool running 24/7. Use it when job latency matters more than idle cost — interactive analytics, SLA-bound pipelines, and tools where the user is waiting on the response.

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

export async function createWarmSparkApplication(name: string): Promise<string> {
  const r = await emrServerless.send(new CreateApplicationCommand({
    name,
    type: 'SPARK',
    releaseLabel: 'emr-7.2.0',
    // Warm pool: 2 drivers + 10 executors always running
    initialCapacity: {
      Driver: {
        workerCount: 2,
        workerConfiguration: { cpu: '4 vCPU', memory: '16 GB', disk: '20 GB' },
      },
      Executor: {
        workerCount: 10,
        workerConfiguration: { cpu: '4 vCPU', memory: '16 GB', disk: '20 GB' },
      },
    },
    // Billing guardrail — prevents runaway concurrency
    maximumCapacity: { cpu: '400 vCPU', memory: '3000 GB', disk: '20000 GB' },
    autoStopConfiguration: { enabled: false },  // must be off to keep warm workers running
    autoStartConfiguration: { enabled: true },
  }));
  return r.applicationId!;
}

Pre-initialized capacity cost breakdown at 10 Executor workers + 2 Driver workers (all 4 vCPU/16 GB) running 24/7: 12 workers × 4 vCPU × $0.052/vCPU-hr + 12 × 16 GB × $0.0057/GB-hr = $2.50/hr ($1,800/mo). Evaluate against the alternative: a persistent m5.xlarge (4 vCPU/16 GB) EC2 cluster of equivalent size at ~$0.192/hr × 12 = $2.30/hr On-Demand or $0.69/hr Spot. For high-frequency interactive workloads, a persistent EC2 cluster may be cheaper; for infrequent bursts where idle cost matters, the serverless default (no pre-init, $0 idle) wins.

Worker sizing guide

EMR Serverless enforces a fixed memory-to-CPU ratio of 4 GB per vCPU. Valid configurations scale in doublings of CPU:

OOM patterns: driver OOM (java.lang.OutOfMemoryError: GC overhead limit exceeded) during result collection → increase driver memory first. Executor OOM during shuffle or broadcast join → increase executor memory. Disk OOM (java.io.IOException: No space left on device) on large aggregations → increase disk from 20 GB to 100–200 GB. Disk is consumed by shuffle spill — wide aggregations on large datasets frequently exhaust the default 20 GB disk before the memory limit is hit.

Pattern 3: Spark job patterns — command-runner.jar, S3A magic committer, Glue metastore, and dynamic allocation

Once the cluster or Serverless application is running, Spark job configuration determines whether jobs produce correct output, use cluster resources efficiently, and preserve table definitions across cluster restarts. Three configuration decisions have the highest impact: the S3 output committer (data correctness on failure), the Hive metastore (table persistence), and dynamic allocation (cost during idle periods).

Spark-submit via command-runner.jar

EMR installs several command wrappers in /usr/bin on every cluster master node: spark-submit, hadoop, hive, pig, and flink. The command-runner.jar step JAR executes whichever command you pass as the first argument — this is the standard mechanism for running Spark jobs without bundling a custom launcher JAR. Always use --deploy-mode cluster so the driver runs on a core node rather than the master — this prevents the master node from becoming a CPU/memory bottleneck on large jobs.

import { EMRClient, AddJobFlowStepsCommand } from '@aws-sdk/client-emr';

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

export async function submitPySparkJob(params: {
  clusterId: string;
  name: string;
  scriptS3Path: string;      // s3://bucket/scripts/my-job.py
  scriptArgs?: string[];
  extraConf?: Record<string, string>;
}): Promise<string> {
  const sparkConf = {
    // S3A magic committer — safe multi-part writes on S3
    'spark.hadoop.fs.s3a.committer.name': 'magic',
    'spark.hadoop.fs.s3a.committer.magic.enabled': 'true',
    'spark.sql.sources.commitProtocolClass':
      'org.apache.spark.internal.io.cloud.PathOutputCommitProtocol',
    'spark.sql.parquet.output.committer.class':
      'org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter',
    'spark.hadoop.fs.s3a.committer.abort.pending.uploads': 'true',
    // Glue Data Catalog as Hive metastore — tables survive cluster termination
    'spark.hadoop.hive.metastore.client.factory.class':
      'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory',
    // Dynamic allocation without external shuffle service
    'spark.dynamicAllocation.enabled': 'true',
    'spark.dynamicAllocation.shuffleTracking.enabled': 'true',
    'spark.dynamicAllocation.minExecutors': '1',
    'spark.dynamicAllocation.maxExecutors': '20',
    'spark.dynamicAllocation.initialExecutors': '2',
    'spark.dynamicAllocation.executorIdleTimeout': '60s',
    'spark.dynamicAllocation.cachedExecutorIdleTimeout': '1800s',
    ...params.extraConf,
  };

  const confArgs = Object.entries(sparkConf)
    .flatMap(([k, v]) => ['--conf', `${k}=${v}`]);

  const r = await emr.send(new AddJobFlowStepsCommand({
    JobFlowId: params.clusterId,
    Steps: [{
      Name: params.name,
      ActionOnFailure: 'CANCEL_AND_WAIT',
      HadoopJarStep: {
        Jar: 'command-runner.jar',
        Args: [
          'spark-submit',
          '--deploy-mode', 'cluster',
          '--master', 'yarn',
          ...confArgs,
          params.scriptS3Path,
          ...(params.scriptArgs ?? []),
        ],
      },
    }],
  }));

  return r.StepIds![0];
}

S3A committer comparison

The choice of S3A output committer is the most consequential Spark configuration for correctness. The default file committer performs a copy+delete rename, which is not atomic on S3 and leaves corrupt partial files if the job is interrupted. For all EMR and EMR Serverless jobs writing to S3, use the magic committer.

The magic committer requires S3 Multipart Upload (enabled by default on all S3 buckets) and Hadoop 3.3.1+ (included in EMR 6.x and 7.x). The only exception: S3 Express One Zone directory buckets do not support the magic committer — use directory there.

Glue Data Catalog as the Hive metastore

By default, Spark on EMR uses a local Derby-based Hive metastore on the master node's EBS volume. When the cluster terminates, all table definitions are lost. The Glue Data Catalog metastore stores table definitions in Glue — they persist across cluster restarts and are shared with Athena, Glue ETL, and Lake Formation.

// Add to spark-submit --conf args on any job that creates or reads tables
const glueMetastoreConf = {
  'spark.hadoop.hive.metastore.client.factory.class':
    'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory',
};

// In PySpark — creates a table in Glue that survives cluster termination
// spark.sql("CREATE TABLE IF NOT EXISTS analytics.events USING PARQUET LOCATION 's3a://bucket/events/'")
// spark.sql("SELECT * FROM analytics.events WHERE dt = '2026-11-07'")

The EMR_EC2_DefaultRole instance profile must include permissions for glue:GetDatabase, glue:GetTable, glue:CreateTable, and glue:UpdateTable. These are included in the AmazonElasticMapReduceforEC2Role managed policy attached to the default instance profile — if you created a custom instance profile, add an inline policy for these Glue Catalog permissions.

Dynamic allocation and partition overwrite mode

Dynamic allocation lets YARN release idle executors back to the cluster between job stages, reducing cost on long-running clusters that sit at WAITING between jobs. With shuffleTracking.enabled: true, Spark tracks which executors hold shuffle data and won't release them prematurely — this removes the need for a Spark external shuffle service, which is not available on EMR by default.

// Dynamic allocation configuration — apply to all jobs on long-running clusters
const dynamicAllocationConf = {
  'spark.dynamicAllocation.enabled': 'true',
  'spark.dynamicAllocation.shuffleTracking.enabled': 'true',   // no external shuffle service needed
  'spark.dynamicAllocation.shuffleTracking.timeout': '300s',
  'spark.dynamicAllocation.minExecutors': '1',
  'spark.dynamicAllocation.maxExecutors': '50',       // set to match cluster core capacity
  'spark.dynamicAllocation.initialExecutors': '4',
  'spark.dynamicAllocation.executorIdleTimeout': '60s',          // release general idle executors
  'spark.dynamicAllocation.cachedExecutorIdleTimeout': '1800s',  // keep cached-data executors longer
};

Partition overwrite mode: when writing partitioned output with mode('overwrite'), spark.sql.sources.partitionOverwriteMode=dynamic (the default) overwrites only the partitions present in the current output dataframe — suitable for incremental date-partitioned writes where you only want to replace today's partition. static mode deletes the entire output path before writing — use it only when you intend a full table replacement.

Cancelling runaway steps

Use CancelStepsCommand to abort a step that is taking too long or consuming unexpected resources. SEND_INTERRUPT sends SIGTERM and waits 10 seconds before SIGKILL — allows graceful Spark shutdown handlers to flush output. Use TERMINATE_PROCESS for immediate SIGKILL on a hung driver.

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

export async function cancelStep(clusterId: string, stepId: string): Promise<void> {
  const r = await emr.send(new CancelStepsCommand({
    ClusterId: clusterId,
    StepIds: [stepId],
    StepCancellationOption: 'SEND_INTERRUPT',
    // Use 'TERMINATE_PROCESS' if the driver is hung in a non-interruptible native call
  }));

  const result = r.CancelStepsInfoList?.[0];
  if (result?.Status === 'FAILED') {
    throw new Error(`Failed to cancel step ${stepId}: ${result.Reason}`);
  }
}

After cancellation the step transitions to CANCELLED (if it was pending) or INTERRUPTED (if SIGKILL was required while running). With ActionOnFailure: 'CANCEL_AND_WAIT', the cluster stays at WAITING after cancellation — you can submit a corrected step or terminate the cluster explicitly.

Pattern 4: EMR Studio — CreateStudio, CreateEditor, SparkMagic notebooks, and Git integration

When an MCP server tool needs to provision interactive notebook environments for data engineering or ML experimentation, EMR Studio provides a managed JupyterLab environment that connects to EMR clusters or Serverless applications for Spark execution. The workspace model separates the JupyterLab server (the editor — where notebooks are stored) from the compute (the cluster or application — where Spark code actually runs).

Three-layer architecture and billing model

EMR Studio has three distinct layers with separate billing:

The workspace is a lightweight JupyterLab server — it stores notebooks in S3 and manages kernel connections, but all Spark code executes on the attached cluster or Serverless application via the Apache Livy REST API. The network path between the workspace and the cluster goes through port 18888; the security groups must be configured correctly or SparkMagic fails to establish the Livy connection silently.

Creating a studio with correct security group wiring

The WorkspaceSecurityGroupId and EngineSecurityGroupId in CreateStudioCommand control the Livy API connectivity. The workspace SG must allow HTTPS outbound to the engine SG on TCP port 18888; the engine SG must allow inbound TCP 18888 from the workspace SG. If either rule is missing, the PySpark kernel connects and immediately shows a connection error.

import {
  EMRClient,
  CreateStudioCommand,
  ListStudiosCommand,
} from '@aws-sdk/client-emr';

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

export async function createStudio(params: {
  name: string;
  vpcId: string;
  subnetIds: string[];
  workspaceSecurityGroupId: string;  // must have TCP outbound to engine SG on 18888
  engineSecurityGroupId: string;     // must have TCP inbound from workspace SG on 18888
  defaultS3Location: string;         // s3://bucket/emr-studio/ — notebook storage
  serviceRoleArn: string;
  idcInstanceArn?: string;           // provide for SSO auth; omit for IAM auth
}): Promise<string> {
  const r = await emr.send(new CreateStudioCommand({
    Name: params.name,
    AuthMode: params.idcInstanceArn ? 'SSO' : 'IAM',
    VpcId: params.vpcId,
    SubnetIds: params.subnetIds,
    WorkspaceSecurityGroupId: params.workspaceSecurityGroupId,
    EngineSecurityGroupId: params.engineSecurityGroupId,
    DefaultS3Location: params.defaultS3Location,
    ServiceRole: params.serviceRoleArn,
    IdcInstanceArn: params.idcInstanceArn,
    Tags: [{ Key: 'ManagedBy', Value: 'alivemcp' }],
  }));
  return r.StudioId!;
}

export async function listStudios(): Promise<Array<{ id: string; name: string; url: string }>> {
  const studios = [];
  let marker: string | undefined;
  do {
    const r = await emr.send(new ListStudiosCommand({ Marker: marker }));
    for (const s of r.Studios ?? []) {
      studios.push({ id: s.StudioId!, name: s.Name!, url: s.Url! });
    }
    marker = r.Marker;
  } while (marker);
  return studios;
}

Creating, starting, and managing workspace editors

Workspaces persist notebooks to S3 at DefaultS3Location/studio-id/workspace-id/*.ipynb. Use StopEditorCommand (not DeleteEditorCommand) at end of session to end billing while preserving the notebooks. Only delete workspaces when the work is permanently complete.

import {
  CreateEditorCommand,
  StartEditorCommand,
  StopEditorCommand,
  DeleteEditorCommand,
  GetEditorCommand,
  AttachEditorCommand,
  DetachEditorCommand,
  type EditorState,
} from '@aws-sdk/client-emr';

export async function createWorkspace(params: {
  studioId: string;
  name: string;
  subnetId: string;
}): Promise<string> {
  const r = await emr.send(new CreateEditorCommand({
    Name: params.name,
    StudioId: params.studioId,
    SubnetId: params.subnetId,
  }));
  return r.EditorId!;
}

export async function startWorkspace(
  editorId: string,
  clusterId?: string,      // attach to cluster on start
  applicationId?: string,  // OR attach to Serverless app on start
): Promise<void> {
  await emr.send(new StartEditorCommand({ EditorId: editorId, ClusterId: clusterId, ApplicationId: applicationId }));

  const deadline = Date.now() + 5 * 60 * 1000;
  while (Date.now() < deadline) {
    const r = await emr.send(new GetEditorCommand({ EditorId: editorId }));
    const state = r.Editor?.Status?.State as EditorState;
    if (state === 'READY') return;
    if (state === 'FAILED') {
      throw new Error(`Workspace ${editorId} failed: ${r.Editor?.Status?.StateChangeReason?.Message}`);
    }
    await new Promise(res => setTimeout(res, 10_000));
  }
  throw new Error(`Workspace ${editorId} did not reach READY within 5 minutes`);
}

// Stop to end billing while preserving S3 notebooks
export async function stopWorkspace(editorId: string): Promise<void> {
  await emr.send(new StopEditorCommand({ EditorId: editorId }));
}

// Delete permanently — does NOT delete S3 objects
export async function deleteWorkspace(editorId: string): Promise<void> {
  await emr.send(new DeleteEditorCommand({ EditorId: editorId }));
}

// Reattach workspace to a different cluster (detach first if already attached)
export async function attachToCluster(editorId: string, clusterId: string): Promise<void> {
  await emr.send(new AttachEditorCommand({ EditorId: editorId, ClusterId: clusterId }));
}

export async function detachFromCompute(editorId: string): Promise<void> {
  await emr.send(new DetachEditorCommand({ EditorId: editorId }));
}

SparkMagic kernels and Livy session lifecycle

EMR Studio uses SparkMagic — a set of Jupyter kernels that communicate with the attached cluster via the Apache Livy REST API on port 18888. Each kernel variant runs code in a different language:

Switching kernels within the same workspace starts a new Livy session — all cached DataFrames and defined variables from the previous session are lost. For MCP tools that programmatically create notebooks or restart sessions, budget ~30 seconds for the new Livy session to initialize before the first cell executes successfully. The %%spark, %%pyspark, and %%sparkr magic commands route cell execution to the cluster; cells without a magic command run locally in the workspace instance (Python cells outside SparkMagic context).

Git repository integration

Linking a Git repository to a workspace enables notebook version control — changes are committed to the linked remote. Credentials are stored in AWS Secrets Manager, not in the workspace configuration.

import {
  CreateRepositoryCommand,
  LinkRepositoryCommand,
  ListRepositoriesCommand,
} from '@aws-sdk/client-emr';

export async function createGitRepository(params: {
  name: string;
  gitUrl: string;             // https://github.com/org/repo
  gitBranch: string;
  credentialSecretArn: string; // Secrets Manager ARN — {"username": "git", "password": "token"}
}): Promise<string> {
  const r = await emr.send(new CreateRepositoryCommand({
    Name: params.name,
    GitRepository: {
      Url: params.gitUrl,
      Branch: params.gitBranch,
      RepositoryAuthConfiguration: { SecretArn: params.credentialSecretArn },
    },
  }));
  return r.Repository?.Id!;
}

export async function linkRepositoryToWorkspace(
  editorId: string,
  repositoryId: string,
): Promise<void> {
  await emr.send(new LinkRepositoryCommand({ EditorId: editorId, RepositoryId: repositoryId }));
}

The Secrets Manager secret must be in the same AWS region as the EMR Studio. For GitHub, use a fine-grained personal access token with repository read/write scope, stored as {"username": "git", "password": "github_pat_xxx"}. For GitLab, store a project access token with read_repository and write_repository scopes. The secret ARN is stored in the repository record — if you rotate the token, update the secret value in Secrets Manager and the new credentials are used on the next workspace-to-cluster sync without re-linking the repository.

Related pages

• EMR cluster lifecycle — RunJobFlow, cluster state machine, AddJobFlowSteps, instance fleet — full EC2-backed cluster management reference
• EMR Serverless — CreateApplication, StartJobRun, pre-initialized capacity, cost model — serverless Spark without cluster management
• EMR Spark patterns — spark-submit, S3A magic committer, Glue metastore, dynamic allocation — Spark configuration layer for all EMR deployment types
• EMR Studio — CreateStudio, CreateEditor, SparkMagic, Git integration — interactive notebook workspace management
• AWS Glue ETL jobs for MCP servers
• AWS Lake Formation for MCP servers
• Athena workgroups for MCP servers
• AWS Lake Formation for MCP Servers: Resource Registration, Permission Grants, LF-Tag TBAC, and Governed Table Transactions