How to Run Apache Spark on AWS EMR: Jobs, Serverless, S3, and Monitoring

Simple explanation

Think of running Spark on AWS like managing a warehouse operation. Your PySpark script is the work order. S3 is the warehouse: raw materials arrive on one set of shelves, finished goods go out on another. Amazon EMR is the warehouse itself, and Apache Spark is how the work gets organised inside it.

When a job runs, Spark assigns a driver (the foreman) and a set of executors (the workers). The driver reads the work order and divides the data into chunks. Each executor picks up its chunk, processes it in parallel, and writes its results back to S3. The driver then consolidates everything into the final output.

There are two ways to run Spark on EMR: with a cluster you manage (EMR on EC2) or without a cluster at all (EMR Serverless). The rest of this page explains when to choose each and exactly how to run jobs with both.

How Spark runs on AWS

Here is the execution flow from script to output:

1. Upload your script to S3 and submit a job. On EMR on EC2 this is a “step”; on EMR Serverless it is a “job run”.
2. EMR provisions compute. On EMR on EC2, the cluster is already running. On EMR Serverless, AWS provisions workers for this specific job (typically 1–3 minutes).
3. The Spark driver starts. In cluster mode it runs inside the cluster. It reads your script arguments, plans the execution graph, and assigns tasks to executor processes.
4. Executors process the data in parallel. Each executor reads its partition from S3 via EMRFS, processes it, and writes intermediate results to memory or local disk.
5. Output is written to S3 or Redshift. When all tasks finish, the driver consolidates results and writes the final output.
6. Logs and metrics are emitted. Step logs go to your S3 log bucket, cluster metrics go to CloudWatch, and Spark UI shows the full execution timeline.

Reading and writing S3 works natively because EMRFS is pre-installed on every EMR cluster. Your script uses standard s3:// paths with no extra configuration beyond IAM permissions on the execution role.

When to use this

This page is the right fit if you need to:

• Process large datasets (tens of GB to terabytes) with Python or Scala
• Run ETL or ELT pipelines that need distributed compute
• Join, aggregate, or transform data in a data lake before loading it into a warehouse
• Move from local Spark development to production jobs on AWS

If your job is simple SQL-based transformation on catalogued data, AWS Glue or Amazon Athena may be simpler. If you are designing the broader pipeline architecture first, read the data pipeline design guide before coming back here.

EMR on EC2 vs EMR Serverless vs Glue

If you run more than a few Spark jobs per hour, EMR on EC2 with a persistent cluster is usually more cost-effective. The per-job provisioning overhead of EMR Serverless adds up fast at high frequency. For daily or hourly batch jobs, EMR Serverless is simpler and removes cluster lifecycle management entirely.

Prerequisites before you run a job

Before submitting any Spark job on EMR, confirm you have the following in place.

• S3 bucket for scripts. Upload your PySpark script here before submitting. Example: s3://my-scripts/aggregate_orders.py.

• S3 bucket for data. Input data and output results both live in S3. Use separate prefixes: raw/ for input, aggregated/ for output.

• S3 bucket for logs. Set the log URI when creating the cluster or submitting the Serverless job run. Without it, logs disappear when the cluster terminates.

• IAM execution role. EMR on EC2 uses a service role and an EC2 instance profile. EMR Serverless uses an execution role. The role needs s3:GetObject, s3:PutObject, and s3:ListBucket on your data and log buckets. See IAM roles explained if you are new to role-based permissions on AWS.

• A running cluster (EMR on EC2 only). You need a cluster ID (j-XXXXXXXXXXXX) before adding steps. See the EMR overview for cluster creation.

Submit a Spark job to EMR on EC2

The standard way to run Spark on EMR on EC2 is to add a step. Each step runs a spark-submit command pointing to a script on S3. Use cluster mode so the driver runs inside the cluster, not on the machine you submit from.

# Upload the script to S3 first
aws s3 cp aggregate_orders.py s3://my-scripts/aggregate_orders.py

