MCP Server EMR Spark — spark-submit, S3A, Glue metastore, dynamic allocation

Submitting spark-submit steps via command-runner.jar

EMR installs several command wrappers in /usr/bin on the 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 way to run Spark jobs without bundling them into a custom launcher JAR.

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[];       // positional args passed to the Python script
  extraConf?: Record<string, string>; // additional --conf key=value pairs
}): Promise<string> {
  // Build --conf flags
  const sparkConf = {
    // S3A magic committer for safe multi-part writes
    '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',
    // Glue Data Catalog as Hive metastore
    'spark.hadoop.hive.metastore.client.factory.class':
      'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory',
    // Dynamic executor allocation
    'spark.dynamicAllocation.enabled': 'true',
    'spark.dynamicAllocation.shuffleTracking.enabled': 'true',
    'spark.dynamicAllocation.minExecutors': '1',
    'spark.dynamicAllocation.maxExecutors': '20',
    'spark.dynamicAllocation.initialExecutors': '2',
    ...params.extraConf,
  };

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

  const response = 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',   // driver runs on a core node
            '--master', 'yarn',
            ...confArgs,
            params.scriptS3Path,
            ...scriptArgs,
          ],
        },
      },
    ],
  }));

  return response.StepIds![0];
}

// Scala/Java JAR submission — same pattern, different args
export async function submitJarJob(params: {
  clusterId: string;
  name: string;
  jarS3Path: string;
  mainClass: string;
  jarArgs?: string[];
}): Promise<string> {
  const response = 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',
            '--class', params.mainClass,
            params.jarS3Path,
            ...(params.jarArgs ?? []),
          ],
        },
      },
    ],
  }));

  return response.StepIds![0];
}

S3A protocol and the magic committer

Spark on EMR supports three S3 URL schemes: the legacy s3:// (deprecated, backed by EMRFS), s3n:// (deprecated), and s3a:// (current, backed by the Hadoop S3A connector). Always use s3a:// in new jobs. The critical S3A setting is the output committer, which controls how Spark writes output files and what happens on failure.

// Magic committer configuration block — add to all S3-output Spark jobs
const magicCommitterConf = {
  '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',
  // Abort incomplete multipart uploads older than 1 day
  'spark.hadoop.fs.s3a.committer.abort.pending.uploads': 'true',
};

The magic committer requires the destination S3 bucket to have Multipart Upload enabled (which is the default for all S3 buckets). It also requires Hadoop 3.3.1+ — EMR 6.x and 7.x include a compatible Hadoop version. The only scenario where magic committer is insufficient is writing to S3 Express One Zone (directory buckets) — use the directory committer there instead.

Using AWS Glue Data Catalog as the Hive metastore

By default, Spark on EMR uses a local Derby-based Hive metastore stored on the master node's EBS volume. When the cluster terminates, all table definitions are lost. Configure the Glue Data Catalog as the metastore instead — table definitions persist across cluster restarts and are shared with other services (Athena, Glue ETL, Lake Formation).

// Add to your spark-submit --conf args
const glueMetastoreConf = {
  'spark.hadoop.hive.metastore.client.factory.class':
    'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory',
  // Optional: target a specific Glue catalog account (for cross-account)
  // 'spark.hadoop.hive.metastore.glue.catalogid': '123456789012',
};

// In a PySpark script: CREATE TABLE DDL writes to Glue, survives cluster termination
// spark.sql("CREATE TABLE IF NOT EXISTS my_db.events USING PARQUET LOCATION 's3a://bucket/events/'")
// spark.sql("SELECT * FROM my_db.events WHERE date = '2026-11-06'")

For this to work the EC2 instance profile (EMR_EC2_DefaultRole) must have IAM permissions for glue:GetDatabase, glue:GetTable, glue:CreateTable, glue:UpdateTable, and related Glue Catalog operations. These are included in the AmazonElasticMapReduceforEC2Role managed policy attached to the default instance profile.

Dynamic executor allocation

YARN dynamic allocation lets Spark request executors as needed and release them when idle, rather than holding a fixed number for the entire job. On long-running clusters this significantly reduces cost — a cluster sitting at WAITING state between jobs releases all executors back to YARN.

const dynamicAllocationConf = {
  'spark.dynamicAllocation.enabled': 'true',
  // Shuffle tracking lets dynamic allocation work without an external shuffle service
  'spark.dynamicAllocation.shuffleTracking.enabled': 'true',
  'spark.dynamicAllocation.shuffleTracking.timeout': '300s',
  // Executor count bounds
  'spark.dynamicAllocation.minExecutors': '1',
  'spark.dynamicAllocation.maxExecutors': '50',
  'spark.dynamicAllocation.initialExecutors': '4',
  // Release idle executors after 60s
  'spark.dynamicAllocation.executorIdleTimeout': '60s',
  // Keep executors with cached data longer (avoid expensive re-caching)
  'spark.dynamicAllocation.cachedExecutorIdleTimeout': '1800s',
};

