Amazon EMR

Overview

Amazon EMR is a managed platform on AWS for running open-source big data frameworks such as Apache Spark, Hive, Presto or Trino, Apache Flink and Hadoop. You describe the cluster you want, the service provisions the machines, installs the software and connects it to storage such as Amazon S3. You then submit jobs, and the cluster runs them in parallel across many nodes.

EMR is a good fit when you need to process very large tables or streams with open frameworks and want control over versions, configuration and tuning. It also appears in machine learning pipelines as the place where huge raw tables are cleaned, joined, sampled and scored before or after a model. For tabular foundation models, described on /model/research, EMR is a common way to turn billions of raw rows into model-ready shards and to apply a model to full tables in batch.

What it is

EMR stands for Elastic MapReduce, a name that comes from the original MapReduce model. Today it is a broad platform, not only MapReduce.

Think of a kitchen that can grow. A single cook cannot prepare a thousand meals in an hour. EMR hires many cooks (nodes) on demand, hands them recipes (jobs), splits the ingredients (data) among them, and sends them home when the work is done. You pay for the cooks while they work.

Deployment options

EMR comes in more than one shape.

• EMR on EC2: classic clusters made of virtual machines you configure.
• EMR on EKS: runs jobs on a Kubernetes cluster so analytics shares infrastructure with other workloads.
• EMR Serverless: you submit jobs and the service manages capacity without clusters.

This page focuses on the cluster based form, and the ideas carry over to the others.

Node roles

An EMR cluster has groups of nodes with different jobs.

• Primary node: coordinates the cluster, runs resource management and exposes interfaces. In older terms, the master node.
• Core nodes: run tasks and store data if the cluster includes HDFS.
• Task nodes: run tasks only, with no HDFS storage. They are good for elastic capacity and can use spot instances.

Frameworks it can run

Depending on the release, EMR can install a set of applications such as Spark, Hive, Presto or Trino, Flink, HBase, Hudi, Iceberg support, Jupyter based notebooks, Livy and others. Each EMR release bundles tested versions of these tools.

Storage choices

• S3 through the EMR file system: the usual choice for durable data. Clusters become disposable.
• HDFS on cluster disks: fast, temporary storage for shuffles or intermediate data.
• Local disks and instance stores: scratch space.

The recommended pattern is to keep data in S3, compute on short lived clusters, and use HDFS only for temporary work.

Where it came from

The story starts with the MapReduce paper at Google in the early 2000s and the open-source Hadoop project that followed. Hadoop gave companies a way to store data on commodity machines with HDFS and to process it with MapReduce. Running Hadoop was hard. It needed hardware, tuning and constant care.

Cloud providers saw that the pain was operational, not conceptual. AWS launched a managed Hadoop service in the late 2000s. Over time the platform:

• Added Spark, which became the main engine because it was faster and easier than MapReduce.
• Added Hive, Presto, HBase and many other frameworks.
• Introduced autoscaling and later managed scaling.
• Embraced S3 as the main storage, which enabled transient clusters.
• Added table format support, such as Iceberg, Hudi and Delta formats via connectors.
• Expanded to Kubernetes and serverless forms.

The shift from long running, HDFS centered clusters to short lived, S3 centered clusters is the key idea in the history. It changed Hadoop from a fixed installation into an elastic job service.

How it works

Step 1: you describe a cluster

You choose a release, applications, instance types and counts for each node group, networking, security settings and a bootstrap plan. Configuration can be passed as JSON classifications that set options such as Spark memory or Hive metastore location.

Analogy: writing the staffing plan for a kitchen: how many chefs, which stoves, which recipe books.

Step 2: the service provisions it

EMR launches the virtual machines, installs the chosen applications, sets up the resource manager and networking, and reports when the cluster is ready. Startup typically takes minutes.

Step 3: you submit work

There are several ways to run work.

• Steps: units of work added to the cluster, such as a Spark application, run in order.
• Direct submission: connect to the primary node and run spark-submit or SQL shells.
• Notebooks: interactive sessions attached to a cluster.
• Orchestrators: schedulers such as Apache Airflow can create clusters, add steps and shut them down.

Step 4: the resource manager allocates

A resource manager, often YARN, decides which nodes run which containers. The framework asks for executors, the resource manager grants them, and tasks flow to them.

Analogy: a restaurant manager who assigns cooks to stations.

Step 5: tasks read and write data

In a Spark job, the data is split into partitions. Each task reads a partition from S3, applies transformations, and writes results. Between stages, data may be shuffled across the network. Shuffle is the costly step, so good jobs reduce it.