# Submit a PySpark job as an EMR step
aws emr add-steps \
  --cluster-id j-XXXXXXXXXXXX \
  --steps '[{
    "Name": "Daily Order Aggregation",
    "ActionOnFailure": "CONTINUE",
    "HadoopJarStep": {
      "Jar": "command-runner.jar",
      "Args": [
        "spark-submit",
        "--deploy-mode", "cluster",
        "--executor-memory", "4g",
        "--executor-cores", "2",
        "--num-executors", "10",
        "--conf", "spark.sql.shuffle.partitions=200",
        "s3://my-scripts/aggregate_orders.py",
        "--input", "s3://my-data-lake/raw/orders/",
        "--output", "s3://my-data-lake/aggregated/orders/",
        "--date", "2026-03-18"
      ]
    }
  }]'

# Check step status
aws emr list-steps --cluster-id j-XXXXXXXXXXXX

# Get details on a specific step
aws emr describe-step \
  --cluster-id j-XXXXXXXXXXXX \
  --step-id s-XXXXXXXXXXXX

The ActionOnFailure setting controls what happens if a step fails:

• CONTINUE: subsequent steps still run
• TERMINATE_CLUSTER: shuts the cluster down immediately
• CANCEL_AND_WAIT: skips remaining steps but keeps the cluster alive

In production, prefer CONTINUE or CANCEL_AND_WAIT. A terminated cluster removes your ability to inspect logs on the primary node if you have not configured an S3 log URI.

Sample PySpark job

This script reads raw order data from S3, filters and aggregates it, then writes results back to S3 as Parquet. The —input, —output, and —date arguments match the step definition above exactly.

import argparse
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, count, to_date

def main(input_path: str, output_path: str, date: str):
    spark = SparkSession.builder \
        .appName("DailyOrderAggregation") \
        .getOrCreate()

    # EMRFS handles s3:// paths natively — no extra config needed
    df = spark.read.json(input_path)

    daily = df.filter(to_date(col("placed_at")) == date) \
              .filter(col("status") == "completed")

    aggregated = daily.groupBy("product_category") \
                      .agg(
                          count("order_id").alias("order_count"),
                          spark_sum("amount").alias("total_revenue")
                      ) \
                      .orderBy(col("total_revenue").desc())

    # Partition by date so downstream Athena queries skip irrelevant partitions
    aggregated.write \
              .mode("overwrite") \
              .parquet(f"{output_path}/date={date}/")

    print(f"Wrote {aggregated.count()} rows to {output_path}/date={date}/")
    spark.stop()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    parser.add_argument("--date", required=True)
    args = parser.parse_args()
    main(args.input, args.output, args.date)

Partitioning output by date (/date=2026-03-18/) is important for downstream query performance. When Athena or a later Spark job reads this data, it skips entire date partitions instead of scanning every file. See partitioned tables for how partition pruning works in practice.

Run Spark with EMR Serverless

EMR Serverless removes the need to create or maintain a cluster. You create one application (a reusable runtime environment), then submit individual job runs to it. AWS provisions compute per job and releases it when the job finishes. You pay only for what each run uses.

# Create a Spark application — do this once, reuse across many job runs
aws emr-serverless create-application \
  --name daily-aggregation-app \
  --type SPARK \
  --release-label emr-7.0.0 \
  --initial-capacity '{
    "DRIVER": {
      "workerCount": 1,
      "workerConfiguration": {
        "cpu": "2vCPU",
        "memory": "4GB"
      }
    },
    "EXECUTOR": {
      "workerCount": 5,
      "workerConfiguration": {
        "cpu": "4vCPU",
        "memory": "8GB"
      }
    }
  }'

# Submit a job run — arguments match --input, --output, --date in the script
aws emr-serverless start-job-run \
  --application-id 00fabcdefabcdef \
  --execution-role-arn arn:aws:iam::123456789012:role/EMRServerlessRole \
  --job-driver '{
    "sparkSubmit": {
      "entryPoint": "s3://my-scripts/aggregate_orders.py",
      "entryPointArguments": [
        "--input", "s3://my-data-lake/raw/orders/",
        "--output", "s3://my-data-lake/aggregated/orders/",
        "--date", "2026-03-18"
      ],
      "sparkSubmitParameters": "--conf spark.executor.cores=4 --conf spark.executor.memory=8g"
    }
  }' \
  --configuration-overrides '{
    "monitoringConfiguration": {
      "s3MonitoringConfiguration": {
        "logUri": "s3://my-emr-logs/serverless/"
      }
    }
  }'