When shuffleTracking.enabled is true, Spark tracks which executors hold shuffle data referenced by pending map stages, so it won't release them prematurely. Without shuffle tracking, dynamic allocation requires a Spark external shuffle service — not needed on EMR when shuffle tracking is on.

Set maxExecutors to match your cluster's available capacity. Each m5.xlarge core node has 4 vCPU and 16 GB — with 1 executor per node and spark.executor.cores=4, a 10-node core fleet supports up to 10 executors. Setting maxExecutors higher than available capacity wastes time on pending executor requests that never fulfill.

Reading and writing Parquet, ORC, and CSV on S3

# In PySpark (submitted via spark-submit)
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('my-emr-job').getOrCreate()
sc = spark.sparkContext
sc.setLogLevel('WARN')  # reduce log noise in syslog

# Read Parquet partition from S3
input_df = spark.read.parquet('s3a://my-data-bucket/events/date=2026-11-06/')

# Apply transformations
result_df = (input_df
  .filter(input_df.event_type == 'page_view')
  .groupBy('user_id')
  .agg({'session_id': 'count'})
  .withColumnRenamed('count(session_id)', 'session_count'))

# Write output — magic committer handles atomic S3 writes
result_df.write \
  .mode('overwrite') \
  .partitionBy('date') \
  .parquet('s3a://my-output-bucket/page-view-counts/')

# Register in Glue for downstream Athena queries (requires Glue metastore conf)
result_df.write \
  .mode('overwrite') \
  .option('path', 's3a://my-output-bucket/page-view-counts/') \
  .saveAsTable('analytics.page_view_counts')

spark.stop()

When writing partitioned Parquet, Spark creates one directory per partition value (e.g. date=2026-11-06/). With mode('overwrite') and partition columns, Spark overwrites only the partitions present in the output dataframe — not the entire table. This is controlled by spark.sql.sources.partitionOverwriteMode: set it to dynamic (default) for partition-granular overwrites, or static to overwrite the entire table path.

Cancelling runaway steps

Use CancelStepsCommand to abort a step that is taking too long or consuming unexpected resources. Steps in PENDING state are cancelled immediately; steps in RUNNING state receive a SIGTERM to the YARN application.

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

export async function cancelStep(
  clusterId: string,
  stepId: string,
): Promise<CancelStepsRequestStatus> {
  const response = await emr.send(new CancelStepsCommand({
    ClusterId: clusterId,
    StepIds: [stepId],
    StepCancellationOption: 'SEND_INTERRUPT',  // graceful SIGTERM first
    // Use 'TERMINATE_PROCESS' to send SIGKILL if SIGTERM is ignored
  }));

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

  return result?.Status as CancelStepsRequestStatus;
}

SEND_INTERRUPT sends SIGTERM to the driver and allows graceful shutdown handlers to run. If the job does not terminate within 10 seconds, EMR sends SIGKILL. Use TERMINATE_PROCESS to skip straight to SIGKILL — useful when the driver is hung in a non-interruptible native call. After cancellation, the step state transitions to CANCELLED (or INTERRUPTED if the step was already running when SIGKILL was sent).

Collecting output sizes before commit

When an MCP tool needs to report on job output (row count, file size) without reading the full dataset, use Spark's DataFrame metrics rather than a full count() scan. For moderate-size outputs the most efficient pattern is to write first, then call describe from S3.

# In PySpark — after writing output, verify row count before returning success
# Option 1: collect count before write (requires full scan of input)
row_count = result_df.count()
print(f'OUTPUT_ROWS={row_count}')   # MCP tool greps syslog for this line

# Option 2: read S3 metadata after write (no data scan)
import subprocess, json
s3_path = 's3a://my-output-bucket/page-view-counts/'
proc = subprocess.run(
  ['aws', 's3', 'ls', '--recursive', s3_path.replace('s3a://', 's3://')],
  capture_output=True, text=True,
)
files = [line for line in proc.stdout.strip().split('\n') if line]
total_bytes = sum(int(line.split()[2]) for line in files)
print(f'OUTPUT_FILES={len(files)} OUTPUT_BYTES={total_bytes}')

Grep the step's syslog for these print lines after the step completes. This avoids a second Spark job just to count output rows, and the output is captured in the EMR step log which your MCP tool already reads for failure diagnosis.

Related pages

• EMR cluster management — RunJobFlow, AddJobFlowSteps, termination protection — cluster launch and management
• EMR Serverless — CreateApplication, StartJobRun, pre-initialized capacity — serverless Spark without EC2 clusters
• EMR Studio — CreateStudio, CreateEditor, SparkMagic notebooks — interactive notebook development
• AWS Glue Data Catalog for MCP servers
• Athena Iceberg tables for MCP servers