Step 6: metadata comes from a catalog

Table definitions can come from the AWS Glue Data Catalog or a Hive metastore. This lets jobs refer to sales_curated.orders rather than a raw path. With AWS Lake Formation integration, EMR can enforce column and row rules for governed tables.

Step 7: the cluster scales

Two approaches exist.

• Managed scaling: the service adds or removes core and task capacity based on workload metrics, within limits you set.
• Custom autoscaling rules: you define rules on metrics.

Scaling in is safer for task nodes than for core nodes that hold HDFS data.

Step 8: logs and monitoring

Logs can be sent to S3. Metrics go to the monitoring service. History servers show Spark and YARN details. These are essential to find skew, spills and failures.

Step 9: the cluster ends

For transient clusters, the cluster terminates when the last step finishes. For long running clusters, you terminate it when no longer needed. Because data lives in S3, nothing is lost.

Key concepts and terms

A worked example

A team wants to convert raw order logs to clean Parquet and compute daily customer features for a model.

The Spark job

# job_features.py
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("daily-customer-features").getOrCreate()

orders = spark.read.parquet("s3://example-lake-clean/orders/")

day = "2026-01-15"
today = orders.filter(F.col("order_date") == day)

features = (
    today.groupBy("customer_id")
    .agg(
        F.count("*").alias("n_orders"),
        F.sum("amount").alias("total_amount"),
        F.max("order_ts").alias("last_order_ts"),
        F.avg("amount").alias("avg_amount"),
    )
    .withColumn("feature_date", F.lit(day))
)

(
    features.repartition(8)
    .write.mode("overwrite")
    .partitionBy("feature_date")
    .parquet("s3://example-lake-curated/customer_daily/")
)
spark.stop()

Create a transient cluster and run the step

import boto3

emr = boto3.client("emr", region_name="us-east-1")

resp = emr.run_job_flow(
    Name="example-daily-features",
    ReleaseLabel="emr-7.0.0",
    Applications=[{"Name": "Spark"}],
    LogUri="s3://example-emr-logs/",
    Instances={
        "InstanceGroups": [
            {"Name": "primary", "InstanceRole": "MASTER", "InstanceType": "m5.xlarge", "InstanceCount": 1},
            {"Name": "core", "InstanceRole": "CORE", "InstanceType": "m5.2xlarge", "InstanceCount": 3},
        ],
        "KeepJobFlowAliveWhenNoSteps": False,
        "Ec2SubnetId": "subnet-00000000",
    },
    Steps=[{
        "Name": "features",
        "ActionOnFailure": "TERMINATE_CLUSTER",
        "HadoopJarStep": {
            "Jar": "command-runner.jar",
            "Args": ["spark-submit", "--deploy-mode", "cluster",
                     "s3://example-code/job_features.py"],
        },
    }],
    JobFlowRole="example-emr-ec2-role",
    ServiceRole="example-emr-service-role",
    ManagedScalingPolicy={"ComputeLimits": {
        "UnitType": "Instances",
        "MinimumCapacityUnits": 3,
        "MaximumCapacityUnits": 10,
    }},
)
print(resp["JobFlowId"])

The release label and instance types are examples. Choose a current release and sizes that fit your data. Setting KeepJobFlowAliveWhenNoSteps to false makes the cluster transient.

A Spark configuration classification

[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.sql.adaptive.enabled": "true",
      "spark.sql.shuffle.partitions": "400",
      "spark.sql.parquet.compression.codec": "snappy"
    }
  }
]

Adaptive query execution lets Spark adjust partition counts and join strategies at run time, which helps with skew.

A rough numeric walk-through

Say the day's orders hold 200 million rows, and the Parquet files for the day total about 20 GB. With 3 core nodes of 8 cores each, you have 24 cores. If each task handles a 128 MB split, there are about 160 tasks, so the job runs in several waves across 24 cores. Adding capacity shortens the time until tasks per core drop near one wave. Beyond that point, extra nodes add little, because the shuffle and the final write dominate.

Scoring a table with a model in batch

A common pattern is to call a model service from each partition of the table.

def score_partition(rows):
    import requests
    batch = []
    for r in rows:
        batch.append(r.asDict())
        if len(batch) == 500:
            yield from call_and_yield(batch)
            batch = []
    if batch:
        yield from call_and_yield(batch)