Read from S3 and write results to S3 or Redshift

Most Spark jobs on EMR read from and write back to S3. EMRFS handles this natively: your code uses s3:// paths and the connector manages the API calls. The IAM role on your cluster or EMR Serverless execution role must have read and write access to the relevant buckets.

A common next step in production data pipelines is loading processed results from S3 into Amazon Redshift for BI queries. The Redshift Spark connector does this by staging data in S3 and then issuing a COPY command:

# Write a Spark DataFrame to Redshift via the Redshift Spark connector
aggregated.write \
    .format("io.github.spark_redshift_utils.spark_redshift") \
    .option("url", "jdbc:redshift://my-cluster.us-east-1.redshift.amazonaws.com:5439/mydb") \
    .option("dbtable", "analytics.daily_order_summary") \
    .option("tempdir", "s3://my-temp-bucket/redshift-temp/") \
    .option("aws_iam_role", "arn:aws:iam::123456789012:role/RedshiftS3Access") \
    .mode("overwrite") \
    .save()

Writing through S3 and COPY is orders of magnitude faster than row-by-row JDBC inserts. For all the ways to get data into Redshift, including COPY, Firehose, and Data Sharing, see the loading data into Redshift guide.

Monitor and troubleshoot Spark jobs

When a Spark job fails, treat it like a triage situation: do not start guessing from the outside. Go straight to the error log first, then use the other tools to understand performance once the job is actually running.

Step logs in S3 (check this first)

EMR writes step-level logs to the S3 log URI you configured at cluster creation. Navigate to:

s3://your-log-bucket/{cluster-id}/steps/{step-id}/

Open stderr first. This file contains the actual exception and stack trace. The stdout file contains any print() output from your script. For EMR Serverless, logs go to the logUri you set in the s3MonitoringConfiguration block of the job run submission.

Spark UI (use this for performance problems)

Spark UI runs on port 18080 of the primary node and shows job stages, task timelines, executor utilisation, and shuffle metrics. It is the right tool once the job is running but slower than expected. Use it to spot data skew (one task taking 10x longer than others) or excessive shuffle (tasks spilling gigabytes to disk).

# Create an SSH tunnel to access Spark UI at localhost:18080
ssh -i my-key.pem -N -L 18080:primary-node-dns:18080 hadoop@primary-node-dns

# Then open http://localhost:18080 in your browser

CloudWatch metrics

EMR sends cluster metrics to CloudWatch automatically. The most useful ones to watch:

• YARNMemoryAvailablePercentage: drops when executors are memory-constrained
• HDFSUtilization: HDFS space usage on core nodes
• CoreNodesPending: indicates autoscaling is waiting for capacity
• StepsFailed: set an alarm on this to get notified when any step fails

See the CloudWatch overview for how to create alarms and configure notification channels for EMR health events.

Key Spark settings that matter first

Think of these four settings as the main controls on a forklift. Get them roughly right and the job runs well. Ignore them and the job either crawls or crashes with an out-of-memory error.

• spark.executor.memory: memory per executor. A safe starting point is 75% of the instance’s available RAM, leaving headroom for the OS and Spark overhead. If your job spills to disk, increase this first.

• spark.executor.cores: CPU cores per executor. 4–5 cores per executor is a common starting point for m5 and r5 instance types. Too many cores per executor increases memory contention between concurrent tasks.

• spark.sql.shuffle.partitions: partitions created after a join or aggregation. Default is 200. For small datasets that creates unnecessary overhead; for large datasets it causes memory pressure. Aim for 2–4x the total number of executor cores in your cluster.

• spark.dynamicAllocation.enabled: lets Spark request and release executors as workload changes across stages. Set to true for jobs where different stages have very different parallelism requirements.