def call_and_yield(batch):
    import requests
    resp = requests.post("https://model.example.internal/score", json={"rows": batch}, timeout=60)
    resp.raise_for_status()
    for row, s in zip(batch, resp.json()["scores"]):
        yield (row["order_id"], float(s))

scores = spark.createDataFrame(
    orders.rdd.mapPartitions(score_partition), ["order_id", "score"]
)
scores.write.mode("overwrite").parquet("s3://example-lake-curated/order_scores/")

Batching requests amortizes network cost. In production, add retries, back pressure and limits so the cluster does not overwhelm the model service.

Reading the Spark history to find the bottleneck

After a run, open the history server and look at these signals in order.

1. Stage durations: which stage takes most of the time?
2. Task time distribution: if the slowest task is far longer than the median, there is skew.
3. Shuffle read and write sizes: very large shuffles point to joins or aggregations that could be reduced.
4. Spill to disk: spilling means memory per task is too small or partitions are too big.
5. Garbage collection time: high values suggest memory pressure.
6. Input size per task: tiny inputs per task mean too many small files.

Fix the biggest signal first, rerun, and compare. Changing five settings at once makes it impossible to know what helped.

Choosing instance types in plain terms

Measure with a sample first. The cheapest cluster is the one that finishes the job correctly with the least idle time, not the one with the lowest hourly rate.

Orchestrating a pipeline

A daily pipeline usually has these tasks, in order.

1. Wait for the day's raw files to arrive.
2. Create a transient cluster.
3. Run the cleaning step and write clean Parquet.
4. Run the feature step and write curated tables.
5. Run data quality checks, such as row counts and null rates.
6. Register new partitions in the catalog.
7. Terminate the cluster.
8. Notify downstream jobs, or a model scoring job, that new data is ready.

An orchestrator keeps retries, timeouts and alerts in one place. If step 5 fails, the pipeline stops before bad data is published.

# Illustrative pipeline definition, not tied to a specific tool
pipeline: daily-features
schedule: "0 3 * * *"
tasks:
  - name: create_cluster
  - name: clean
    depends_on: [create_cluster]
  - name: features
    depends_on: [clean]
  - name: quality_checks
    depends_on: [features]
    on_failure: stop_and_alert
  - name: register_partitions
    depends_on: [quality_checks]
  - name: terminate_cluster
    depends_on: [register_partitions]
    run_always: true

Note the run_always flag on the last task. A failed run must still shut the cluster down, or it will keep costing money.

Handling failures

Distributed jobs fail in ordinary ways: a node disappears, a task runs out of memory, a source file is corrupt. Spark retries failed tasks by default. Your design should add the rest.

• Make writes idempotent, so rerunning a day replaces that day cleanly.
• Write to a temporary prefix, validate, then promote.
• Keep the input immutable, so a rerun reads the same data.
• Record the cluster ID, code version and input snapshot in a run log.

Strengths and limitations

Strengths

• Runs the standard open-source tools, so skills and code transfer in and out.
• Elastic: scale out for big jobs, shut down when done.
• Separation of storage and compute through S3.
• Wide choice of instance types, including spot capacity for savings.
• Control over versions, configuration and tuning.
• Integration with the Glue catalog, Lake Formation, and many other cloud services.
• Supports batch, streaming, interactive and SQL workloads.
• Several deployment forms for different operating styles.

Limitations

• You still own tuning: partitions, memory, shuffle, skew and file sizes.
• Cluster startup takes minutes, which matters for small or frequent jobs.
• Operational surface is larger than a fully serverless service: networking, roles, bootstrap, versions.
• Spot interruptions need fault tolerant design.
• Poorly sized clusters waste money or run slowly.
• Long running clusters accumulate drift and cost.
• Small files and S3 listing can slow jobs unless a table format is used.
• Not a low latency query service. For ad hoc SQL, consider Amazon Athena or Trino.

How it compares with alternatives

A frequent pattern mixes them. Use Athena for quick queries, EMR or a serverless job form for heavy transformations, and a warehouse for business reporting.

Role in tabular foundation models

Tabular foundation models learn from raw tables. See /model/architecture for how they read rows and columns, and /model/benchmarks for how they are compared. EMR does not run these models by itself, but it moves, shapes and scores the data around them at scale.

Building the pre-training corpus

Pre-training needs a large, varied collection of tables. Real tables arrive in many formats and qualities. A Spark job on EMR can:

• Read raw CSV, JSON and Parquet from many sources.
• Normalize types and null markers.
• Drop tables that are too small, too wide or mostly empty.
• De-duplicate tables and rows.
• Write each table, or each sampled chunk of it, as Parquet shards on S3.

Synthetic tables, which are generated from random structural processes, can also be produced in parallel. Each executor generates a share of the tables from a different seed, and the outputs are written to shards. That makes data generation embarrassingly parallel.

Preserving row and column structure

A tabular model treats a table as a set of rows with typed columns. EMR jobs should keep column order and types stable when reshaping data, and should store a small schema file with each shard. Avoid transformations that mix up the meaning of columns, such as pivoting without recording how.

Mixed types and missing values

Cleaning code on Spark can standardize how missing values are written, turn sentinel codes into real nulls, and cast columns to proper types. It can also compute per-column statistics, such as null share and distinct count, which help decide whether a column needs imputation or should be dropped.

No column names

If many tables have blank or cryptic headers, a job can assign neutral names like col_0001 and keep the original header text in a side file. That lets models that do not use names train on a consistent form, while preserving information for models that can use semantics.

In-context learning data

For in-context learning, the model reads a context set of labeled rows. Spark can build such contexts at scale: sample rows, split them into context and query sets, and write them as training episodes. Each episode is a small table, so a corpus of episodes is just many small Parquet files, ideally batched together to avoid small file problems.

Batch inference, imputation and anomaly detection

Once a model exists, EMR can apply it to full tables in batch, as shown in the example above. For imputation, a job scores missing cells and writes filled values with a flag. For anomaly detection, it scores every row and writes the scores for review. Partitioned output, written to new locations, makes comparisons and rollbacks easy.

Continual learning

When new tables arrive daily, a scheduled EMR job can add them to the training set, compute drift statistics against the old data, and decide which new shards to include. An orchestrator such as Apache Airflow can sequence cleaning, sharding, training and evaluation jobs, and record the snapshot of data used by each run.

Sector and semantic classification

Classification of tables and columns needs features such as value patterns, length distributions and cardinality. These are cheap to compute with Spark across millions of columns. A job can write a feature table for each column, then a classifier or a model assigns a semantic type, and the labels go back to the catalog.

Governance

With the Lake Formation integration, EMR jobs can respect column and row rules when reading governed tables. This lets a training job read only approved tables and columns. Pair it with data lineage to record which inputs produced which shards.

Practical guidance and best practices

Cluster design

• Keep data in S3 and make clusters transient where possible.
• Use a small, stable core group and elastic task nodes.
• Use spot instances for task nodes and fault tolerant work, with a fallback to on-demand capacity.
• Use instance fleets or several instance types to improve spot availability.
• Pick memory optimized instances for shuffle heavy jobs and compute optimized ones for CPU heavy work.
• Use managed scaling with sensible minimum and maximum limits.

Spark tuning

• Turn on adaptive query execution.
• Size shuffle partitions to the data, not to defaults.
• Avoid skew: salt hot keys or let adaptive skew handling split them.
• Broadcast small tables in joins.
• Cache only data that is reused and fits in memory.
• Write Parquet with sensible file sizes, and avoid thousands of tiny output files.
• Prefer built-in functions over Python user functions, which are slower.

Data layout

• Partition by columns that queries filter on, such as date.
• Use a table format such as Apache Iceberg or Delta Lake when you need updates and consistent snapshots.
• Compact small files regularly.
• Keep schemas stable and documented in the catalog.

Security

• Use roles for the service and for instances, with least privilege.
• Place clusters in private subnets, with endpoints for storage access.
• Encrypt data at rest and in transit.
• Use a catalog and Lake Formation for table level rules.
• Do not embed credentials in code or bootstrap scripts.

Operations

• Send logs to S3 and keep history servers accessible.
• Tag clusters for cost tracking.
• Set idle timeouts to end clusters that sit unused.
• Test on a small sample before a full run.
• Pin release labels so behavior does not change unexpectedly, and upgrade deliberately.

For model pipelines

• Write each stage output to a new, versioned prefix.
• Save the Spark configuration and code version with each run.
• Include checks for row counts, null rates and schema before and after each stage.
• Limit concurrency when calling model services from executors.

Common mistakes

1. Using persistent clusters by habit: idle clusters cost money and drift in configuration.
2. Storing important data only on HDFS: it disappears with the cluster.
3. Ignoring skew: one hot key makes a single task run for hours.
4. Too many tiny output files: the next job spends its time listing and opening files.
5. Default shuffle settings: wrong partition counts cause spills or idle cores.
6. Python row-by-row functions: slow compared with built-in column operations.
7. Spot-only core nodes: losing core nodes can hurt running jobs and HDFS data.
8. No cost tags: nobody knows which team or job drives the bill.
9. Oversized clusters for small data: more nodes do not speed up a job that is limited by shuffle or by a single file.
10. Hardcoded credentials: use roles.
11. Calling model services without limits: executors can overload an endpoint, causing errors and retries.
12. No data versioning: results cannot be reproduced.

Frequently asked questions

What is Amazon EMR used for?

It is used to run large scale data processing with open-source frameworks such as Spark, Hive, Flink and Hadoop. Typical work includes ETL, large joins, feature building, log analysis and batch scoring of big tables stored in S3.

What is the difference between core and task nodes?

Core nodes run tasks and can also store data when HDFS is used, so removing them carelessly can lose data or slow jobs. Task nodes only run tasks, which makes them safe to add and remove and a good place for spot instances.

Should I store data in HDFS or S3 on EMR?

Store durable data in S3 and use HDFS only for temporary or intermediate data. S3 lets clusters be short lived, because the data survives after the cluster ends.

What is a transient cluster?

It is a cluster created for a specific job and terminated when the job finishes. It saves cost and avoids configuration drift, and it works well when data lives in S3 and orchestration creates the cluster on demand.

How does EMR differ from EMR Serverless and EMR on EKS?

EMR on EC2 gives you clusters you configure and manage. EMR Serverless runs your jobs without you managing clusters. EMR on EKS runs jobs on a Kubernetes cluster you already operate. The framework code is similar, while the operating model differs.

Can EMR use the Glue Data Catalog?

Yes. You can configure Spark, Hive and others to use the Glue Data Catalog as their metastore, so tables are shared with services such as Athena. With Lake Formation integration, access rules on those tables can be enforced.

How do I make Spark jobs on EMR faster?

Reduce shuffles, fix skew, broadcast small tables, size partitions properly, use columnar formats, enable adaptive execution and avoid slow user defined functions. Check the history server to see which stage dominates before adding more machines.

Can EMR be used for machine learning on tables?

Yes. It is widely used to clean data, build features, create training shards and score full tables in batch. It can also host distributed training libraries, though dedicated training services are often used for model training itself.

How are costs controlled on EMR?

Use transient clusters, managed scaling, spot instances for task capacity, idle timeouts, right-sized instances and cost tags. Reducing the data scanned through partitioning and column pruning also lowers run time.

Is EMR good for streaming?

It can run streaming frameworks such as Spark Structured Streaming and Flink. For continuous jobs, make sure the cluster is persistent or use a form designed for long running applications, and plan for checkpoints and recovery.

Glossary

• Cluster: a set of machines working together on distributed jobs.
• Node: a single machine in a cluster.
• Primary node: the node that coordinates cluster work.
• Core node: a worker that can also store HDFS data.
• Task node: a worker that only runs tasks.
• HDFS: a distributed file system used by Hadoop.
• YARN: a resource manager that schedules containers on a cluster.
• Spark: a distributed data processing engine.
• Executor: a process that runs tasks in Spark.
• Partition: a chunk of data processed by one task.
• Shuffle: movement of data between nodes between stages.
• Skew: uneven data distribution that overloads some tasks.
• Spot instance: discounted spare capacity that may be reclaimed.
• Step: a unit of work submitted to a cluster.
• Bootstrap action: a script that runs when a node starts.
• Managed scaling: automatic cluster resizing within limits.
• Transient cluster: a short lived cluster tied to a job.
• Parquet: a columnar file format common for analytics.
• Table format: a metadata layer that adds transactions to files.
• Tabular foundation model: a model pre-trained on many tables to work on new ones.

Summary

Amazon EMR is a managed platform for running open-source big data frameworks on elastic clusters, with S3 as the durable store. You choose the release, node groups and configuration, submit steps or jobs, and let managed scaling and transient clusters control cost. It rewards good Spark habits, such as sensible partitions, skew control and healthy file sizes, and it integrates with the Glue catalog and Lake Formation for shared tables and governance.

For tabular foundation models, EMR is the heavy lifting layer around the model. It builds corpora from real and synthetic tables, preserves row and column structure, standardizes types and missing values, creates in-context episodes, computes column statistics for classification, and applies models in batch for imputation and anomaly detection. With versioned outputs and orchestration, it also supports continual learning. Read on with Apache Spark, EMR on EKS, EMR Serverless and Amazon S3 to see the neighboring pieces.

Related

It belongs to AWS data analytics in my stack, under Data engineering and lakehouse. Others in the same group: