Tag Archives: Amazon EMR

A systematic approach to benchmarking SQL processing engines on AWS

2026-05-19 Anubhav Awasthi

Post Syndicated from Anubhav Awasthi original https://aws.amazon.com/blogs/big-data/a-systematic-approach-to-benchmarking-sql-processing-engines-on-aws/

Selecting the right SQL processing solution for large-scale data analytics is a critical decision for organizations. As data volumes grow exponentially, the technology landscape has evolved to offer diverse options for processing and analyzing this information efficiently. This post presents a systematic framework for evaluating and benchmarking SQL processing engines on AWS, using Apache JMeter to conduct practical performance testing at scale.

The AWS analytics ecosystem

AWS offers a rich portfolio of SQL processing solutions to meet various analytical needs:

Serverless query services – Amazon Athena is a serverless, interactive query service that uses standard SQL to analyze data in Amazon Simple Storage Service (Amazon S3), offering automatic scaling, parallel query execution, and pay-per-query pricing with no infrastructure management required
Data warehouse solutions – Amazon Redshift offers scalable, high-performance cloud data warehousing with serverless options, zero-ETL integrations, AI-powered query assistance, and seamless machine learning (ML) integration for modern analytics at scale
Managed open source engines – Amazon EMR supports Apache Spark SQL, Apache Trino (formerly PrestoSQL), and other distributed query frameworks
Self-managed options – You can deploy open source engines like Apache Spark, Apache Flink, and Trino on Amazon Elastic Kubernetes Service (Amazon EKS) for greater control
Partner solutions – You can access specialized big data analytics tools through AWS Marketplace

These options are further enhanced by modern open table formats such as Apache Iceberg, Delta Lake, and Apache Hudi, which bring crucial enterprise features like ACID (Atomicity, Consistency, Isolation, and Durability) transactions, schema evolution, and time travel capabilities to data lakes. These SQL processing solutions operate under the AWS Shared Responsibility Model. AWS manages the security of the underlying infrastructure and services, and customers are responsible for secure configuration, access management, and data protection within their testing environments. This division of responsibility remains important when evaluating and benchmarking different SQL engines. Proper security configuration and implementation by customers is essential for maintaining a secure analytics environment.

Evaluation challenges in SQL engine selection

The rich ecosystem of SQL processing options creates significant evaluation challenges. Each SQL engine employs unique architectural approaches and optimization strategies, making direct comparisons complex. Organizations embarking on this evaluation journey face several interconnected obstacles:

Creating environments that accurately reflect production scenarios
Developing test datasets that mirror real-world data characteristics and volumes
Replicating real-world query patterns and concurrency levels
Maintaining uniform testing conditions across different engine architectures
Controlling infrastructure expenses throughout the evaluation process

Performance considerations at petabyte scale

When evaluating solutions for petabyte-scale deployments, the complexity intensifies considerably. Several critical factors come into play:

Resource management – Distributed SQL engines require precise balancing of CPU, memory, and storage resources. Suboptimal resource allocation can lead to query failures and performance degradation, particularly as data volumes grow.
Data distribution patterns – How data is distributed across partitions or nodes significantly impacts query performance. Data skew can create processing bottlenecks, with some nodes handling disproportionate workloads while others remain underutilized.
Concurrency handling – High-concurrency environments demand sophisticated workload scheduling and resource isolation mechanisms. The ability to maintain consistent performance under varying concurrent loads becomes a critical differentiator between solutions.
Meaningful metrics – Performance evaluation at scale requires comprehensive metrics analysis:
- Mean, median, and percentile response times (particularly p90 and p95)
- Query throughput under varying concurrency levels
- Scalability characteristics across diverse workload types
- Resource utilization efficiency during peak loads

Limitations of traditional benchmarks

Although industry-standard benchmarks like TPC-DS and TPC-H provide valuable insights, our experience with multiple customer engagements has shown that tailored, workload-specific testing often reveals performance characteristics not captured by these standardized tests. This is especially true for complex, multi-tenant environments with diverse query patterns. Organizations that complement standard benchmarks with workload-specific testing typically experience shorter proof-of-concept cycles, optimized evaluation costs, and more efficient testing operations. This comprehensive approach helps reduce uncertainty in the final solution selection process.

Prerequisites

Before you dive into the evaluation process, make sure you have the following prerequisites:

An AWS account with appropriate permissions to create and manage Amazon Elastic Compute Cloud (Amazon EC2) instances and access the SQL engines you plan to benchmark.
Basic familiarity with AWS services, particularly Amazon EC2 and the SQL engines you intend to evaluate (such as Athena, Amazon Redshift, or Amazon EMR).
Experience with SQL and data analytics concepts.
Access to the SQL engines you choose to benchmark. This post assumes you’ve already set up the engines you want to test. For setup instructions, refer to the AWS documentation for each service.
A dataset suitable for your benchmarking needs. Dataset creation and loading are not covered in this post. Build petabyte-scale synthetic test data with Amazon EMR on EC2 provides prescriptive guidance to generate test datasets at scale. Make sure your test datasets are stored in S3 buckets with encryption enabled (using SSE-KMS or SSE-S3) and that all service connections use TLS for data in transit.

Benefits of Apache JMeter

As organizations scale their analytics workloads to petabyte levels, there is a growing need for a robust, structured approach to SQL query performance testing. Although many organizations develop custom testing frameworks or use various benchmarking tools, these approaches often lack standardization and can be difficult to replicate across different SQL engines. The complexity of modern data architectures, combined with the variety of available SQL processing solutions, demands a systematic evaluation methodology. Apache JMeter emerges as a powerful solution to address this challenge. Though traditionally known for web application testing, JMeter’s extensible architecture and robust feature set make it particularly well-suited for SQL performance testing at scale.JMeter offers several advantages for evaluating SQL engines:

Support for multiple protocols and connections
Ability to simulate complex concurrent workloads
Built-in performance metrics and reporting
Extensible architecture for custom testing scenarios
Integration capabilities with continuous integration and continuous delivery (CI/CD) pipelines

Through this proposed framework, which has been validated across multiple customer engagements at petabyte scale, we aim to help organizations make more informed decisions when selecting a SQL processing solution. Our experience working with customers to assess various AWS Analytics services and open source solutions has demonstrated that a systematic evaluation approach significantly reduces proof-of-concept cycles and optimizes resource investments. This framework has helped organizations effectively evaluate services like Athena, Amazon Redshift, and Amazon EMR, alongside open source solutions such as Trino on Amazon EKS, based on their specific workload profiles and performance requirements.With this methodology, organizations can accomplish the following:

Navigate the complex landscape of large-scale data processing technologies
Reduce proof-of-concept cycles from months to weeks
Minimize infrastructure costs during evaluation phases
Make data-driven decisions about technology selection
Better align technology choices with business requirements
Establish repeatable testing patterns for future evaluations

Testing methodology in practice

A successful SQL engine evaluation requires understanding and replicating real-world workload patterns. Our methodology, refined through numerous customer engagements, focuses on comprehensive testing across multiple dimensions while remaining adaptable to specific organizational needs.

Query pattern selection

We begin by selecting representative query patterns that mirror production workloads:

Aggregation queries that summarize large datasets using operations like SUM, AVG, and COUNT
Complex join operations that test the engine’s ability to combine data efficiently across multiple tables
String operations that evaluate text processing capabilities
Nested queries that assess the engine’s optimization capabilities for complex query structures

A carefully selected set of 8–10 queries typically provides sufficient coverage while keeping the evaluation manageable. These should reflect your actual workload characteristics and business requirements.

Data volume variations

Testing across different data volumes is important for understanding scalability characteristics. We structure our tests around varying data scan ranges:

Small-scale scans – Queries accessing 1–7 days of data (megabytes to gigabytes)
Large-scale scans – Queries spanning 14–30 days (terabytes to petabytes)

This approach evaluates both I/O efficiency with large datasets and metadata handling with smaller, frequent queries, helping understand how services like Amazon EMR, Amazon Redshift, or Athena optimize query execution across different access patterns.

Concurrency testing

Real-world analytics environments rarely process single queries in isolation. Our methodology incorporates the following features:

Progressive concurrency testing starting at lower levels (typically 16, 32, 64, and 128 parallel queries), though these numbers can be adjusted based on your test infrastructure capacity and specific requirements. We recommend starting with smaller concurrency levels and gradually scaling up to understand performance characteristics
Varied query complexity and frequency (referred to as query weights) to simulate realistic workload distributions. This means some queries are run more often or are more resource-intensive than others, mimicking real-world usage patterns.
Mixed query patterns running simultaneously to test resource management.
Consistent execution across different date ranges to evaluate scaling behavior.

This approach is particularly important when evaluating managed services like the workload management capabilities of Amazon Redshift or the resource allocation strategies of Amazon EMR.

Query weight distribution

Production environments typically see varying frequencies of different query types. Our framework incorporates weighted query distribution to simulate real-world scenarios more accurately. In a typical distribution, frequent lightweight queries might represent 60% of the workload, complex analytical queries might comprise 30%, and resource-intensive data processing operations might make up the remaining 10%.This weighted approach makes sure performance testing reflects actual usage patterns rather than artificial benchmarking scenarios. The exact distribution should mirror your organization’s specific workload patterns.

Sequential vs. concurrent testing

Our methodology implements two distinct testing phases:

Sequential testing – Establishes baseline performance metrics:
- Runs each query type independently across different date ranges
- Runs multiple iterations to provide consistency and identify variability
- Helps understand individual query performance characteristics
Concurrent testing – Simulates real-world multi-user scenarios:
- Implements weighted query distributions
- Tests different concurrency levels to identify scaling limitations
- Evaluates resource management capabilities of different engines

JMeter efficiently implements both testing phases while maintaining consistent test conditions across SQL engines. Its ability to handle various JDBC connections makes it particularly suitable for testing AWS analytics services.Through this structured approach, organizations can gather comprehensive performance data reflecting their specific use cases, enabling informed SQL engine selection decisions while maintaining core principles of systematic evaluation and realistic workload simulation.

Test plans

To evaluate SQL engines’ performance under varying workloads, we designed two test scenarios: sequential and concurrent execution plans. Each scenario was executed across different data volumes by adjusting the query date range filters to cover 1, 7, 14, and 30 days. These variations simulate typical analytical workloads with progressively increasing data sizes.For sequential runs, each test was treated as a distinct batch, grouping all queries (Query 1 to Query 9) under the same date range—each query will scan data for 1, 7, 14, and 30 days with appropriate date filtering in the query’s where predicate. We used JMeter to capture average query response times for each batch. This configuration was run three times, and the final metrics reflect the average response time across these iterations to ensure reliability and account for environmental variance.Although three iterations provide initial insights, if you observe significant variations in results (typically more than 10% deviation between runs), consider expanding to 10 or more iterations. This additional sampling helps establish statistical significance, identify true performance patterns, and distinguish outliers (beyond three standard deviations) from normal variations. Document any consistent anomalies, because they may indicate important performance or security considerations for your specific environment.The following table shows the sample test plans template for the sequential test plan run.

Dataset Time Range	Run	Query Weights
Dataset Time Range	Run	Query 1	Query 2	Query 3	Query 4	Query 5	Query 6	Query 7	Query 8	Query 9
1 day	Run 1
	Run 2
	Run 3
	Avg
7 days	Run 1
	Run 2
	Run 3
	Avg
14 days	Run 1
	Run 2
	Run 3
	Avg
30 days	Run 1
	Run 2
	Run 3
	Avg

For the concurrent test plan, we introduced a probabilistic weighted distribution to the queries (Query 1 to Query 9), simulating a more realistic production-like environment where query frequency varies based on business relevance and usage patterns. This added a layer of complexity to better reflect how the SQL engine would perform under real-world concurrent access patterns.The following table shows the sample test plans template for the concurrent test plan run.

Dataset Time Range	Concurrent Runs	Query Weights
Dataset Time Range	Concurrent Runs	Query 1	Query 2	Query 3	Query 4	Query 5	Query 6	Query 7	Query 8	Query 9
1 days	8	11%	11%	11%	11%	11%	11%	11%	11%	11%
	16	10%	5%	24%	5%	5%	5%	24%	14%	10%
	32	8%	3%	24%	5%	5%	5%	24%	16%	8%
	64	7%	3%	24%	6%	4%	6%	26%	16%	9%
	128	1%	4%	19%	8%	5%	7%	14%	20%	22%
*7 days	8	11%	11%	11%	11%	11%	11%	11%	11%	11%
	16	10%	5%	24%	5%	5%	5%	24%	14%	10%
	32	8%	3%	24%	5%	5%	5%	24%	16%	8%
	64	7%	3%	24%	6%	4%	6%	26%	16%	9%
	**128	1%	4%	19%	8%	5%	7%	14%	20%	22%
14 days	8	11%	11%	11%	11%	11%	11%	11%	11%	11%
	16	10%	5%	24%	5%	5%	5%	24%	14%	10%
	32	8%	3%	24%	5%	5%	5%	24%	16%	8%
	64	7%	3%	24%	6%	4%	6%	26%	16%	9%
	128	1%	4%	19%	8%	5%	7%	14%	20%	22%
30 days	8	11%	11%	11%	11%	11%	11%	11%	11%	11%
	16	10%	5%	24%	5%	5%	5%	24%	14%	10%
	32	8%	3%	24%	5%	5%	5%	24%	16%	8%
	64	7%	3%	24%	6%	4%	6%	26%	16%	9%
	128	1%	4%	19%	8%	5%	7%	14%	20%	22%

For example, for configuration of *7 days concurrent run with **128 concurrency, the proposed configuration distributes Query 1 to Query 9 with appropriate weighted submissions such that Query 9 is executed the greatest number of times in the overall 128 executions submitted across all 9 queries for this run.

JMeter setup

To begin, you must set up JMeter on a machine that can handle the desired test load. An EC2 instance is a flexible and cost-effective option. Choose an instance type with sufficient vCPUs to support your maximum planned concurrency. For example, a c6i.4xlarge or higher is typically suitable for moderate to high throughput testing scenarios. For the operating system, you might choose Amazon Linux, which is optimized for AWS. For production-grade testing environments, deploy the JMeter EC2 instance in a private subnet of a virtual private cloud (VPC) with appropriate security groups that allow only required connections. This network isolation helps maintain security while executing performance tests. Consider using Amazon Virtual Private Cloud (Amazon VPC) endpoints for secure access to AWS services.

After the instance is provisioned, install Java (Java 17 LTS or Java 21 LTS) and download the latest version of JMeter. Be sure to configure the system with appropriate JVM options to allocate sufficient heap memory for large-scale test executions. Refer to Getting Started to learn more.

# Install Java
sudo yum update -y # For Amazon Linux
sudo yum install java-17-amazon-corretto -y

# Download JMeter and place the appropriate jdbc driver for the engine of your selection under lib folder
wget https://downloads.apache.org//jmeter/binaries/apache-jmeter-5.6.3.tgz
tar -xvzf apache-jmeter-5.6.3.tgz
cd apache-jmeter-5.6.3/lib

# Launch JMeter in GUI mode (if using a GUI-capable setup) or use CLI for remote testing
./bin/jmeter

JMeter concepts

Before you create test plans in JMeter, it’s important to understand a few foundational concepts that influence how your test plan behaves—such as thread groups, user-defined variables, and JDBC connection. These components enable the simulation of real-world query loads, including concurrency and pacing.

Test plans

The test plan is the top-level container for a JMeter test. It defines the overall testing strategy, including the queries to execute, their parameters, and the concurrent user behavior. These plans are represented as jmx files that can then be used for CLI-based execution. JMeter supports both GUI and CLI modes. It is highly recommended that you use the JMeter GUI primarily for creating test plans as jmx, and use the CLI for large load tests. You can also run thread groups consecutively for sequential execution. The default behavior is to run all thread groups in parallel suited for concurrent execution. Refer to Building a Test Plan to learn more about options available with test plans.

User-defined variables

User-defined variables are global parameters that you can reuse throughout the test plan. They are helpful for defining database credentials, server URLs, or query parameters. For example:DB_URL=jdbc:trino://trino-cluster.example.com:8889?SSL=true #Enable SSL/TLS

You can configure authentication (user name and password) through your organization’s approved methods, such as AWS Secrets Manager (see Move hardcoded secrets to AWS Secrets Manager) AWS Identity and Access Management (IAM) roles, or other secure credential management systems.

Thread groups

A thread group represents a group of virtual users (threads) executing test actions. Each thread simulates a single user sending requests to the SQL engine. This can be used to simulate concurrent runs. For example, in the preceding template, Query 3 has 19% weightage across 128 runs. This means .19*128=25 total runs, so we set the thread group to 25.

JDBC connection configuration

JDBC connection configuration sets up the database connection for the test. It specifies the database URL, driver, and credentials required for executing SQL queries. Key fields to configure are database URL and JDBC driver class. The following table summarizes the different configuration settings.

SQL Engine	JDBC Driver	JDBC Driver Class
Trino on EMR	`trino-jdbc-<trino_version>-amzn-0.jar`	`io.trino.jdbc.TrinoDriver`
Athena	Athena JDBC 3.x driver	`com.amazon.athena.jdbc.AthenaDriver`
Amazon Redshift	Amazon Redshift JDBC driver	`com.amazon.redshift.jdbc.Driver`
Trino on EKS	Trino JDBC driver	`io.trino.jdbc.TrinoDriver`

JDBC requests

The JDBC request executes SQL queries against the database using the configuration defined in the JDBC connection configuration.

For example, following command runs the JMeter in CLI mode:

# Run benchmarks in CLI mode 
./jmeter -n -t <path_to>.jmx -l <local path for log>.log -e -o <local path for>/output/

The output folder will contain an HTML report with different statistics. The following screenshot illustrates 128 concurrent runs.

Monitoring and logging

For comprehensive visibility and audit requirements, enable AWS CloudTrail logging, VPC Flow Logs, and service-specific logs (like Amazon S3 access logs). These logs can be centralized in Amazon CloudWatch Logs for monitoring and analysis. This provides proper audit trails while evaluating different SQL engines and helps track access patterns and potential security events.

Post-test steps

After running your JMeter tests, proceed with the following steps:

Review the HTML report’s key metrics, including response times, throughput, and error rates across different query types and concurrency levels.
Run identical test plans across your candidate SQL engines for direct performance comparison.
Refine your test plans based on initial findings, focusing on areas where performance differences are significant.
Factor in the cost implications alongside performance metrics to make a balanced decision.

These steps can help you systematically evaluate and select the most suitable SQL engine for your analytics workloads.

Resources

In the preceding steps, we walked through a UI-based setup for JMeter along with test plans. We have created a few sample JMeter test plans for both sequential and concurrent runs along with sample test reports. You can modify the plans to fit your needs.

JMeter sample report
JMeter test plan for sequential run
JMeter test plan for concurrent run

Clean up

After you complete your benchmarking process, clean up the resources to avoid unnecessary costs:

Stop or delete the EC2 instances used for running JMeter.
Depending on which SQL engines you used for testing, clean up active resources.
Review your AWS Management Console to confirm no active resources remain.
If you created test datasets in Amazon S3 or other storage services specifically for this benchmarking, consider deleting them if they’re no longer needed.
Although JMeter test plans and results don’t incur AWS costs, organize or delete local files as needed for your record-keeping.

Summary

Selecting the right SQL processing solution for large-scale analytics demands a systematic, data-driven approach. Our JMeter framework can help organizations effectively evaluate different SQL engines by simulating real-world workload patterns across various query types, data volumes, and concurrency levels. This methodology reduces proof-of-concept cycles and provides insights beyond traditional benchmarks, helping you assess managed AWS services like Athena and Amazon Redshift and open source solutions on Amazon EKS.

About the authors

Build petabyte-scale synthetic test data with Amazon EMR on EC2

2026-05-19 Anubhav Awasthi

Post Syndicated from Anubhav Awasthi original https://aws.amazon.com/blogs/big-data/build-petabyte-scale-synthetic-test-data-with-amazon-emr-on-ec2/

As you scale your data systems, you face a challenge: how to test thoroughly without putting customer data at risk. Using production data for testing can expose sensitive customer information to unauthorized access or breaches. For customers in regulated industries like finance and healthcare, this risk isn’t only a concern. It’s unacceptable. A data breach during testing could compromise their privacy, damage their trust, and expose organizations to significant compliance penalties. Synthetic test data solves this problem by generating artificial datasets that replicate the structure and patterns of real data without containing any actual customer information. This approach means you can test performance, validate data pipelines, and develop new features while ensuring that customer data remains protected and compliance requirements are met.

As data volumes grow from terabytes to petabytes, the architecture for generating synthetic data must evolve to meet increasing demands for scale, performance, and data quality. In this post, we show how you can build a scalable synthetic data generation solution using Amazon EMR, Apache Spark, and the Faker library.

The challenge of synthetic data generation

Traditional benchmark datasets like TPC-DS provide standardized schemas and predetermined data volumes for consistent testing environments across different systems. However, they fall short in meeting real-world testing requirements. These benchmarks don’t capture industry-specific patterns or the complex relationships found in actual production data. Their rigid schemas and simplified distributions fail to reflect business requirements, and scaling them while maintaining data consistency proves difficult. Perhaps most critically, generating massive datasets with traditional approaches requires specialized architectures to avoid proportional increases in compute costs and time.

Requirements for production-grade synthetic data

Effective workload validation demands synthetic data that mirrors production distributions while maintaining referential integrity across related tables and entities. The generation process must scale horizontally to accommodate growing data volumes while delivering deterministic results. Given identical input parameters, the system should produce the same dataset across multiple runs, supporting consistent testing cycles and comparative analysis.

Beyond technical requirements, synthetic data addresses compliance needs by minimizing exposure of personally identifiable information (PII) and protected health information (PHI) in non-production environments. This approach satisfies GDPR, HIPAA, and CCPA requirements while supporting secure cross-border data transfer, regular stress testing without compromising sensitive information, and providing an audit-friendly alternative to data masking that preserves analytical properties.

Solution overview

Architecting a synthetic data generation system that scales from terabytes to petabytes requires balancing several competing demands: the system must scale horizontally while maintaining data quality, generate large volumes efficiently, manage compute and storage resources cost-effectively, and support various schemas and output formats.

Our architecture addresses these challenges through four core components. Apache Spark on Amazon EMR provides the distributed computing framework necessary for large-scale generation. The Faker library offers synthetic data generation functions that integrate with Spark. Amazon Simple Storage Service (Amazon S3) with Apache Iceberg serves as the storage layer. We chose Iceberg for its schema and partition evolution capabilities without data rewrites, atomic transactions for consistency, precise time travel features for reproducible testing, and optimized performance at extreme scale. Amazon EMR handles dynamic resource allocation and cluster management.

The following diagram illustrates the solution architecture.

Solution architecture showing Amazon EMR generating synthetic data with Apache Spark and Faker, storing results in Amazon S3 with Apache Iceberg

Synthetic data generation at scale with Amazon EMR

Amazon EMR emerges as a particularly powerful solution for this use case, offering several advantages that directly address our requirements. It facilitates scaling of compute resources through instance fleets and Spot Instances, which can reduce costs by up to 90% compared to On-Demand pricing. The service provides built-in performance optimization for Spark applications with real-time monitoring through Amazon CloudWatch integration.

The managed infrastructure reduces operational overhead by handling the underlying Spark ecosystem and cluster lifecycle, while still providing control over scaling policies, instance types, and configurations. Integration with Amazon S3, AWS Glue, and Amazon Athena facilitates end-to-end data generation and testing workflows. Support for multiple programming languages and notebooks provides flexibility in implementing generation logic tailored to specific testing scenarios.

The synthetic data generation process follows a systematic approach designed for efficiency and scalability, as illustrated in the following diagram.

Synthetic data generation workflow showing the systematic process from configuration through data generation to storage

Although synthetic data generation isn’t a sensitive workload, it’s important to maintain robust security throughout the data generation process. Amazon EMR provides security features that align with organizational compliance requirements.

For comprehensive security guidance specific to Amazon EMR deployments, refer to Security in Amazon EMR. The solution follows the AWS Shared Responsibility Model, where AWS manages the security of the cloud infrastructure, and customers maintain responsibility for data security, access management, and compliance controls in the cloud. Specifically for synthetic data generation workloads, AWS manages the security of the underlying Amazon EMR infrastructure, network, and service operations, and customers implement appropriate security controls for their data generation pipelines. Consider the following key areas:

Data protection – Enable encryption at rest and in transit using Amazon EMR security configurations, including Amazon S3 encryption and TLS certificates for inter-node communication.
Network security – Deploy Amazon EMR clusters in private subnets with security groups following least privilege, and enable the Amazon EMR block public access feature.
Access control – Implement AWS Identity and Access Management (IAM) roles with least privilege for Amazon EMR service roles, Amazon Elastic Compute Cloud (Amazon EC2) instance profiles, and runtime roles to isolate job access. Fine-grained table-level and column-level permissions can be controlled using AWS Lake Formation. Additional authentication options are available using Kerberos and LDAP.

Optimize Faker for petabyte-scale data generation

When generating synthetic data at petabyte scale, using Faker’s implementations can quickly lead to performance bottlenecks. To overcome these limitations, adopt a combination of different optimization approaches instead of the default setup. Some of the approaches we adopted in this scenario are discussed in this section.

Faker instance pooling

The following code creates multiple Faker instances to avoid contention when generating data in parallel:

NUM_FAKER_INSTANCES = 10
faker_pool = [Faker() for _ in range(NUM_FAKER_INSTANCES)]

Consistent seed management

The following code provides reproducible data generation across distributed executors:

for faker in faker_pool:
    faker.seed_instance(42)  # For reproducibility
    random.seed(42)

Random access to Faker pool

The following code distributes load across multiple Faker instances to reduce contention:

faker = faker_pool[random.randint(0, NUM_FAKER_INSTANCES-1)]

Broadcast variables for reference data

The following code efficiently distributes reference data to all executors:

tenant_ids_broadcast = spark.sparkContext.broadcast(tenant_ids)
protocols_bc = spark.sparkContext.broadcast(protocols)

Batch generation of synthetic data

The following code generates fake data in batches rather than one-by-one:

return spark.range(1, num_endpoints + 1)
    .withColumn("hostname", random_hostname_udf())

ThreadPoolExecutor for parallel processing

The following code uses Python’s threading for parallel operations within executors:

def parallel_write_with_sync(dataframe_configs, max_workers=3):
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Parallel processing

Optimize Amazon EMR and Spark

When processing massive datasets with Spark on Amazon EMR, carefully tuning configurations can substantially enhance performance beyond the standard settings. In this section, we discuss ways to optimize the execution environment, so you can efficiently handle petabyte-scale workloads with synthetic data generation. By strategically using Spark’s advanced features and configuring Amazon EMR for your specific use case, you can improve throughput, reduce processing time, and maximize resource utilization.

Arrow configuration

The following code enables Apache Arrow for efficient data transfer between Python and JVM. The default value is false.

.config("spark.sql.execution.arrow.pyspark.enabled", "true")

Enable this configuration when your PySpark application frequently converts data between Python and JVM, especially for large DataFrames or when using Pandas operations. Keep this setting disabled for pure Spark SQL workloads or when memory is constrained.

This optimization is most effective in the following scenarios:

When processing large-scale datasets that require frequent conversion between Python and JVM.
In a PySpark application where large DataFrame operations and Pandas integration are needed.
With data science workloads that combine Python UDFs with Spark SQL operations.

Consider the following trade-offs:

Arrow maintains in-memory columnar format, resulting in increased memory consumption.
Not all data types are fully supported in older versions of Spark.
It might introduce overhead for very small datasets where conversion costs outweigh the benefits.

Adaptive query execution

The following code allows Spark to dynamically optimize query execution plans. The default value is true in Spark 3.2 and later, and false in earlier versions.

.config("spark.sql.adaptive.enabled", "true")

This optimization is generally recommended to keep enabled for most workloads. Consider disabling only when you have highly optimized, predictable queries where the adaptive overhead isn’t beneficial, or when troubleshooting query performance issues.

This optimization is most effective in the following scenarios:

Complex join operations with unknown or skewed data distributions.
Multi-stage queries where initial plans might be suboptimal.
When processing data with changing characteristics over time.

Consider the following trade-offs:

You may experience additional overhead during the query planning phase.
You might occasionally choose suboptimal plans for certain edge cases.

Parallelism configuration

The following code sets appropriate parallelism for distributed data processing based on the volume of data you’re generating. The default value for spark.default.parallelism is the total number of cores on all executor nodes or 2, whichever larger. The default value for spark.sql.shuffle.partitions is 200.

.config("spark.default.parallelism", 1000)
.config("spark.sql.shuffle.partitions", 1000)

Adjust this configuration when the default of 200 shuffle partitions creates too many small tasks (increase data volume) or too few large tasks (decrease for smaller datasets). Generally, aim for partition sizes of 100–200 MB. Modify default.parallelism when your RDD operations need different parallelism than the CPU-based default.

This optimization is most effective in the following scenarios:

When generating consistent volumes of synthetic data across multiple runs.
When you have predictable resource requirements.
When you need to precisely control executor utilization.

Consider the following trade-offs:

Static configuration might not adapt well to varying data volumes.
Too many partitions can lead to task scheduling overhead.
Too few partitions might cause memory pressure on executors.

Memory management

The following code optimizes memory allocation for execution and storage. The default value for spark.memory.fraction is 0.6, and for spark.memory.storageFraction is 0.5.

.config("spark.memory.fraction", 0.8)
.config("spark.memory.storageFraction", 0.3)

Increase memory.fraction from 0.6 to 0.8 when your workload is memory-intensive and you’re not using the JVM heap for other purposes. Adjust storageFraction based on your caching vs. execution memory needs. Decrease to 0.3 if you do minimal caching but have complex computations, and increase to 0.7 or higher for cache-heavy workloads.

This optimization is most effective in the following scenarios:

Workloads that are memory-intensive and need fine-grained control.
Workloads that balance between execution memory and cached data.
During synthetic data generation that has many interdependent fields.

Consider the following trade-offs:

Incorrect memory configuration can lead to frequent spills to disk or out-of-memory (OOM) errors.
You might need to change the configuration to suit different workload characteristics.
The settings must be monitored and tuned for optimal performance.

Limited Python UDF usage

The following code uses Spark’s built-in functions where possible instead of Python user-defined functions (UDFs). No additional configuration is needed. This is a coding practice.

.withColumn("risk_score", F.round(F.rand() * 9 + 1, 2).cast(DecimalType(3, 2)))

We recommend using Spark functions over Python UDFs when the same functionality can be achieved. Use Python UDFs only when complex business logic can’t be expressed using Spark’s built-in functions, or when integrating with specialized Python libraries.

This optimization is most effective in the following scenarios:

Simple transformations that can be performed using Spark functions.
High-throughput workloads where serialization overhead needs to be minimized.

Consider the following trade-offs:

This approach is less flexible compared to customer Python-based transformations or functions.
You might need to use complex expressions to accomplish certain data patterns.
There is a potential learning curve to familiarize yourself with Spark functions.

DataFrame caching

The following code caches frequently used DataFrames to avoid regenerating data. The default behavior doesn’t use caching. DataFrames are recomputed on each action.

endpoints_df = generate_endpoints().cache()

Use this optimization to cache DataFrames that are accessed multiple times in your application. Monitor memory usage and use MEMORY_AND_DISK storage level for large DataFrames. Uncache DataFrames when they’re no longer needed to free memory.

This optimization is most effective in the following scenarios:

When reusing reference data across multiple operations (can result in performance gains).
For workloads where the same data is processed on multiple occasions.

Consider the following trade-offs:

Too much caching might lead to memory process.
Planning is required to manage cache in environments where memory is scarce.

Optimal partitioning

By default, Spark determines partitioning based on input data and previous operations. The following code makes sure data is properly distributed across executors:

.repartition(20)

Use repartition() when you need to increase partitions for better parallelism or support even data distribution. Use coalesce() when reducing partitions to avoid small files. Generally, target 100–200 MB per partition for optimal performance.

This optimization is most effective in the following scenarios:

When controlling data distribution and avoiding data skew is very important.
Before executing an expensive operation that will benefit from balanced data distribution.
When optimizing downstream consumption use cases.

Consider the following trade-offs:

This option is more expensive than coalesce(). For large datasets, repartition() can lead to large shuffle.
The approach requires trial and experimentation to determine the optimal partition count.
There is no “one-size-fits-all” setting. Different applications or operations might gain performance with different partitioning.

Partition-aware writing

By default, data is written without partitioning. The following code organizes data for efficient storage and retrieval:

{"df": network_events_df, "name": "network_events", "partition_cols": ["tenant_id"]}

Partition data when you have predictable query patterns that filter on specific columns. Choose partition columns that are frequently used in WHERE clauses and have reasonable cardinality (avoid too many small partitions or too few large ones).

This optimization offers the following benefits:

Allows for highly parallel write operation across multiple executors.
Organizes the data that is close to real-world production data.
Allows for partition pruning when querying the data.

Consider the following trade-offs:

Excess partitioning or too fine-grained partitioning might result in small files.
It might result in data skew because of hot partitions.
You might encounter storage and metadata overhead because of excessive partitions.

Best practices

Through our journey from terabytes to petabytes, we’ve identified several best practices:

Begin with a modest dataset and incrementally scale, allowing for identification of bottlenecks at each stage.
Implement robust data validation checks to confirm synthetic data maintains expected properties at scale.
Regularly review and adjust Amazon EMR configurations, using Spot Instances and right-sizing clusters.
Develop parameterized job scripts that can adjust data volume, complexity, and cluster resources dynamically.
Design your synthetic data schema and generation logic to quickly accommodate new fields or changing distributions over time.

Conclusion

Our journey from terabytes to petabytes of synthetic data generation demonstrates how Amazon EMR, combined with Spark and Faker, can effectively address large-scale testing needs. The architecture we explored in this post scales to meet demanding data generation requirements while maintaining data quality and cost-efficiency.

We showed how starting with a solid foundation at terabyte scale, then gradually expanding through Amazon EMR managed services and Spot Instances, helps organizations build robust synthetic data pipelines. The combination of efficient data generation techniques, proper validation, and continuous monitoring provides reliable results at scale.

To begin implementing your own synthetic data generation system, start small, test thoroughly, and scale incrementally. For implementation guidance, refer to Generate production-grade synthetic data at petabyte-scale using Apache Spark and Faker on Amazon EMR.

About the authors

Streamlined monitoring and debugging for Amazon EMR on EC2

2026-05-12 Parul Saxena

Post Syndicated from Parul Saxena original https://aws.amazon.com/blogs/big-data/streamlined-monitoring-and-debugging-for-amazon-emr-on-ec2/

As organizations scale their data processing and analytics workloads on Amazon EMR on EC2, observability across cluster health, job execution, and resource usage becomes increasingly important. Teams often manage log collection across distributed nodes, correlate Amazon EMR steps with underlying YARN applications, and configure monitoring agents to capture the right level of detail for their environment.

With Amazon EMR release 7.11.0 and updates to the Amazon EMR console, Amazon EMR on EC2 introduces observability capabilities that streamline these workflows further. In this post, we walk you through five key enhancements: Amazon CloudWatch Logs integration, step-level Amazon Simple Storage Service (Amazon S3) logging controls, expanded console UIs for YARN and Tez, Amazon EMR step to YARN application ID mapping, and enhanced custom metrics with updated documentation.

What’s new

The following sections cover key improvements across the Amazon EMR console, logging, metrics collection, and documentation to give you deeper, end-to-end visibility into your Amazon EMR clusters and workloads.

1. CloudWatch Logs integration

Starting with Amazon EMR release 7.11.0, you can stream cluster logs to Amazon CloudWatch Logs in near real time without requiring custom bootstrap actions or manual agent configuration. With Amazon CloudWatch logging enabled, Amazon EMR automatically captures and streams Amazon EMR step execution logs, Spark driver, and Spark executor logs as they’re generated. This makes them immediately available for monitoring, troubleshooting, and post-mortem analysis through the CloudWatch console or API.

You can enable CloudWatch logging through the Amazon EMR console during cluster creation or programmatically using the AWS Command Line Interfaced (AWS CLI) and SDK by including the Amazon CloudWatch Agent in your application configuration and specifying your logging preferences in the configuration section.

With minimal configuration, Amazon EMR captures step logs and Spark driver logs by default, streaming them to a log group named /aws/emr/{cluster_id}. For production workloads requiring stricter organizational and security controls, you can customize the log group name, define a log stream prefix for streamlined filtering, enable encryption with an AWS Key Management Service (AWS KMS) key, and explicitly select which log types to capture. The following example demonstrates a fully customized configuration:

aws emr create-cluster
--name "EMR cluster with custom CloudWatch Logs"
--release-label emr-7.11.0
--applications Name=Spark Name=AmazonCloudWatchAgent
--instance-type m7g.2xlarge
--instance-count 3
--use-default-roles
--monitoring-configuration '
"CloudWatchLogConfiguration":
"Enabled": true,
"LogGroupName": "/my-company/emr/production",
"LogStreamNamePrefix": "cluster-prod",
"EncryptionKeyArn": "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012",
"LogTypes": {
"STEP_LOGS": ["STDOUT", "STDERR"],
"SPARK_DRIVER": ["STDOUT", "STDERR"],
"SPARK_EXECUTOR": ["STDERR", "STDOUT"]
}
}
}'

This configuration directs the logs to a custom log group (/my-company/emr/production), prefixes log stream names with cluster-prod for consistent identification across clusters, encrypts log data at rest using the specified KMS key, and captures the full set of available log types: step stdout/stderr, Spark driver, and Spark executor output. Because logs are streamed to CloudWatch as they’re written, you have near real-time visibility into job execution without waiting for log aggregation to S3 or establishing direct connectivity to cluster nodes. Combined with CloudWatch Logs Insights, you can run structured querying across log streams, making it straightforward to trace failures, correlate errors across driver and executor logs, and build metric filters or alarms based on specific log patterns.

2. Step-level S3 logging improvements

S3 logging capabilities now provide granular control over how step logs are organized and secured. You can now specify a dedicated S3 log destination and AWS KMS encryption key at the individual Amazon EMR step level. This allows different steps within the same cluster to write logs to separate S3 paths with independent encryption configurations. This is particularly useful for multi-tenant clusters or workflows with varying data classification requirements.

Step-level logging is configured through the StepMonitoringConfiguration parameter, which accepts an S3MonitoringConfiguration object where you can define the target S3 path and an AWS KMS key for encryption at rest:

"StepMonitoringConfiguration": { "S3MonitoringConfiguration": { "LogUri": "s3://your-s3-bucket/", "EncryptionKeyArn": "arn:aws:kms:your-kms-key-arn" } }

This configuration is optional. When omitted, the step inherits the default S3 log path and encryption settings defined at the cluster level during creation. With this configuration, you can override logging behavior only for the steps that require it, while maintaining a consistent default for the rest of your workflow.

3. Enhanced console with direct access to monitoring UIs

Additional live application UIs are accessible directly from the Amazon EMR Console. These console-hosted interfaces remove the need to configure SSH (Secure Shell) tunnels, set up proxies, or establish any direct network connectivity to cluster nodes to reach application web UIs. The newly added interfaces include:

YARN ResourceManager UI – Monitor cluster-wide resource allocation, queue usage, and application lifecycle states across running and completed YARN applications. This interface also provides direct access to container-level logs for running YARN applications, enabling real-time debugging without requiring node-level access.
Tez UI – Inspect Hive query execution plans, DAG visualizations, vertex-level performance metrics, and task-level counters for queries executed through the Tez execution engine (for example, Hive and Pig workloads).

These join the existing Spark History Server and YARN timeline interfaces already available through the console. By surfacing these UIs, administrators can grant developers and analysts visibility into cluster workloads and application diagnostics without exposing direct network access to cluster infrastructure while maintaining tighter security boundaries and preserving full observability into job execution and resource consumption.

With these additions, Amazon EMR now offers three complementary approaches to accessing application web interfaces, each suited to different operational requirements. Live Application UIs provide console-hosted access to web interfaces on running clusters. They’re recommended for environments where direct network connectivity to cluster nodes must be restricted from end users. On-Cluster Web UIs offer full, unrestricted access to the complete set of native application web interfaces running on cluster nodes, suited for administrators and engineers who require deep, low-level visibility. Persistent Web UIs retain application-level data beyond cluster lifetime, so you can analyze and troubleshoot workloads on terminated clusters. Together, these options give you the flexibility to balance security boundaries, access scope, and data retention based on your team’s specific monitoring and debugging workflows.

4. EMR step to YARN application ID mapping

The Amazon EMR console now surfaces the YARN Application ID directly within the EMR step details panel. For each step executing a Spark, Hive, or other YARN-based workload, the console displays the submitted YARN Application ID associated with that step, establishing a direct link between the EMR step abstraction and the underlying YARN application. With this mapping, you can:

Directly correlate EMR steps to YARN applications – when a step fails or exhibits unexpected behavior, you can immediately identify the exact YARN application to investigate rather than manually cross-referencing timestamps or job names across interfaces.
Access live monitoring tools – with the YARN application ID readily available, you can navigate directly to the YARN ResourceManager Live UI or the Spark History Server to inspect resource consumption, task-level execution details, and application state for both running and completed jobs.
Retrieve logs for detailed troubleshooting – the application ID serves as the key lookup for retrieving container-level logs persisted to Amazon S3, significantly reducing the time to root-cause failures or diagnose performance regressions.

To use this feature, open the Steps tab on your Amazon EMR cluster detail page and select the step that you want to investigate. The YARN Application ID appears in the step details panel. From there, you can use the ID to navigate to the YARN ResourceManager Live UI at http://resourcemanager-host:8088/cluster/app/<application_id>, open the corresponding view in the Spark History Server, or locate the associated container logs in your configured S3 log destination.

5. Enhanced custom metrics and observability documentation

By default, Amazon EMR automatically sends cluster-level metrics to Amazon CloudWatch at five-minute intervals, covering YARN application states, node health, HDFS utilization, and I/O activity. With Amazon EMR Release 7.0 and later, enabling the Amazon CloudWatch Agent extends this baseline with additional detailed metrics collected at one-minute intervals across cluster nodes. Furthermore, Amazon EMR 7.1 introduced custom metric classifications that you can use to define precisely which component-level metrics to collect from Hadoop, YARN, and HBase subsystems, like DataNode I/O activity, NodeManager JVM heap utilization, container resource consumption, and HBase performance counters. Each classification supports configurable export intervals, giving you control over collection granularity based on your monitoring requirements.

After enabled, custom metrics are accessible directly from the Monitoring tab in the Amazon EMR console, where you can use a classification filter to switch between HDFS, YARN, HBase custom metric groupings that you’ve defined. Metric configurations can also be updated on running clusters through the console’s reconfiguration workflow, so you can adapt your monitoring strategy as workload requirements evolve without cluster downtime. For environments using Prometheus, metrics can also be forwarded to Amazon Managed Service for Prometheus and visualized through Grafana dashboards.

The following documentation and tutorials are available to help you get the most out of these capabilities:

Enhanced Custom Metrics Guide provides step-by-step instructions for configuring CloudWatch Agent to publish custom metrics.
EMR Observability Best Practices provides a comprehensive guide covering monitoring strategies, metric selection, and troubleshooting workflows.
Service Status Monitoring provides a tutorial on monitoring and publishing Amazon EMR application status.
Monitor Apache Spark applications on Amazon EMR with Amazon CloudWatch provides a tutorial to publish detailed Spark metrics to CloudWatch and identify performance bottlenecks in Spark application.

Getting started

These observability improvements are available now for Amazon EMR on EC2. To get started:

CloudWatch Logs integration and step-level log configuration: To use these capabilities, launch a new cluster with Amazon EMR release 7.11.0 or later.
For console enhancements: Navigate to your existing Amazon EMR clusters in the AWS Console to access Live Application UI links and YARN Application ID mappings in step details, with no additional configuration required.
For custom metrics: Review our Enhanced Custom Metrics documentation to configure the CloudWatch Agent for publishing Hadoop, YARN, and HBase component metrics using custom classification files.

Conclusion

With these enhancements, Amazon EMR on EC2 provides deeper visibility into cluster health, job execution, and resource usage, helping you reduce time to root cause and focus on delivering value from your data. Note that enabling CloudWatch Logs integration and custom metrics incurs additional CloudWatch charges based on log ingestion volume and metric publishing frequency.

If you have feedback or questions, reach out to your AWS account team or post on the AWS re:Post.

About the authors

Detect and resolve HBase inconsistencies faster with AI on Amazon EMR

2026-05-12 Yu-Ting Su

Post Syndicated from Yu-Ting Su original https://aws.amazon.com/blogs/big-data/detect-and-resolve-hbase-inconsistencies-faster-with-ai-on-amazon-emr/

HBase operations teams spend hours manually correlating logs, metadata, and consistency reports to identify root causes. Traditional approaches require deep expertise and extensive investigation across scattered data sources, directly impacting MTTR and operational efficiency. As HBase deployments scale and expertise becomes increasingly scarce, organizations face mounting pressure to maintain service reliability while managing growing operational complexity. The manual nature of troubleshooting creates bottlenecks that delay incident resolution, increase operational costs, and risk service degradation during critical business periods.

In this post, we show you how to build an AI-powered troubleshooting solution using Amazon OpenSearch Service vector search and intelligent analysis. This solution reduces HBase inconsistency resolution from hours to minutes and root cause identification from days to hours through natural language queries over operational data. This democratizes HBase troubleshooting capabilities across teams and reducing dependency on specialized expertise.

Solution overview

The solution addresses HBase troubleshooting challenges through data processing, vector search, and AI-powered analysis. It processes operational data from Amazon EMR clusters, generates semantic vector embeddings, and enables natural language queries for intelligent troubleshooting.
Key components include:

Amazon EMR HBase: Runs HBase workloads with Amazon S3 as the HBase rootdir for durable, scalable storage
Data Processing: Extracts and processes HBase logs, HBCK reports, and metadata with vector embeddings
Amazon OpenSearch Service: Provides vector search capabilities with k-NN algorithms for semantic analysis
AI Analysis Interface: Enables natural language queries with context-aware recommendations
Custom Knowledge Base: Supports organization-specific runbooks and troubleshooting procedures by ingesting Git repositories via Kiro CLI‘s /knowledge add command, enabling the AI assistant to reference custom operational guides alongside HBase source code and operational tools

The preceding diagram illustrates how the HBase log analysis system troubleshoots inconsistencies through automated workflows across AWS services.

When an operations team needs to investigate HBase issues, the engineer connects over SSH to the Amazon EMR primary node and runs the error collection script, which gathers logs from HBase master and RegionServer nodes and uploads them to Amazon S3. Next, the engineer connects to the Analytics Amazon Elastic Compute Cloud (Amazon EC2) instance and executes the automated processing script, which downloads logs from Amazon S3, generates semantic vector embeddings, and injects them into Amazon OpenSearch Service for k-NN-based semantic search. The engineer then queries the Kiro CLI AI Assistant using natural language to investigate. Kiro searches Amazon OpenSearch Service for relevant log entries and uses Amazon Bedrock to analyze patterns, correlate errors across components, and provide actionable recommendations. This reduces troubleshooting time from hours to minutes. The system operates within an Amazon Virtual Private Cloud (Amazon VPC) with private subnets for Amazon EMR and Analytics Amazon EC2, AWS Identity and Access Management (AWS IAM) roles for access control, Parameter Store for configuration, and Amazon CloudWatch for monitoring.

Prerequisites

For this walkthrough, you need the following prerequisites:

AWS account setup

An AWS account with administrative access for initial deployment
AWS Command Line Interface (AWS CLI) configured with administrative credentials

Required AWS IAM permissions

For infrastructure deployment

Your deployment user or role needs the following permissions:

Your deployment user or role requires sufficient access to AWS CloudFormation, Amazon S3, AWS IAM, and AWS System Manager.
The user or role must have the ability to create AWS CloudFormation stacks.

Infrastructure deployment:

For infrastructure deployment, you need AWS CloudFormation stack management permissions.
You also require sufficient access to create and manage the following resources:
- Amazon OpenSearch Service domains
- Amazon EC2 instances, Amazon VPCs, security groups, and networking components
- AWS IAM roles and policies
- AWS Systems Manager Parameter Store entries
- Amazon CloudWatch Logs groups
- Amazon S3 bucket for access logs and session logs

Runtime service roles

The AWS CloudFormation stack automatically creates two specialized AWS IAM roles designed with least-privilege access principles.

The first role is the Amazon OpenSearch Service Role, which manages Amazon VPC networking and Amazon CloudWatch logging for the Amazon OpenSearch Service domain.

The second role is the Application Role, which provides minimal Amazon OpenSearch Service and Amazon S3 access specifically for log processing applications and secure log ingestion operations.

Network requirements

Amazon VPC with private subnets for secure Amazon OpenSearch Service deployment
NAT Gateway for outbound internet access from private subnets
Security groups configured for HTTPS-only communication

Running Kiro CLI on Amazon EC2

Kiro platform requirements:

Kiro subscription

Active Kiro License: Valid subscription to Kiro platform
User Account: Registered Kiro user account with appropriate permissions
API Access: Kiro API keys or authentication tokens for CLI access

AWS Identity Center integration

AWS IAM Identity Center Setup: AWS IAM Identity Center enabled in your AWS organization
Permission Sets: Configured permission sets for Kiro users with appropriate AWS access
User Assignment: Users assigned to relevant AWS accounts and permission sets
SAML/OIDC Configuration: Identity provider integration if using external identity systems

Additional prerequisites

Python 3.7+ and Node.js installed locally
Python 3.11+ for AWS Lambda runtime environment (required for OpenSearch MCP server compatibility)
Sufficient service quotas for Amazon OpenSearch Service instances and Amazon EC2 resources
Recommended access to the analysis instance via AWS Systems Manager Session Manager (recommended). Amazon EMR clusters running HBase workloads
EMR_EC2_Default_Role of Amazon EMR EC2 instance profile can execute describe-stacks on AWS CloudFormation stacks in us-east-1
Basic familiarity with HBase operations

The deployment follows AWS security best practices with resource-specific permissions, regional restrictions, and encrypted data storage. All AWS IAM policies implement least-privilege access patterns to help secure operation of the log analysis pipeline.

Walkthrough

This walkthrough demonstrates deploying and configuring the AI-powered HBase troubleshooting solution in five key steps:

Deploy AWS infrastructure using AWS CloudFormation
Configure Amazon EMR analysis log collection
Process and index HBase data
Enable AI-powered analysis
Add custom knowledge base (optional)

The complete solution is available in our GitHub repository.

Step 1: Deploy the infrastructure

Deploy the required AWS infrastructure including Amazon OpenSearch Service domain, Amazon EC2 instances, and AWS IAM roles.

To deploy the infrastructure

Deploy AWS CloudFormation stack. Please update [email protected] to an email address for security alerts and Advanced Intrusion Detection Environment (AIDE) reports:

# Deploy to development environment
aws cloudformation create-stack \
  --stack-name dev-hbase-log-analysis \
  --template-body file://cloudformation/hbase-log-analysis-simple.yaml \
  --parameters \
    ParameterKey=EnvironmentName,ParameterValue=dev \
    ParameterKey=EC2InstanceType,ParameterValue=m7g.xlarge \
    ParameterKey=SecurityAlertEmail,[email protected] \
  --capabilities CAPABILITY_IAM \
  --region us-east-1
# Wait for deployment to complete (~15-20 minutes)
aws cloudformation wait stack-create-complete \
  --stack-name dev-hbase-log-analysis \
  --region us-east-1

Note the deployment outputs including Amazon OpenSearch Service endpoint and Amazon EC2 instance details in the AWS CloudFormation console.

The deployment creates:

Amazon OpenSearch Service domain with vector search capabilities
Amazon EC2 instance for data processing and AI analysis
AWS IAM roles with appropriate permissions
Security groups and Amazon VPC configuration

Step 2: Connect to Amazon EC2 instance and set up system

Connect to the Amazon EC2 instance using AWS Systems Manager (SSM) and set up the required components.

To connect and set up the system

Run the following commands to get the instance ID from AWS CloudFormation outputs and connect via AWS Systems Manager (SSM):

# Get instance ID
INSTANCE_ID=$(aws cloudformation describe-stacks \
  --stack-name dev-hbase-log-analysis \
  --query 'Stacks[0].Outputs[?OutputKey==`EC2InstanceId`].OutputValue' \
  --output text \
  --region us-east-1)
# Connect via SSM
aws ssm start-session --target $INSTANCE_ID --region us-east-1

Clone the repository and run automated setup:

# On EC2 instance
sudo su - ec2-user

# Re-install aws cli
sudo dnf remove awscli -y

# For ARM64 (Graviton instances - default)
curl "https://awscli.amazonaws.com/awscli-exe-linux-aarch64.zip" -o "awscliv2.zip"

# For x86_64 (if using non-Graviton instances)
# curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"

unzip awscliv2.zip
sudo ./aws/install

# update $PATH in ~/.bashrc
echo 'export PATH=$PATH:/usr/local/bin/' >> ~/.bashrc

# Reload ~/.bashrc
source ~/.bashrc

# Fork and clone the source code repository on GitHub: sample-emr-hbase-inconsistencies-detection-recovery-mcp-kiro
git clone https://github.com/YOUR_USERNAME/sample-emr-hbase-inconsistencies-detection-recovery-mcp-kiro.git hbase-analysis
cd hbase-analysis

# Run automated setup
chmod +x ./scripts/setup/automated-system-setup.sh
./scripts/setup/automated-system-setup.sh \
  --emr-version emr-7.12.0 \
  --stack-name dev-hbase-log-analysis \
  --region us-east-1

The automated setup script installs:

System dependencies (awscli, git, unzip)
uv package manager and OpenSearch MCP Server
Kiro CLI and configuration with AWS IAM Identity Center authentication. The script will automatically add Apache HBase open source repo and Apache HBase open source operational tools to knowledge bases
HBase source repositories for your Amazon EMR version
Python dependencies and MCP server configuration

Add your own knowledge base to Kiro CLI

To enhance Kiro CLI’s analysis capabilities with Apache HBase open-source repositories, your organization’s HBase runbooks and troubleshooting guides, you can add your own knowledge base repositories. Here are the commands. Please periodically validate and maintain your runbook contents so that they remain accurate and up-to-date, reflecting any changes in your HBase environment, configurations, or operational procedures.:

# Navigate to the HBase repositories directory
cd /opt/hbase-repositories
# Clone your organization's HBase runbook repository
git clone <runbook-repository-url> <your-own-runbook-repo>
# Example:
# git clone https://github.com/your-org/hbase-runbooks.git hbase-runbooks
# git clone https://gitlab.company.com/ops/hbase-troubleshooting.git hbase-troubleshooting
# Add your custom repositories to Kiro CLI knowledge base manually (run these commands inside kiro-cli):
echo "/knowledge add --name \"Your custom HBase knowledge base\" --path /opt/hbase-repositories/<your-own-runbook-repo>" | kiro-cli
# Example:
# echo "/knowledge add --name \"Company HBase runbooks\" --path /opt/hbase-repositories/hbase-runbooks" | kiro-cli
# echo "/knowledge add --name \"HBase troubleshooting guides\" --path /opt/hbase-repositories/hbase-troubleshooting" | kiro-cli

Step 3: Configure Amazon EMR log analysis collection

Set up data collection from your Amazon EMR clusters to gather HBase logs, metadata, and consistency reports using the recommended direct collection method.
To configure Amazon EMR log analysis collection

On your Amazon EMR cluster primary node, run the following commands to download the collection scripts:

# On EMR primary node
sudo su - hadoop

# Fork and clone the source code repository on GitHub: sample-emr-hbase-inconsistencies-detection-recovery-mcp-kiro
git clone https://github.com/YOUR_USERNAME/sample-emr-hbase-inconsistencies-detection-recovery-mcp-kiro.git hbase-analysis
cd hbase-analysis

Run the interactive collection wizard:

# Run collection wizard
python3 scripts/utilities/emr_log_collection/emr_cluster_wizard_v2.py

Input the parameters like the EMR cluster’s jobflow ID, the log analysis Amazon S3 bucket name, and the lookback hours. The default value of the lookback hours is 4 hours.

The collection wizard performs these actions:

Collects HBase logs from local filesystem. Please reference to prerequisites for the access permission.
Runs sudo -u hbase hbase hbck -details (or hbck2 for HBase 2.x)
Runs hdfs dfs -ls -R /hbase or aws s3 ls <hbase-root-dir> –recursive
Runs hbase shell <<< 'scan "hbase:meta"'
Creates properly named files matching analysis system requirements
Uploads to Amazon S3 with correct naming conventions

Here’s the data collection summary:

You can check the uploaded contents through AWS CLI.

aws s3 ls s3://<log-path> --recursive

Here’s a screenshot of the outputs.

On the Analysis Amazon EC2 instance, download collected files to the Analysis Amazon EC2 instance.

# On analytics EC2 instance
sudo su - ec2-user

# Download logs from S3
mkdir -p /tmp/hbase-log-analysis
cd /tmp/hbase-log-analysis
aws s3 sync s3://<S3-BUCKET-NAME>/emr-logs/<EMR-JOBFLOW-ID>/ .

You can get your jobflow ID from Amazon EMR console:

The generated files (hbase-hbase-master-ip-xxx-xxx-xxx-xxx.ec2.internal.log.gz, hbase-hbase-regionserver-ip-xxx-xxx-xxx-xxx.ec2.internal.log.gz, hbck_report.txt, hbase_rootdir_paths.txt, hbase_meta.txt, hbase_processes.txt, log_copy_summary.txt) should be aligned with the automated processing script requirements as following.

Step 4: Process and index data

Process the collected HBase data and create vector embeddings for intelligent search capabilities.To process and index the data, please navigate to the project directory on the Analysis EC2 instance, and run automated-log-processing.sh:

sudo su – ec2-user
cd ~/hbase-analysis
chmod +x ./scripts/processing/automated-log-processing.sh
./scripts/processing/automated-log-processing.sh \
  --job-flow-id j-YOUR-JOB-FLOW-ID \
  --stack-name dev-hbase-log-analysis

The processing scripts extract and parse HBase logs and generate dimensional vector embeddings from HBase log messages using sentence transformer models to enable semantic search beyond keyword matching. The system uses the all-MiniLM-L6-v2 model by default (producing 384-dimensional embeddings), but supports configurable models with different embedding dimensions, automatically adapting the OpenSearch vector index to match the chosen model’s output. The system processes comprehensive HBase operational data including region operations, compaction activities, Write-Ahead Log events, memstore operations, and cluster management information from HMaster and RegionServer logs. Vector embeddings capture error messages, exception stack traces, performance warnings, and multi-line log entries through intelligent text preprocessing. This semantic representation enables advanced troubleshooting where users can query conceptually for “region server performance issues” or “memory pressure” and receive contextually relevant results across different log files and time periods. The vector search capabilities support error correlation by grouping similar exceptions, performance analysis by identifying related bottlenecks, and operational pattern recognition. Each log entry is stored in Amazon OpenSearch Service with original metadata (timestamp, log level, source file, job flow ID) alongside the embedding vector, enabling both structured queries and AI-powered semantic analysis. This approach transforms raw HBase logs into a searchable knowledge base supporting anomaly detection, trend analysis, and predictive insights for proactive cluster management and troubleshooting.

All scripts use AWS IAM authentication automatically. Here’s a screenshot of the data processing outputs.

Step 5: Enable AI-powered analysis

Configure the AI analysis interface to enable natural language queries against your HBase operational data.

To set up AI-powered analysis

Launch Kiro CLI (already configured by automated setup):

kiro-cliCheck mcp and knowledge bases. /mcp list

/knowledge show

If you cannot see these 2 knowledge bases, you can manually add them through the following commands:

# Note: Large repositories (~500MB) may take a while to index. Check progress with: /knowledge show
/knowledge add --name "HBase operational tools" --path /opt/hbase-repositories/hbase-operator-tools"
/knowledge add --name "Apache HBase source code" --path /opt/hbase-repositories/hbase"

Use natural language queries to analyze your HBase data. The AI analysis uses both the OpenSearch MCP Server for querying indexed data and the Filesystem knowledge bases for accessing HBase source code. You can add your custom runbooks for Kiro’s reference as well.

For HBase inconsistency analysis:

# HBase Inconsistency Detection and Remediation Guidelines
## Search Strategy
- Use fuzzy search for case variations/typos, term query for exact region IDs, match_phrase for paths, query_string for logs
- Always use .keyword subfields for exact text matching
- Cross-reference filesystem (wildcard: {"wildcard": {"path": "*<region_id>*"}}) with hbase:meta (match: {"match": {"row_key": "<region_id>"}})
- The total region count in hbase meta must match the total matched document count of wildcard path like "*/.regioninfo" in hbase rootdir path.  
- All terms of region_name.keyword for a region encoded name must match a wildcard path like "*/.regioninfo"
- All terms of table_name.keyword for a table must match a wildcard path like "*/.tabledesc*"
- 1595e783b53d99cd5eef43b6debb2682 is the master store region that will locate in <hbase-root-dir>/MasterData/data/master/store/1595e783b53d99cd5eef43b6debb2682/
- May cross check with the raw logs in /tmp/hbase-log-analysis/
## Issue Types
Orphan regions, missing .regioninfo, missing/extra regions in hbase:meta, rowkey holes, stuck RIT, master initialization failures
## Analysis Steps
### 1. Cross-Reference Meta vs Filesystem
- Filesystem regions NOT in hbase:meta → ORPHAN REGION
- Meta regions NOT in filesystem → MISSING REGION
### 2. Validate Region Chain Continuity
- Sort regions by STARTKEY, verify region[i].ENDKEY == region[i+1].STARTKEY
- First STARTKEY must be '', last ENDKEY must be ''
- Gaps → ROWKEY HOLE
### 3. Check Region States
- state != 'OPEN' → Check RIT
- Missing server assignment → UNASSIGNED
- Multiple servers → SPLIT BRAIN
- "deployed_servers" field must have only one region server address like "ip-xxx-xxx-xxx-xxx.ec2.internal,16020,1770781485397" . The value should not be null or have multiple values. 
### 4. Validate .regioninfo Files
- Missing .regioninfo in region directory → CORRUPT REGION
### 5. Cross-Check HBCK Report
- Compare orphan counts, RIT regions, filesystem vs meta region counts
### 6. Analyze Logs
- Search: "updating hbase:meta row=<region>", "STUCK", "RIT", "Failed" + "<region>", "Split"/"Merge" + "<region>"
## Remediation
- Reference knowledge bases: "Apache HBase source code", "HBase operational tools"
- Use hbck2: /usr/lib/hbase-operator-tools/hbase-hbck2.jar
- Prefix commands with sudo -u hbase
- Use aws s3 for S3-based rootdir
- Wait 300s after creating holes before hbck fixMeta (catalog janitor cycle)
- Use unassign instead of deprecated close_region
- If the region does not have .regioninfo in  <hbase-root-dir>/data/<namespace>/<table-name>/<region-encoded-name>/ but hbase:meta has that region's information and that region has been deployed on a healthy region server, you can use hbase shell to unassign and assign the region to re-generate .regioninfo
- Always add "sudo -u hbase hbase" before "hbase shell" and "hbase hbck" commands
## Job flow
Target: <your-job-flow-id>
Inconsistency to detect: All kinds of inconsistencies

You can trust or input “y” or “t” to grant Kiro to search through mcp and knowledge bases.

You may get some outputs like this: Kiro checked for any HBase issue.

Kiro summarized the examination results.

Kiro provided mitigation commands after Kiro summarized the issue.

Cleaning up

To avoid incurring future charges, delete the resources created during this walkthrough.

To clean up the resources

Delete the AWS CloudFormation stack from AWS Management Console:

Clean up Amazon EMR cluster resources (if created only for this walkthrough):

Verify resource cleanup in the AWS Console to verify that all resources are deleted and review your AWS bill to confirm no unexpected charges.

Important considerations:

Amazon OpenSearch Service domains take several minutes to fully delete
Amazon S3 buckets with versioning retain object versions
Use smaller instance types for development to optimize costs
Monitor usage with AWS Cost Explorer

Conclusion

In this post, we showed you how to build an AI-powered HBase troubleshooting solution that transforms manual log analysis into an automated workflow. By combining Amazon OpenSearch Service vector search with Amazon Bedrock-powered analysis through the Kiro CLI, operations teams can resolve complex HBase inconsistencies faster and gain deeper operational insights. The solution demonstrates how AI augments human expertise to improve operational efficiency, reducing HBase inconsistency resolution from hours to minutes and root cause identification from days to hours. Ready to transform your HBase operations? Get started with the GitHub repository and explore the Amazon OpenSearch Service documentation for additional guidance on vector search capabilities.

Acknowledgments

The author would like to thank Xi Yang, Anirudh Chawla, and Sasidhar Puthambakkam for their contributions to developing the technical solution. Xi Yang is a Senior Hadoop System Engineer and Amazon EMR subject matter expert at AWS. Anirudh Chawla is an AWS Analytics Specialist Solution Architect who helps organizations empower businesses to harness their data effectively through AWS’s analytics platform. Sasidhar Puthambakkam is a Senior Hadoop Systems Engineer and Amazon EMR Subject Matter Expert who provides architectural guidance for complex BigData workloads.

About the authors

How to use streamlined permissions for Amazon S3 Tables and Iceberg materialized views

2026-05-11 Srividya Parthasarathy

Post Syndicated from Srividya Parthasarathy original https://aws.amazon.com/blogs/big-data/how-to-use-streamlined-permissions-for-amazon-s3-tables-and-iceberg-materialized-views/

Apache Iceberg has emerged as the open table format for data lakes. It handles petabyte-scale datasets, lets teams evolve schemas and partitions in place, and supports time travel and incremental processing for data lake management at scale. Amazon S3 Tables provide a fully managed Apache Iceberg table experience in Amazon S3, optimized for analytics workloads, and integrate with the AWS Glue Data Catalog so AWS analytics services such as Amazon Redshift, Amazon EMR, Amazon Athena, Amazon SageMaker, and AWS Glue query your data. Together, they form the foundation of a modern data lake architecture on AWS.

S3 Tables integrate with the AWS Glue Data Catalog using AWS Identity and Access Management (IAM) – based authorization. If you manage analytics workloads across these services, you can now define permissions across storage, catalog, and compute in a single IAM policy. This gives teams already using IAM a straightforward path to govern access to S3 Tables resources without changing their existing permission model. For fine-grained access controls, you can opt in to AWS Lake Formation at any time through the AWS Management Console, AWS Command Line Interface (AWS CLI), API, or AWS CloudFormation.

Iceberg materialized views created in the Glue Data Catalog extend this foundation by letting you store pre-computed query results as Iceberg data on Amazon S3. When a query repeats aggregations or joins across large datasets, the engine reads directly from the materialized view’s S3 location rather than reprocessing the base tables. A materialized view can reside in S3 Tables or in an S3 general purpose bucket, independent of where its base tables live, which lets you place pre-computed results wherever fits your access patterns and cost model best.

In this post, we walk through how to set up and manage S3 Tables in the AWS Glue Data Catalog, create and query Iceberg materialized views, and configure access controls that work across your analytics stack with IAM-based authorization.

Solution overview

The above architecture illustrates how S3 Tables integrate with AWS Glue Data Catalog using IAM-based authorization, so you can define the necessary permissions across storage, catalog, and query engines in a single IAM policy. This permission model accelerates onboarding for new teams and workloads.

Key architecture components include:

Storage Layer: Data stored as Iceberg tables in Amazon S3 Tables

Catalog Layer: AWS Glue Data Catalog serves as the single metadata repository.

Compute Layer – Amazon Athena, AWS Glue, Amazon Redshift, and Amazon EMR connect to a single data Catalog to access Iceberg tables.

Security: AWS IAM authorizes access to resources in storage, catalog, and compute layers.

Prerequisites:

To follow along with this post, you must have an AWS account and an IAM role or user with appropriate permissions and familiarity to the following services:

IAM
AWS Glue Data Catalog
Amazon S3
Amazon Athena
Amazon Redshift
Amazon EMR

For the minimum permissions required for the role/user for metadata and data access, refer to required IAM permissions documentation.

Solution walkthrough

In this walkthrough, you will integrate S3 Tables with the AWS Glue Data Catalog, create Iceberg materialized views, and query data using multiple analytics engines. You will also learn to use materialized views when you have complex aggregations queried frequently but underlying data changes. You can follow these steps to implement the solution. It will take about 45–60 minutes to complete this walkthrough.

Setup S3 Tables and integrate with Glue Data Catalog

Navigate to Amazon S3 console:

On the left menu, select Table buckets.
Choose the Create table bucket button.

In the next screen, we will fill the name of the bucket as salesbucket. Please ensure the Enable Integration configuration is checked. This step integrates S3 Tables with AWS Glue Data Catalog.

Keep the other options as default and choose Create table bucket.
After it is created, you will be redirected back to the list of table buckets. Choose the table bucket salesbucket.
Select the Create table with Athena button.
Create a namespace in S3 Tables which is equivalent to a database in AWS Glue Data Catalog. Enter namespace (database) name as “sales” and click Create namespace.

Choose Create table with Athena, and a new tab will be open with the Amazon Athena console.
When the Amazon Athena console opens, you will see an example of a query to create a table and examples to insert rows in that table. You could use this query block by uncommenting the code and executing each statement individually by highlighting it. At the end, you will have data in the table.

Query S3 Tables and create materialized view using Amazon EMR:

To run the instruction on Amazon EMR, complete the following steps to configure the cluster:

Create an IAM role for the Amazon EMR instance profile following the Amazon EMR Management Guide. Add the following as policies and trust relationship for working on materialized views.

Replace ACCOUNT_ID with your AWS account ID, Instance_profile_role to the Amazon EMR instance profile role, and REGION with your AWS Region.

{
   "Version":"2012-10-17",
   "Statement":[
      {
         "Sid":"GlueDataCatalogPermissions",
         "Effect":"Allow",
         "Action":[
            "glue:GetCatalog",
            "glue:GetDatabase",
            "glue:CreateTable",
            "glue:GetTable",
            "glue:GetTables",
            "glue:UpdateTable",
            "glue:DeleteTable"
         ],
         "Resource":[
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:catalog",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:catalog/s3tablescatalog",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:catalog/s3tablescatalog/*",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:database/salesdb",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:database/salesdb/*",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:database/s3tablescatalog",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:database/s3tablescatalog/*",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:table/s3tablescatalog/*",
            "arn:aws:glue:<REGION>:<ACCOUNT ID>:table/*/*"
         ]
      },
      {
         "Sid":"S3TablesDataAccessPermissions",
         "Effect":"Allow",
         "Action":[
            "s3tables:GetTableBucket",
            "s3tables:GetNamespace",
            "s3tables:GetTable",
            "s3tables:GetTableMetadataLocation",
            "s3tables:GetTableData",
            "s3tables:ListTableBuckets",
            "s3tables:CreateTable",
            "s3tables:PutTableData",
            "s3tables:UpdateTableMetadataLocation",
            "s3tables:ListNamespaces",
            "s3tables:ListTables",
            "s3tables:DeleteTable"
         ],
         "Resource":[
            "arn:aws:s3tables:<REGION>:<ACCOUNT ID>:bucket/*"
         ]
      },
      {
         "Effect":"Allow",
         "Action":"iam:PassRole",
         "Resource":"arn:aws:iam::<ACCOUNT ID>:role/service-role/<Instance_profile_role>"
      }
   ]
}

Add the following to the trust policy in addition to existing:

 {
            "Sid": "",
            "Effect": "Allow",
            "Principal": {
                "Service": "glue.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }

Launch an Amazon EMR cluster 7.12.0 or higher with instance profile role created in the previous step and with Iceberg enabled. For more information, refer to Use an Iceberg cluster with Spark.
Connect to the primary node of your Amazon EMR cluster by using SSH, and run the following command to start a Spark application with the required configurations:

Replace bucket_name with your bucket name.

spark-sql \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.glue_catalog.type=glue \
  --conf spark.sql.catalog.glue_catalog.warehouse=s3://<bucket_name> \
  --conf spark.sql.catalog.glue_catalog.glue.region=<region> \
  --conf spark.sql.catalog.glue_catalog.glue.id=<accountid>:s3tablescatalog/salesbucket \
  --conf spark.sql.catalog.glue_catalog.glue.account-id=<accountid> \
  --conf spark.sql.catalog.glue_catalog.client.region=<region> \
  --conf spark.sql.optimizer.answerQueriesWithMVs.enabled=true \
  --conf spark.sql.defaultCatalog=glue_catalog

Run the following queries to query the daily_sales table.

spark-sql ()> use sales;
spark-sql (sales)> select * from daily_sales;
2024-01-15 Laptop 900.0
2024-01-15 Monitor 250.0
2024-01-16 Laptop 1350.0
2024-02-01 Monitor 300.0
2024-02-01 Keyboard 60.0
2024-02-02 Mouse 25.0
2024-02-02 Laptop 1050.0
2024-02-03 Laptop 1200.0
2024-02-03 Monitor 375.0

Create Materialized view.

CREATE MATERIALIZED VIEW sales_mv as 
SELECT 
    product_category,
    COUNT(*) as units_sold,
    SUM(sales_amount) as total_revenue, 
    AVG(sales_amount) as average_price 
FROM 
    glue_catalog.sales.daily_sales 
GROUP BY 
    product_category;

A newly created materialized view is populated with the initial query results but does not update automatically as base table data changes. To keep it current, specify a REFRESH EVERY clause when creating the view. This accepts a time interval and unit, so you can define how often the materialized view is recomputed from the base tables.

Add refresh interval.

CREATE MATERIALIZED VIEW sales_mv 
SCHEDULE REFRESH EVERY 2 HOURS as 
SELECT 
    product_category,
    COUNT(*) as units_sold,
    SUM(sales_amount) as total_revenue, 
    AVG(sales_amount) as average_price 
FROM 
    glue_catalog.sales.daily_sales 
GROUP BY 
    product_category;

Alternatively, you can refresh them manually.

For manual full refresh, you can use the following command:

REFRESH MATERIALIZED VIEW sales_mv FULL;

For manual incremental refresh, you can use the following command:

REFRESH MATERIALIZED VIEW sales_mv;

For more details, refer to Refreshing materialized views.

Query the MV.

spark-sql (sales)> select * from sales_mv
Keyboard 1 60.0 60.0
Laptop 4 4500.0 1125.0
Mouse 1 25.0 25.0
Monitor 3 925.0 308.3333333333333

After the Iceberg materialized views are created, you can access them using IAM principals that have required IAM permissions to Glue Data Catalog resource and its underlying storage.

Iceberg materialized views are flexible in how they combine base tables and access control modes. Base tables can reside in S3 general-purpose buckets (with IAM or Lake Formation access control), in S3 Tables (through the s3tablescatalog catalog), or a combination of these—all within a single materialized view definition. The materialized view itself can use either IAM or AWS Lake Formation access control, independently of its base tables.

For more details, refer to How materialized views work with AWS Glue.

Query using Athena:

Additionally, you can query the same materialized view from Athena SQL. The following image shows the same query run on Athena and the resulting output.

Query using Amazon Redshift:

To query the S3 Tables in AWS Glue Data Catalog using Amazon Redshift, you must create a database in the default catalog in Glue Data Catalog that points to the S3 Tables catalog.

On the AWS Glue console, choose Databases, and then choose Add Database.

Choose the Glue Database resource link option, add a name for the database, choose salesbucket on the target catalog and sales as the target database. Then select Create database.

After creating the database, we will see the “salesdb” resource link under Databases on AWS Glue Data Catalog.

Create IAM role with the following policy for the Amazon Redshift schema creation. Replace the AWS Region and account ID for your account.

{
   "Version":"2012-10-17",
   "Statement":[
      {
         "Sid":"GlueDataCatalogPermissions",
         "Effect":"Allow",
         "Action":[
            "glue:GetCatalog",
            "glue:GetDatabase",
            "glue:CreateTable",
            "glue:GetTable",
            "glue:GetTables",
            "glue:UpdateTable",
            "glue:DeleteTable"
         ],
         "Resource":[
            "arn:aws:glue:<REGION>:<ACCOUNTID>:catalog",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:catalog/s3tablescatalog",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:catalog/s3tablescatalog/*",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:database/salesdb",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:database/salesdb/*",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:database/s3tablescatalog",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:database/s3tablescatalog/*",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:table/s3tablescatalog/*",
            "arn:aws:glue:<REGION>:<ACCOUNTID>:table/*/*"
         ]
      },
      {
         "Sid":"S3TablesDataAccessPermissions",
         "Effect":"Allow",
         "Action":[
            "s3tables:GetTableBucket",
            "s3tables:GetNamespace",
            "s3tables:GetTable",
            "s3tables:GetTableMetadataLocation",
            "s3tables:GetTableData",
            "s3tables:ListTableBuckets",
            "s3tables:CreateTable",
            "s3tables:PutTableData",
            "s3tables:UpdateTableMetadataLocation",
            "s3tables:ListNamespaces",
            "s3tables:ListTables",
            "s3tables:DeleteTable"
         ],
         "Resource":[
            "arn:aws:s3tables:<REGION>:<ACCOUNTID>:bucket/*"
         ]
      }
   ]
}

Create an Amazon Redshift provisioned cluster or Amazon Redshift Serverless, attaching the IAM role created in previous step.

To access the AWS Glue Catalog and the resource link, you can now log in to Amazon Redshift as a local user. We use the admin user and Amazon Redshift Query Editor v2.

To create the external schema, you must run the following command: Replace ACCOUNT_ID with your AWS Account ID, IAM_ROLE to IAM role created for schema access, and REGION with your AWS Region.

CREATE EXTERNAL SCHEMA salesdb
FROM DATA CATALOG DATABASE 'salesdb'
IAM_ROLE 'arn:aws:iam::<ACCOUNT_ID>:role/<IAM_ROLE>'
REGION '<REGION>'
CATALOG_ID '<ACCOUNT_ID>';

After you have created the external schema, it will show up on the left side, under the dev database. The table that we created, daily_sales, is available and we can query directly from Amazon Redshift using a local user.

Cleanup:

After completing the walkthrough, follow these steps to remove the resources and avoid ongoing charges. These cleanup steps will permanently delete the data, including the daily_sales table and sales_mv materialized view. Make sure that you have backed up the data that you need to retain before proceeding.

To avoid incurring future charges, clean up the resources that you created during this walkthrough:

Remove the Glue Data Catalog resources
Delete the table bucket
Terminate and Delete the Amazon Redshift cluster
Terminate and Delete the Amazon EMR cluster
Delete the IAM roles/policies created

Conclusion

Amazon S3 Tables now integrate with AWS Glue Data Catalog through IAM-based authorization via a single IAM policy. By consolidating permissions for storage, catalog, and query engines into one IAM policy, you can streamline authorization with AWS analytics services like Amazon Athena, Amazon EMR, and AWS Glue. You can use this streamlined IAM authorization model to build your data lake faster while maintaining enterprise-grade security. For organizations with additionally granular data access requirements, AWS Lake Formation remains available to layer fine-grained access controls on top of this foundation. This is configurable through the AWS Management Console, CLI, API, or CloudFormation. This integration allows AWS analytics users to use IAM and scale their analytics capabilities with reduced operational complexity.

To learn more about to S3 Tables and integration with Glue Data catalog, visit: Amazon S3 Tables integration with AWS analytics services overview and Integrating with Amazon S3 Tables.

About the authors

A guide to Airflow worker pool optimization in Amazon MWAA

2026-05-01 Boyko Radulov

Post Syndicated from Boyko Radulov original https://aws.amazon.com/blogs/big-data/a-guide-to-airflow-worker-pool-optimization-in-amazon-mwaa/

Optimizing the Airflow worker pool configuration in Amazon Managed Workflows for Apache Airflow (Amazon MWAA), the AWS fully managed Apache Airflow service, is an important yet often overlooked strategy for scaling workflow operations. Tasks queued for longer periods can create the illusion that additional workers are the solution, when in reality the root cause might lie elsewhere. The decision to scale isn’t always straightforward. DevOps engineers and system administrators frequently face the challenge of determining whether adding more workers will solve their performance issues or only increase operational cost without addressing the root cause.

This post explores different patterns for worker scaling decisions in Amazon MWAA, focusing on the task pool mechanism and its relationship to worker allocation. By examining specific scenarios and providing a practical decision framework, this post helps you determine whether adding workers is the right solution for your performance challenges, and if so, how to implement this scaling effectively.

Main patterns

This section discusses the most frequently seen problems that raise the question if adding additional workers would improve the health of your environment.

High CPU

Airflow serves as a workflow management platform that coordinates and schedules tasks to be run on external processing services. It acts as a central orchestrator that can trigger and monitor tasks across various data processing systems like AWS Glue, AWS Batch, Amazon EMR, and other specialized data processing tools. Rather than processing data itself, Airflow’s strength lies in managing complex workflows and coordinating jobs between different systems and services.

In Analytics and Big Data environments, there is a prevalent misconception that saturated resources automatically warrant adding more capacity. However, for Amazon MWAA, understanding your workflow characteristics and optimization opportunities should precede scaling decisions.

As you scale up your workflows, resource utilization of the Airflow clusters naturally increases. When workers consistently operate at full capacity, it may seem intuitive to add additional compute resources. However, this approach often masks underlying inefficiencies rather than resolving them.

For example, in Amazon MWAA if you are running a single task that is consuming 100% of the available CPU on your Amazon MWAA worker, adding additional workers will not resolve the problem as the task is not optimized nor split into smaller parts. As such, increasing the number of minimum workers will not bring the expected effect but will only increase the operating costs.

When your Amazon MWAA workers are consistently running above 90% CPU or Memory utilization, you’ve reached a critical decision point. Before taking actions, it is essential to understand the root cause. You have three primary options:

Scale horizontally by adding additional workers to distribute the load.
Scale vertically by upgrading to a larger environment class for more resources per worker.
Optimize your DAGs and scheduling patterns to be more efficient and consume fewer resources.

Each approach addresses different underlying issues, and choosing the right path depends on identifying whether you are facing a capacity constraint, resource-intensive task design, or workflow inefficiency. For guidance on optimization strategies, please refer to Performance tuning for Apache Airflow on Amazon MWAA.

To monitor the CPUUtilization and MemoryUtilization on the workers, refer to the Accessing metrics in the Amazon CloudWatch console and choose the corresponding metrics.

Select a time window long enough to show usage patterns.
Set period to 1 Minute.
Set statistics to Maximum.

Long queue time

Sometimes Airflow tasks are stuck in a queued state for a long time, which prevents DAGs from completing on time.

In Amazon MWAA, each environment class comes with configured minimum and maximum worker nodes. Each worker provides a pre-configured concurrency, which is the number of tasks that can run simultaneously on each worker at any given time. The behavior is controlled through celery.worker_autoscale=(max,min).

For example, if you have minimum 4 mw1.small workers, with default Airflow configuration, you will be able to run 20 concurrent tasks (4 workers x 5 max_tasks_per_worker). If your system suddenly requires more than 20 tasks to execute concurrently, this will result in an autoscaling event. Amazon MWAA will decide how to scale your workers efficiently, and trigger the process. The autoscaling process, however, requires additional time to provision new workers resulting in additional tasks in queued status. To mitigate this queuing issue, consider the following:

If the CPU utilization on the workers is low, increasing the max value in celery.worker_autoscale=(max,min) can reduce the time tasks stay in queued state as each worker will be able to process more tasks concurrently. Airflow worker can take tasks up to the defined task concurrency regardless of the availability of its own system resources. As a result, the base worker may reach 100% CPU/Memory utilization before Autoscaling takes effect.
If you do not want to increase the task concurrency on the workers, increasing the minimum worker count can also be beneficial because having more available workers allows a higher number of tasks to run concurrently.

Scheduling delays

Adding new DAGs can not only affect your system resources, but it can also create uneven scheduling patterns. Some DAGs may experience delayed execution because of resource competition, even when the overall environment metrics appear healthy. This scheduling skew often manifests as inconsistent task pickup times, where certain workflows consistently wait longer in the queue while others execute promptly.

When Amazon CloudWatch metrics show increasing variance in task scheduling times, particularly during periods of high DAG activity, it signals the need for environment optimization. This scenario requires careful analysis of execution patterns and resource utilization to determine if:

While adding workers can help distribute the workload, this solution is most effective when the high utilization is primarily because of task execution load rather than DAG parsing or scheduling overhead. Adding more minimum workers will allow you to execute more tasks in parallel. For example, if you observe the value of AWS/MWAA/ApproximateAgeOfOldestTask to be steadily increasing, it means that the workers are not able to consume the messages from the queue fast enough. Additionally, you can also monitor the AWS/MWAA/QueuedTasks to identify similar patterns.
Upgrading the environment class would provide better scheduling capacity. If the Scheduler is showing signs of strain or if you’re seeing high resource utilization across all components, upgrading to a larger environment class might be the most appropriate solution. This provides more resources to both the Scheduler and Workers, allowing for better handling of increased DAG complexity and volume. To validate the same, use AWS/MWAA/CPUUtilization and AWS/MWAA/MemoryUtilization in the Cluster metrics and choose Scheduler, BaseWorker and AdditionalWorker metrics.
Restructuring DAG schedules would reduce resource contention.

The key is to understand your workflow patterns and identify whether the scheduling delays are because of insufficient worker capacity or other environmental constraints.

Anti patterns

This section showcases the most common anti patterns which make MWAA users think that adding more workers will improve performance.

Underutilized workers

When evaluating Amazon MWAA performance bottlenecks, it’s important to distinguish resource constraints and DAG design inefficiencies before scaling the environment.

Sometimes the Amazon MWAA environment has the capacity to run 100 tasks concurrently but your queue metrics (AWS/MWAA/RunningTasks) show only 20 tasks active most of the time with no tasks remaining in queued state. In such scenarios, you are advised to check Amazon CloudWatch for consistently low CPU and memory usage on existing workers during peak workload times. If this is confirmed, it is usually an indication of inefficiencies in DAG design, scheduling patterns, or Airflow configuration.

You have two primary options to address this:

1. Downsize: If you do not expect your workload to increase, it is safe to assume you have over-provisioned your cluster. Start by removing any extra workers first and finally resolve to downsizing your environment class.

2. Optimize: Fine tune your DAG scheduling and airflow configuration through Pools and Airflow configuration for concurrency to increase the throughput of your system.

Misconfigured Airflow configurations that create artificial bottlenecks

In Apache Airflow, performance bottlenecks often occur because of configuration settings, not actual resource constraints. At such times, DAG executions get delayed not because of insufficient compute, but because of incorrect concurrency configuration.

Efficient use of Amazon MWAA requires reviewing not only resource utilization for Workers and Schedulers but also concurrency configurations for artificially created bottlenecks. Sometimes one restrictive configuration prevents the scaling benefits of larger environment or additional workers. Always audit Airflow configurations if performance seems limited even when system metrics suggest spare capacity.

Important consideration: Amazon Managed Workflows for Apache Airflow (Amazon MWAA) does not automatically update the worker concurrency configuration when you change the environment class. This behavior is important to understand when scaling your environment. If you initially create an mw1.small environment, where each worker can handle up to 5 concurrent tasks by default. When you upgrade to a medium environment class (which supports 10 concurrent tasks per worker by default), the concurrency setting remains at 5 for in-place updated environments. You must manually update the concurrency configuration to take full advantage of the increased capacity available in the medium environment class.

Because of this you need to also update the Airflow configurations that control concurrency whenever you update the environment class. To update the concurrency setting after upgrading your environment class, modify the celery.worker_autoscale configuration in your Apache Airflow configuration options. This makes sure your workers can process the maximum number of concurrent tasks supported by your new environment class.

Other times, an Amazon MWAA environment can be constrained by max_active_runs or DAG concurrency controls instead of actual resource limits. These configuration-based throttles prevent tasks from running, even when the worker instances have available compute to handle the workload.

There is an important distinction between the two. Configuration limits act as artificial caps on parallelism, while true resource limits indicate that workers are fully utilizing their CPU or memory capacity. Understanding which type of constraint affects your environment helps you determine whether to adjust configuration settings or scale your infrastructure.

Adjusting Airflow configurations such as Pools, concurrency, max_active_runs solves performance problems without scaling workers. Some of the configurations you can use to control this behavior:

max_active_runs_per_dag (DAG level): Controls how many DAG runs for a given DAG are allowed at the same time. If set to 2, only 2 DAG runs can run concurrently, even if there is plenty of worker capacity left. Extra runs queue, making the DAG executions slow even though workers are idle.
max_active_tasks:Controls the concurrency field in a DAG definition (or setting at environment level) limits the number of tasks from the DAG running at any moment, regardless of overall system capacity or number of workers.
Pools:Pools restrict how many tasks of a certain type (often resource heavy) can run at once. A pool with only 3 slots will throttle any tasks above 3 assigned to that pool, leaving workers idle.
Execution timeouts and retries: If not tuned, failed tasks might fill up slots unnecessarily, stuck tasks can block worker slots and slow queue processing.
Scheduling intervals and dependencies: Overlapping or inefficient scheduling may cause idle periods or excess contention for resources, affecting real throughput.

How Airflow configurations can override each other

Airflow has multiple layers of concurrency and scheduling controls. Some at the environment level, some at the DAG/task level, and others for pools. Sometimes more restrictive settings override more permissive ones, resulting in unexpected queue buildup.

DAG level vs Environment level: If “max_active_runs_per_dag” (DAG level) is lower than the environment-level “max_active_runs_per_dag” or system wide concurrency, the DAG setting is used, throttling tasks even if the environment could do more.

Task level overrides: Individual task definitions can have their own parameters like “max_active_tis_per_dag” which can cap runs per task and create a bottleneck if set lower than global settings.

Order of precedence: The most restrictive relevant configuration at any level (Environment, DAG, Task) effectively sets the upper bound for parallel task execution.

Setting Location	Setting	Effect on task throughput
Environment Level	parallelism	Max total tasks running on Scheduler
DAG Level	max_active_runs	Max simultaneous DAG runs
Task Level	concurrency	Max concurrent task for that DAG

Performance issues often resemble resource exhaustion, but actually derive from overly restrictive configurations. Audit all the preceding parameters carefully. You can loosen restrictive values step by step and monitor their effect before deciding to scale your cluster further. This approach ensures optimal and cost-efficient usage of your cloud resources without paying for idle capacity.

Slow resource depletion from memory leaks

A common scenario for memory leak or slow resource depletion in Amazon MWAA is when DAGs and tasks begin to fail or slow down over time. Scaling workers or increasing environment size does not resolve the underlying issue. This happens because the root cause is not a lack of capacity but rather an application-level leak that causes persistent exhaustion.

For example, as Airflow continuously runs tasks and parses DAGs over time, memory consumption can steadily increase across the environment. This might manifest as an Amazon MWAA metadata database experiencing declining FreeableMemory metrics despite consistent or even reduced workloads. When this occurs, database query performance gradually declines as memory resources become constrained for scheduler/worker & metadata database, ultimately affecting overall environment responsiveness since Airflow depends heavily on its metadata database for critical operations. This scenario is similar to how an application might create database connections without properly closing them, leading to resource exhaustion over time.

Graph: Declining FreeableMemory and MemoryUtilization

Common causes:

Connection pool exhaustion: DAGs that fail to properly close database connections can lead to connection pool exhaustion and memory leaks in the database.
Resource-intensive operations: Complex, long-running queries or XCOM operations against the metadata database can consume excessive memory.
Inefficient DAG design: DAGs with numerous top-level Python calls can trigger database queries during DAG parsing. For instance, using variable.get() calls at the DAG level rather than at the task level creates unnecessary database load.

Recommended solutions:

Implement Amazon CloudWatch monitoring: Establish Amazon CloudWatch alarms for FreeableMemory with appropriate thresholds to detect issues early.
Regular database maintenance: Perform scheduled database clean-up operations to purge historical data that is no longer needed.
Optimize DAG code: Refactor DAGs to move database operations like variable.get() from the DAG level to the task level to reduce parsing overhead.
Connection management: Make sure all database connections are properly closed after use to prevent connection pool exhaustion.

By following the preceding recommendations you can maintain healthy memory utilization for the metadata database and maintain optimal performance of your Amazon MWAA environment without needing to scale workers.

Conclusion

The decision to add workers in Amazon MWAA environments requires careful consideration of multiple factors beyond simple task queue metrics. In this post, we showed that while adding workers can address certain performance challenges, it’s often not the optimal first response to system bottlenecks.

Key considerations before scaling workers include:

Root cause analysis
- Verify whether high CPU/memory usage stems from task optimization issues.
- Examine if queuing problems result from configuration constraints rather than resource limitations.
- Investigate potential memory leaks or resource depletion patterns.
Configuration optimization
- Review and adjust Airflow parameters (concurrency settings, pools, timeouts).
- Understand the interaction between different configuration layers.
- Optimize DAG design and scheduling patterns.

The most successful Amazon MWAA implementations follow a systematic approach: first optimizing existing resources and configurations, then scaling workers only when justified by data-driven capacity planning. This approach ensures cost-effective operations while maintaining reliable workflow performance.

Remember that worker scaling is only one tool in the Amazon MWAA optimization toolkit. Long-term success depends on building a comprehensive performance management strategy that combines proper monitoring, proactive capacity planning, and continuous optimization of your Airflow workflows.

In the next post, we discuss capacity planning and the steps you need to perform before adding additional DAGs in your environment so that you can plan for the additional load and make sure you have enough headroom.

To get started, visit the Amazon MWAA product page and the Performance tuning for Apache Airflow on Amazon MWAA page.

If you have questions or want to share your MWAA scaling experiences, leave a comment below.

About the authors

Implementing Kerberos authentication for Apache Spark jobs on Amazon EMR on EKS to access a Kerberos-enabled Hive Metastore

2026-04-13 Krishna Kumar Venkateswaran

Post Syndicated from Krishna Kumar Venkateswaran original https://aws.amazon.com/blogs/big-data/implementing-kerberos-authentication-for-apache-spark-jobs-on-amazon-emr-on-eks-to-access-a-kerberos-enabled-hive-metastore/

Many organizations run their Apache Spark analytics platforms on Amazon EMR on Amazon Elastic Compute Cloud (Amazon EC2), using Kerberos authentication to secure connectivity between Spark jobs and a centralized shared Apache Hive Metastore (HMS). With Amazon EMR on Amazon EKS, they gained a new option for running Spark jobs with the benefits of Kubernetes-based container orchestration, improved resource utilization, and faster job startup times. However, an HMS deployment supports only one authentication mechanism at a time. This means that they must configure Kerberos authentication for their Spark jobs on Amazon EMR on EKS to connect to the existing Kerberos-enabled HMS.

In this post, we show how to configure Kerberos authentication for Spark jobs on Amazon EMR on EKS, authenticating against a Kerberos-enabled HMS so you can run both Amazon EMR on EC2 and Amazon EMR on EKS workloads against a single, secure HMS deployment.

Overview of solution

Consider an enterprise data platform team that’s been running Spark jobs on Amazon EMR on EC2 for several years. Their architecture includes a Kerberos-enabled standalone HMS that serves as the centralized data catalog, with Microsoft Active Directory functioning as the Key Distribution Center (KDC). As the team evaluates Amazon EMR on EKS for new workloads, their existing HMS must continue serving Amazon EMR on EC2, with both authenticating through the same Kerberos infrastructure. To address this, the platform team must configure their Spark jobs running on Amazon EMR on EKS to authenticate with the same KDC. This is so they can obtain valid Kerberos tickets and establish authenticated connections to the HMS while maintaining a unified security posture across their data platform.

Scope of Kerberos in this solution

Kerberos authentication in this solution secures the connection between Spark jobs and the HMS. Other components in the architecture use AWS and Kubernetes security mechanisms instead.

Solution architecture

Our solution implements Kerberos authentication to secure the connection between Spark jobs and the HMS. The architecture spans two Amazon Virtual Private Clouds (Amazon VPCs) connected using VPC peering, with distinct components handling identity management, compute, and metadata services.

Identity and Authentication layer

A self-managed Microsoft Active Directory Domain Controller is deployed in a dedicated VPC and serves as the KDC for Kerberos authentication. The Active Directory server hosts service principals for both the HMS service and Spark jobs. This separate VPC deployment mirrors real-world enterprise architectures where Active Directory is typically managed by identity teams in their own network boundary, whether on-premises or in AWS.

Data Platform layer

The data platform components reside in a separate VPC and includes an EKS cluster that hosts both the HMS service and Amazon EMR on EKS based Spark jobs persisting data in an Amazon Simple Storage Service (Amazon S3) bucket.

Hive Metastore service

The HMS is deployed in the EKS hive-metastore namespace and simulates a pre-existing, standalone Kerberos-enabled HMS, a common enterprise pattern where HMS is managed independently of any data processing platform. You can learn more about other enterprise design patterns in the post Design patterns for implementing Hive Metastore for Amazon EMR or EKS. The HMS service authenticates with the KDC using its service principal and keytab mounted from a Kubernetes secret.

Apache Spark Execution layer

Apache Spark jobs are deployed using the Spark Operator on EKS. The Spark driver and executor pods are configured with Kerberos credentials through mounted ConfigMaps containing krb5.conf and jaas.conf, along with keytab files from Kubernetes secrets. When a Spark job must access Hive tables, the driver authenticates with the KDC and establishes a secure Simple Authentication and Security Layer (SASL) connection to the HMS.

Authentication flow

The HMS runs as a long-running Kubernetes service that must be deployed and authenticated before Spark jobs can connect.

During HMS deployment:

HMS pod validates its Kerberos configuration. krb5.conf and jaas.conf are mounted from ConfigMaps
Service authenticates with KDC using its principal hive/hive-metastore-svc.hive-metastore.svc.cluster.local@CORP.KERBEROS
keytab is mounted from Kubernetes secret for credential access
Secure Thrift endpoint is established on port 9083 with SASL authentication enabled

When a Spark job must interact with the HMS:

Spark job submission:
1. User submits Spark job through Spark Operator
2. Driver and executor pods are created with Kerberos configuration mounted as volumes
3. krb5.conf ConfigMap provides KDC connection details including realm and server addresses
4. jaas.conf ConfigMap specifies a login module configuration with keytab path and principal
5. Keytab secret contains encrypted credentials for Spark service principal spark/[email protected]
Authentication and connection:
1. Spark driver authenticates with KDC using its principal and keytab to obtain a Ticket Granting Ticket (TGT)
2. When connecting to HMS, Spark requests a service ticket from the KDC for the HMS principal hive/hive-metastore-svc.hive-metastore.svc.cluster.local@CORP.KERBEROS
3. KDC issues a service ticket encrypted with HMS’s secret key
4. Spark presents this service TGT to HMS over the Thrift connection on port 9083
5. HMS decrypts the ticket using its keytab, verifies Spark’s identity, and establishes the authenticated SASL session
6. Executor pods use the same configuration for authenticated operations
Data access:
1. Authenticated Spark job queries HMS for table metadata
2. HMS validates Kerberos tickets before serving metadata requests
3. Spark accesses underlying data in Amazon S3 using IRSA

Implementation workflow

The implementation involves three key stakeholders working together to establish the Kerberos-enabled communication:

Microsoft Active Directory Administrator

The Active Directory Administrator creates service accounts that are used for HMS and Spark jobs. This involves setting up the service principal names using the setspn utility and generating keytab files using ktpass for secure credential storage. The administrator configures the appropriate Active Directory permissions and Kerberos AES256 encryption type. Finally, the keytab files are uploaded to AWS Secrets Manager for secure distribution to Kubernetes workloads.

Data Platform Team

The platform team handles the Amazon EMR on EKS and Kubernetes configurations. They retrieve keytabs from Secrets Manager and create Kubernetes secrets for the workloads. They configure Helm charts for HMS deployment with Kerberos settings and set up ConfigMaps for krb5.conf, jaas.conf, and core-site.xml.

Data Engineering Operations

Data engineers submit jobs using the configured service account with Kerberos authentication. They monitor job execution and verify authenticated access to HMS.

Deploy the solution

In the remainder of this post, you will explore the implementation details for this solution. You can find the sample code in the AWS Samples GitHub repository. For additional details, including verification steps for each deployment stage, refer to the README in the repository.

Prerequisites

Before you deploy this solution, make sure that the following prerequisites are in place:

Access to a valid AWS account and permission to create AWS resources.
The AWS Command Line Interface (AWS CLI) is installed on your local machine.
Git, Docker, eksctl, kubectl, Helm, envsubst, jq, and yq utilities are installed on your local machine.
Familiarity with Kerberos, Apache Hive Metastore (HMS), Apache Spark, Kubernetes, Amazon EKS, and Amazon EMR on Amazon EKS.

Clone the repository and set up environment variables

Clone the repository to your local machine and set the two environment variables. Replace <AWS_REGION> with the AWS Region where you want to deploy these resources.

# Clone the Git repository
git clone https://github.com/aws-samples/sample-emr-eks-spark-kerberos-hms.git
cd sample-emr-eks-spark-kerberos-hms

# Set environment variables
export REPO_DIR=$(pwd)
export AWS_REGION=<AWS_REGION>

Setup Microsoft Active Directory infrastructure

In this section, we deploy a self-managed Microsoft Active Directory with KDC on a Windows Server EC2 instance into a dedicated VPC. This is an intentionally minimal implementation highlighting only the key components required for this blog post.

cd ${REPO_DIR}/microsoft-ad
./setup.sh

Setup EKS infrastructure

This section provisions the Amazon EMR on EKS infrastructure stack, including VPC, EKS cluster, Amazon Aurora PostgreSQL database, Amazon Elastic Container Registry (Amazon ECR), Amazon S3, Amazon EMR on EKS virtual clusters and the Spark Operator. Run the following script.

cd ${REPO_DIR}/data-infra
./setup.sh

Set up VPC peering

This section establishes network connectivity between the Active Directory VPC and EKS VPC for Kerberos authentication. Run the following script:

cd ${REPO_DIR}/vpc-peering
./setup.sh

Deploy Hive Metastore with Kerberos authentication

This section deploys a Kerberos-enabled HMS service on the EKS cluster. Complete the following steps:

Create Kerberos Service Principal for HMS service

cd ${REPO_DIR}/microsoft-ad/
# Create HMS service principal
./manage-ad-service-principals.sh create hive "hive/hive-metastore-svc.hive-metastore.svc.cluster.local"
# Verify the service principal was created
./manage-ad-service-principals.sh list

Deploy HMS service with Kerberos authentication

cd ${REPO_DIR}/hive-metastore
./deploy.sh

Set up Amazon EMR on Amazon EKS with Kerberos authentication

This section configures Spark jobs to authenticate with Kerberos-enabled HMS. This involves creating service principles for Spark jobs and generating the necessary configuration files. Complete the following steps:

Create Service Principal for Spark jobs

cd ${REPO_DIR}/microsoft-ad/
# Create Spark service principal
./manage-ad-service-principals.sh create spark "spark/analytics-team"
# Verify the service principal was created
./manage-ad-service-principals.sh list

Generate Kerberos configurations for Spark jobs

cd ${REPO_DIR}/spark-jobs/
./generate-spark-configs.sh --principal "spark/[email protected]" --namespace emr

Submit Spark jobs

This section verifies Kerberos authentication by running a Spark job that connects to the Kerberized HMS. Complete the following steps:

Submit the test Spark job

cd ${REPO_DIR}/spark-jobs
kubectl apply -f spark-job.yaml

Monitor job execution

# Watch the SparkApplication status
kubectl get sparkapplications -n emr -w
# Check pod status
kubectl get pods -n emr | grep "spark-kerberos"

Verify Kerberos authentication and HMS connection

# Check Spark driver logs for successful authentication
kubectl logs spark-kerberos-job-driver -n emr

The logs should confirm successful authentication, along with a listing of sample databases and tables.

Understanding Kerberos configuration

The HMS requires specific configuration parameters to enable Kerberos authentication, applied through the previously mentioned steps. The key configurations are outlined in the following section.

HMS configuration (metastore-site.xml)

The following configurations are added to metastore-site.xml file.

Setting	Value	Purpose
`hive.metastore.sasl.enabled`	true	Enable SASL authentication
`hive.metastore.kerberos.principal`	`hive/hive-metastore-svc.hive-metastore.svc.cluster.local@CORP.KERBEROS`	HMS service principal
`hive.metastore.kerberos.keytab.file`	`/etc/security/keytab/hive.keytab`	Keytab path

Hadoop security (core-site.xml)

The following configurations are added to the core-site.xml file.

Setting	Value
`hadoop.security.authentication`	kerberos
`hadoop.security.authorization`	true

Spark configuration

Setting	Value	Purpose
`spark.security.credentials.kerberos.enabled`	true	Enable Kerberos for Spark
`spark.hadoop.hive.metastore.sasl.enabled`	true	SASL for HMS connection
`spark.kerberos.principal`	`spark/[email protected]`	Spark service principal
`spark.kerberos.keytab`	`local:///etc/security/keytab/analytics-team.keytab`	Keytab path

Shared Kerberos files

Both HMS and Spark pods mount two common Kerberos configuration files: krb5.conf and jaas.conf, using ConfigMaps and Kubernetes secrets. The krb5.conf file is identical across both services and defines how each component connects to the KDC. The jaas.conf file follows the same structure but differs in the principal and keytab path for each service.

krb5 Configuration

[libdefaults]
	default_realm = CORP.KERBEROS
	dns_lookup_realm = false
	dns_lookup_kdc = false
	ticket_lifetime = 24h
	forwardable = true
	udp_preference_limit = 1
	default_tkt_enctypes = aes256-cts-hmac-sha1-96 aes128-cts-hmac-sha1-96
	default_tgs_enctypes = aes256-cts-hmac-sha1-96 aes128-cts-hmac-sha1-96
	permitted_enctypes = aes256-cts-hmac-sha1-96 aes128-cts-hmac-sha1-96

[realms]
	CORP.KERBEROS = {
		kdc = <ad-server-ip>
		admin_server = <ad-server-ip>
	}

[domain_realm]
	.corp.kerberos = CORP.KERBEROS
	corp.kerberos = CORP.KERBEROS

For more information, see the online documentation for krb5.conf.

JAAS configuration

Client {
 com.sun.security.auth.module.Krb5LoginModule required
 useKeyTab=true
 keyTab="/etc/security/keytab/hive.keytab"
 principal="hive/hive-metastore-svc.hive-metastore.svc.cluster.local@CORP.KERBEROS"
 useTicketCache=false
 storeKey=true
 debug=false;
};

Additional security considerations

This post focuses on core Kerberos authentication mechanics between Spark and HMS. We recommend two additional security hardening steps based on your organization’s security posture and compliance requirements.

Protecting Keytabs at Rest with AWS KMS Envelope Encryption

Keytabs stored as Kubernetes Secrets are only base64-encoded by default, not encrypted at rest. We recommend enabling EKS envelope encryption using an AWS Key Management Service (AWS KMS) customer managed key. With envelope encryption, secret data is encrypted with a Data Encryption Key (DEK), which is encrypted by your customer managed key. This protects keytab content even if the etcd datastore is compromised. To enable this on an existing EKS cluster:

aws eks associate-encryption-config \
  --cluster-name <your-cluster> \
  --encryption-config '[{"resources":["secrets"],"provider":{"keyArn":"arn:aws:kms:<region>:<account-id>:key/<key-id>"}}]'

Refer to the Amazon EKS documentation on envelope encryption for full setup guidance.

Encrypting the Thrift Data Channel with TLS

SASL with Kerberos provides mutual authentication but doesn’t automatically encrypt data over the Thrift connection. Many deployments default to auth QoP, leaving the data channel unencrypted. We recommend either:

Set SASL QoP to auth-conf — enables SASL-layer encryption using Kerberos session keys
Layer TLS over Thrift (preferred) — enables transport-level encryption using modern cipher suites

Enabling TLS on HiveServer2 / Hive Metastore Thrift:

<property>
  <name>hive.server2.use.SSL</name>
  <value>true</value>
</property>
<property>
  <name>hive.server2.keystore.path</name>
  <value>/etc/tls/keystore.jks</value>
</property>

Refer to the Hive SSL/TLS configuration documentation for full details.

Cleaning up

To avoid incurring future charges, clean up all provisioned resources during this setup by executing the following cleanup script.

cd ${REPO_DIR}/
./cleanup.sh

Conclusion

In this post, we demonstrated how to implement Kerberos authentication for Amazon EMR on EKS to securely connect to a Kerberos-enabled HMS. This solution addresses a common challenge faced by organizations with existing Kerberos-enabled HMS deployments who want to adopt Amazon EMR on EKS while maintaining their Kerberos-enabled security posture.

This pattern applies whether you’re migrating from on-premises Hadoop, running hybrid Amazon EMR on EC2 or Amazon EMR on EKS environments, or building a new cloud-native platform. Any scenario where Spark jobs on Kerberos must authenticate with a shared, Kerberos-enabled HMS.

You can use this post as a starting point to implement this pattern and extend it further to suit your organization’s data platform needs.

About the authors

Introducing enhancements to Amazon EMR Managed Scaling

2026-03-26 Amit Maindola

Post Syndicated from Amit Maindola original https://aws.amazon.com/blogs/big-data/introducing-enhancements-to-amazon-emr-managed-scaling/

Amazon EMR Managed Scaling has been helping customers automatically resize their clusters to optimize performance and reduce costs. We are excited to introduce a significant enhancement to this feature: Advanced Scaling for Amazon EMR. This new capability provides additional flexibility to configure the desired resource utilization or performance levels for your cluster using a utilization-performance slider. After the slider is set, EMR Managed Scaling intelligently scales the cluster and optimizes cluster resources based on your configured performance or resource utilization levels.

Customers appreciate the simplicity of EMR Managed Scaling, where they specify the minimum and maximum compute limits for their clusters and EMR Managed Scaling automatically resizes the cluster. EMR Managed Scaling continuously samples key metrics associated with the workloads running on clusters and scales up or down accordingly. However, customers’ workloads are increasingly getting more complex, with variability across dimensions such as data volumes and cost vs. SLA requirements. Consequently, customers prefer to have additional levers to tune the scaling behavior most suitable for their workload. In this post, we discuss the benefits of Advanced Scaling for Amazon EMR and demonstrate how it works through some example scenarios.

Advanced Scaling for Amazon EMR

Previously, customers who wanted to adjust the default EMR Managed Scaling behavior had no other option but to disable EMR Managed Scaling and create custom automatic scaling rules. Custom autoscaling rules created several problems:

Custom autoscaling rules are not shuffle-aware and shuffle data is lost.
Custom autoscaling is not aware of the application driver and can terminate it, failing the entire job.
Custom autoscaling can be slower to respond to real time needs.

These are some of the reasons why custom autoscaling is not the right fit. Customers wanted out-of-the-box support for Managed Scaling to handle the scaling that optimizes for the customers end goal to optimize cost or performance. The new Advanced Scaling capability enhances the existing benefits of EMR Managed Scaling by introducing additional controls and helping you configure the desired resource utilization or performance level for your cluster using a utilization-performance slider. EMR Advanced Scaling then internally translates intent into tailored algorithm strategy (UtilizationPerformanceIndex), such as how quickly to scale, how much to scale, and so on, to make scaling decisions for the cluster. This helps optimize cluster resources while making sure we meet the performance or resource utilization intent set by the customer.

For example, for a cluster running multiple tasks of relatively short duration (order of seconds), EMR Managed Scaling previously used to scale up the cluster aggressively and conservatively scale it down to avoid negative impact to job runtimes. Although this is the right approach for SLA-sensitive workloads, it might not be optimal for customers who are fine with little delay but prefers saving cost. Now, you can configure EMR Managed Scaling behavior suitable for your workload types, and we will apply tailored optimization to intelligently add or remove nodes from the clusters. This helps you achieve the optimal price-performance for your clusters along with increased flexibility of additional user-controls.

The value you set for Advanced Scaling optimizes your cluster to your requirements. Values range from 1-100. Supported values are 1, 25, 50, 75 and 100. If you set the index to values other than these, it results in a validation error. Scaling values map to resource-utilization strategies. The following list defines several of these:

Utilization optimized (1) – This setting prevents resource over provisioning. Use a low value when you want to keep costs low and to prioritize efficient resource utilization. It causes the cluster to scale up less aggressively. This works well for the use case when there are regularly occurring workload spikes and you don’t want resources to ramp up too quickly.
Balanced (50) – This balances resource utilization and job performance. This setting is suitable for steady workloads where most stages have a stable runtime. It’s also suitable for workloads with a mix of short and long-running stages. We recommend starting with this setting if you aren’t sure which to choose.
Performance optimized (100) – This strategy prioritizes performance. The cluster scales up aggressively to ensure that jobs complete quickly and meet performance targets. Performance optimized is suitable for service-level-agreement (SLA) sensitive workloads where fast run time is critical.

Customers can also choose intermediate values (25 and 75) for more nuanced control. The intermediate values available provide a middle ground between strategies to fine tune your cluster’s Advanced Scaling behavior.

Use cases and benefits

Amazon EMR’s Advanced Scaling feature improves cluster management by offering dynamic adaptation to diverse business requirements across industries. The feature enables strategic timing of scaling policies throughout the day, with early morning hours dedicated to workload preparation, peak business hours focusing on maximum performance, evening periods maintaining moderate scaling for post-business processing, and overnight hours optimized for cost-effective batch operations. This comprehensive approach allows organizations to fine-tune their resource allocation based on specific operational patterns, ultimately delivering an optimal balance between performance and cost-efficiency while ensuring business needs are met across different time zones and usage patterns.

Scaling configuration

In the following sections, we walk through a range of scenarios testing against a 3 TB TPC-DS dataset, then walk you through the results of testing a sample job. We wanted to evaluate how Amazon EMR would respond with advanced scaling policies in scenarios optimizing cluster utilization, balancing performance with utilization, and aggressive performance requirements.

With Advanced Scaling currently available through API and console support coming soon, we updated existing cluster configurations. We modified UtilizationPerformanceIndex with 1, 50, and 100, to correspond to the different scaling strategies using the put-managed-scaling-policy API with an advanced scaling strategy, as seen in the following examples:

Scenario 1: Utilization optimized

In this scenario, we used a utilization optimized configuration by setting UtilizationPerformanceIndex to 1:

aws emr put-managed-scaling-policy --cluster-id <'cluster-id'> \ 
  --managed-scaling-policy '{ 
  "ComputeLimits": { 
    "UnitType": "Instances", 
    "MinimumCapacityUnits": 2, 
    "MaximumCapacityUnits": 50, 
    "MaximumOnDemandCapacityUnits": 50, 
    "MaximumCoreCapacityUnits": 2 
  	}, 
  }'

The result of the test yielded a peak of 16 nodes running and 16 requested. The scale-up and scale-down process is conservative. It takes 15 minutes to completely release the nodes after the requested metric subsides, as shown in the following figure. The job completed in 12 minutes, 39 seconds. UtilizationPerformanceIndex of 1 or 25 can be useful when the cluster is running a sequence of jobs with little to zero idle time. It can prevent frequent node churn because nodes will be available for the next set of jobs.

Scenario 2: Balanced

In this scenario, we used a balanced configuration by setting UtilizationPerformanceIndex to 50:

aws emr put-managed-scaling-policy --cluster-id <'cluster-id'> \ 
  --managed-scaling-policy '{ 
  "ComputeLimits": { 
    "UnitType": "Instances", 
    "MinimumCapacityUnits": 2, 
    "MaximumCapacityUnits": 50, 
    "MaximumOnDemandCapacityUnits": 50, 
    "MaximumCoreCapacityUnits": 2 
  	}, 
  }'

The result of the test yielded a peak of 43 nodes running and 32 requested. UtilizationPerformanceIndex of 50 uses a balanced approach for scaling the resources. Nodes requested and running are higher such that you can get a better price-performance ratio. The job completed in 7 minutes, 1 second.

Scenario 3: Performance optimized

In this scenario, we used a performance optimized configuration by setting UtilizationPerformanceIndex to 100:

aws emr put-managed-scaling-policy --cluster-id <'cluster-id'> \ 
  --managed-scaling-policy '{ 
  "ComputeLimits": { 
    "UnitType": "Instances", 
    "MinimumCapacityUnits": 2, 
    "MaximumCapacityUnits": 50, 
    "MaximumOnDemandCapacityUnits": 50, 
    "MaximumCoreCapacityUnits": 2 
	  }, 
  }'

The result of the test yielded a peak of 50 nodes running and 46 requested. UtilizationPerformanceIndex of 100 delivers the highest performance by aggressively scaling resources up and down. You can expect the highest nodes requested and running in this configuration. Scale-down will closely follow the node requested metric and therefore can lead to frequent churn of nodes if there are short idle periods between job submissions. This setting is ideal for latency-sensitive workloads that need to finish under SLA. The example job completed in 6 minutes, 16 seconds.

Comparison

The following table summarizes the differences between these scaling methods and time taken for each.

Scaling Method	Utilization Index	Peak Total Nodes Requested	Peak Total Nodes Running	Job Run Time (Seconds)	Cost to Run job	Use Case
Scenario1 – Utilization optimized	1	16	16	759	Low	Workloads with regular spikes; prioritizes cost efficiency with conservative scaling
Scenario 2 – Balanced	50	32	43	421	Medium	Steady workloads with mixed stage durations; recommended starting point
Scenario 3 – Performance Optimized	100	46	50	376	High	SLA-sensitive workloads requiring fast completion times

Advanced Managed Scaling in Amazon EMR introduces a more nuanced approach to cluster management through the customized scaling strategies to meet your business requirements. This spectrum offers fine-grained control over how clusters respond to workload demands. At one end, with a utilization optimized configuration of 1, the system prioritizes efficient resource usage, scaling up conservatively to maintain cost-effectiveness and taking advantage of existing cluster resources. In the balanced configuration at 50, the strategy aims to strike an equilibrium between resource utilization and job performance. To meet performance SLAs, the performance optimized value of 100 showed aggressive scaling responding to increased demand for resources quickly, regardless of resource consumption. This granular control helps you fine-tune your cluster’s behavior based on your specific needs, balancing cost, efficiency, and performance.

Conclusion

To Summarize, Advanced Scaling for Amazon EMR represents an advancement in cluster management, offering greater control and efficiency. By fine-tuning your clusters’ behavior, you can achieve more cost-effective and performant big data processing. We encourage you to try this new feature and discover how it can optimize your EMR workloads. Start by experimenting with different UtilizationPerformanceIndex values and closely monitor your cluster’s performance and cost metrics. Over time, you will be able to find the perfect balance that meets your specific needs.

To learn more about Amazon EMR Managed Scaling and Advanced Scaling, refer to our documentation. We’re excited to see how you use this new capability to enhance your big data processing on AWS, and we look forward to your feedback as we continue to evolve and improve our services.

About the authors

Building a scalable, transactional data lake using dbt, Amazon EMR, and Apache Iceberg

2026-03-18 Umesh Pathak

Post Syndicated from Umesh Pathak original https://aws.amazon.com/blogs/big-data/building-a-scalable-transactional-data-lake-using-dbt-amazon-emr-and-apache-iceberg/

Growing data volume, variety, and velocity has made it crucial for businesses to implement architectures that efficiently manage and analyze data, while maintaining data integrity and consistency. In this post, we show you a solution that combines Apache Iceberg, Data Build Tool (dbt), and Amazon EMR to create a scalable, ACID-compliant transactional data lake. You can use this data lake to process transactions and analyze data simultaneously while maintaining data accuracy and real-time insights for better decision-making.

Challenges, business imperatives, and technical advantages

Traditional data lakes have long struggled with fundamental limitations. For example, the lack of ACID compliance, data inconsistencies from concurrent writes, complex schema evolution, and the absence of time travel, rollback, and versioning capabilities. These shortcomings directly conflict with growing business demands for concurrent read/write support, robust data versioning and auditing, schema flexibility, and transactional capability within data lake environments. To address these gaps, modern solutions use ACID transactions at scale, optimized storage formats through Apache Iceberg, version control for data on Amazon Simple Storage Service (Amazon S3), and cost-effective, streamlined maintenance—delivering a reliable, enterprise-grade data lake architecture that meets both operational and analytical needs.

Solution overview

The solution is built around four tightly integrated layers that work together to deliver a scalable, transactional data lake.

Raw data is ingested and stored in Amazon S3, which serves as the foundational storage layer. This layer supports multiple data formats and enables efficient data partitioning through Apache Iceberg’s table format. This ensures that data is organized and accessible from the moment it lands. Then, Amazon EMR takes over as the distributed computing engine, using Apache Spark to process large-scale datasets in parallel, handling the heavy lifting of reading, transforming, and writing data across the lake.

Sitting within the processing layer, dbt drives the transformation logic. It applies SQL-based, version-controlled transformations that convert raw, unstructured data in the S3 raw layer into clean, curated datasets stored back in S3. This maintains ACID compliance and schema consistency throughout.

Finally, the curated data is available for consumption through Amazon Athena, which provides a serverless, one-time querying capability directly on S3. With this, analysts and business users can run interactive SQL queries without managing any infrastructure. Together, these components form a continuous pipeline: data flows from ingestion through distributed processing and structured transformation, ultimately surfacing as reliable, query-ready insights.

Amazon EMR is a cloud-based big data service that streamlines the deployment and management of open source frameworks like Apache Spark, Hive, and Trino. It provides a managed Apache Hadoop environment that organizations can use to process and analyze vast amounts of data efficiently.

Data Build Tool is an open source tool that data teams can use to transform and model data using SQL. It promotes best practices for data modeling, testing, and documentation, streamlining maintenance and collaboration on data pipelines.

Apache Iceberg is an open table format designed for large-scale analytics on data lakes. It supports features like transactions, time travel, and data partitioning, which are essential for building reliable and performant data lakes. By using Iceberg, organizations can maintain data integrity and enable efficient querying and processing of data.

When combined, these three technologies provide a powerful solution for building transactional data lakes. Amazon EMR provides the scalable and managed infrastructure for running big data workloads, dbt enables efficient data modeling and transformation, and Apache Iceberg provides data consistency and reliability within the data lake.

Prerequisites

Before proceeding with the solution walkthrough, make sure that the following are in place:

AWS Account – An active AWS account with sufficient permissions to create and manage EMR clusters, S3 buckets, Athena workgroups, and AWS Glue Data Catalog resources
IAM Roles – The following IAM roles must exist and have appropriate permissions:
- EMR_DefaultRole – Service role for Amazon EMR
- EMR_EC2_DefaultRole – Amazon Elastic Compute Cloud (Amazon EC2) instance profile for EMR nodes
AWS Command Line Interface (AWS CLI) – Installed and configured with credentials for your target AWS account and AWS Region (refer to Step 1.1 for setup instructions)
Python 3.8+ – Installed on your local machine or workspace for setting up the dbt virtual environment
Pip – Python package manager available for installing dbt and its dependencies
Git – Installed on the EMR primary node or local environment for version control and dbt package management
Amazon Athena – Athena query editor access with a configured S3 output location for query results
AWS Glue Data Catalog – Enabled as the metastore for Amazon EMR and Athena (no additional setup required if using the default AWS Glue integration)
S3 Bucket Naming – Prepare a unique identifier to suffix S3 bucket names, ensuring global uniqueness across all three buckets created in Step 1.3
Network Access – Make sure that your local machine can reach the Amazon EMR primary node’s DNS over port 10001 (Thrift/HiveServer2) for dbt connectivity; configure security groups accordingly

Solution walkthrough

Step 1: Environment setup

Install the AWS CLI on your workspace by following the instructions in Installing or updating the latest version of the AWS CLI. To configure AWS CLI interaction with AWS, refer to Quick setup.

Create EMR cluster.

Create the following JSON file with the following contents emr-config.json:

[
  {
    "Classification": "iceberg-defaults",
    "Properties": {
      "iceberg.enabled": "true"
    }
  },
  {
    "Classification": "spark-hive-site",
    "Properties": {
      "hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
    }
  }
]

Run the following command on your AWS CLI, updating the preferred AWS Region:

aws emr create-cluster \
--name "Iceberg-DBT-Cluster" \
--release-label emr-7.7.0 \
--applications Name=Spark Name=Hive Name=Livy \
--ec2-attributes InstanceProfile=EMR_EC2_DefaultRole \
--instance-type c3.4xlarge \
--instance-count 1 \
--service-role EMR_DefaultRole \
--configurations file://emr-config.json \
--region [region]

Set up S3 buckets.
Create the following S3 bucket using the AWS CLI after updating the bucket name.

aws s3 mb s3://amzn-s3-demo-transactional-datalake-raw-[unique-identifier]
aws s3 mb s3://amzn-s3-demo-transactional-datalake-curated-[unique-identifier]
aws s3 mb s3://amzn-s3-demo-transactional-datalake-analytics-[unique-identifier]

Step 2. Raw layer implementation

The raw layer serves as the foundation of our data lake, ingesting and storing data in its original form. This layer is important for maintaining data lineage and enabling reprocessing if needed. We use Apache Iceberg tables to store our raw data, which provides benefits such as ACID transactions, schema evolution, and time travel capabilities.

In this step, we create a dedicated database for our raw data and set up tables for customers, products, and sales using Amazon Athena. These tables are configured to use the Iceberg table format and are compressed using the ZSTD algorithm to optimize storage. The LOCATION property specifies where the data will be stored in Amazon S3 so that data is organized and accessible.

After creating the tables, we insert sample data to simulate real-world scenarios. We use this data throughout the rest of the implementation to demonstrate the capabilities of our data lake architecture.

Update the respective bucket name in each create table bucket name from the previous step:

Create database and tables

-- Create Raw Database
CREATE SCHEMA raw_sales_analytics_data_layer;

-- Create Customers Table
CREATE TABLE raw_sales_analytics_data_layer.customers (
    CustomerID string,
    CustomerName string,
    Region string,
    inserted_timestamp timestamp
)
LOCATION 's3://[bucket_name]/raw_sales_analytics_data_layer/customers'
TBLPROPERTIES (
    'table_type'='iceberg', 
    'write_compression'='zstd'
);

-- Create Products Table
CREATE TABLE raw_sales_analytics_data_layer.products (
    productid string,
    productname string,
    category string,
    supplier string,
    inserted_timestamp timestamp
)
LOCATION 's3://[bucket_name]/raw_sales_analytics_data_layer/products'
TBLPROPERTIES (
    'table_type'='iceberg', 
    'write_compression'='zstd'
);

Insert sample data

-- Insert Customers
INSERT INTO raw_sales_analytics_data_layer.customers
VALUES 
    ('201', Jane Doe', 'Central', current_timestamp),
    ('202', Arnav Desai, 'North', current_timestamp),
    ('203', Kwaku Mensah, 'West', current_timestamp);

-- Insert Products
INSERT INTO raw_sales_analytics_data_layer.products
VALUES
    ('1', 'Laptop', 'Electronics', 'AnyAuthority', current_timestamp),
    ('2', 'Smartphone', 'Electronics', 'AnyCompany', current_timestamp);

-- Insert Sales
INSERT INTO raw_sales_analytics_data_layer.sales
VALUES
    ('ORD001', '1', '201', '2025-04-01', 1299.99, current_timestamp),
    ('ORD002', '2', '202', '2025-04-02', 899.99, current_timestamp);

Step 3: dbt setup and configuration

Setting up dbt involves installing the necessary packages, configuring the connection to the data warehouse (in this case, Amazon EMR), and setting up the project structure.

We start by creating a Python virtual environment to isolate our dbt installation. Then, we install dbt-core and the Spark adapter, which allows dbt to connect to the EMR cluster. The profiles.yml file is configured to connect to the EMR cluster using the Thrift protocol, while the dbt_project.yml file defines the overall structure of the dbt project, including model materialization strategies and file formats.

Install prerequisites

# Create Python virtual environment
python -m venv dbt-env
source dbt-env/bin/activate

# Install required packages
pip install dbt-core dbt-spark[PyHive]

# Install git
yum install git

Configure dbt profiles

# ~/.dbt/profiles.yml
sales_analytics:
  target: dev
  outputs:
    dev:
      type: spark
      method: thrift
      host: your-emr-master-dns
      port: 10001
      schema: curated_sales_analytics_data_layer
      threads: 4

Project configuration

# dbt_project.yml
name: 'sales_analytics'
version: '1.0.0'
config-version: 2

profile: 'sales_analytics'

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]

target-path: "target"
clean-targets:
    - "target"
    - "dbt_packages"

models:
  sales_analytics:
    dim:
      +materialized: table
      +file_format: iceberg
    ads:
      +materialized: table
      +file_format: iceberg

Step 4: dbt models implementation

In this step, we implement dbt models, which define the transformations that we will apply to raw data. We start by configuring data sources in the sources.yml file, which allows dbt to reference raw tables easily.

We then create dimension models for customers and products, and a fact model for sales.

These models use incremental materialization strategies to efficiently update data over time. The incremental strategy processes only new or updated records, significantly reducing the time and resources required for each run.

Source configuration

# models/sources.yml
version: 2
sources:
  - name: raw_sales
    database: raw_sales_analytics_data_layer
    schema: raw_sales_analytics_data_layer
    tables:
      - name: customers
        columns:
          - name: CustomerID
            tests:
              - unique
              - not_null
      - name: products
      - name: sales

Dimension models

-- models/dim/dim_customers.sql
{{ config(
    materialized='incremental',
    unique_key='customerid',
    incremental_strategy='merge'
) }}

WITH source_data AS (
    SELECT 
        customerid,
        customername,
        region,
        inserted_timestamp,
        ROW_NUMBER() OVER (
            PARTITION BY customerid 
            ORDER BY inserted_timestamp DESC
        ) as row_number
    FROM {{ source('raw_sales_analytics_data_layer', 'customers') }}
    {% if is_incremental() %}
    WHERE inserted_timestamp > (SELECT MAX(inserted_timestamp) FROM {{ this }})
    {% endif %}
)

SELECT 
    customerid,
    customername,
    region,
    inserted_timestamp
FROM source_data
WHERE row_number = 1

Product models

-- models/dim/dim_products.sql
{{ config(
    materialized='incremental',
    unique_key='productid',
    incremental_strategy='merge'
) }}

WITH source_data AS (
    SELECT
        productid,
        productname,
        category,
        supplier,
        inserted_timestamp,
        ROW_NUMBER() OVER (
            PARTITION BY productid
            ORDER BY inserted_timestamp DESC
        ) as row_number
    FROM {{ source('raw_sales_analytics_data_layer', 'products') }}
    {% if is_incremental() %}
    WHERE inserted_timestamp > (SELECT MAX(inserted_timestamp) FROM {{ this }})
    {% endif %}
)

SELECT
    s.productid,
    s.productname,
    s.category,
    s.supplier,
    s.inserted_timestamp
FROM source_data s
WHERE s.row_number = 1
{% if is_incremental() %}
    AND NOT EXISTS (
        SELECT 1
        FROM {{ this }} t
        WHERE t.productid = s.productid
        AND t.inserted_timestamp >= s.inserted_timestamp
    )
{% endif %}

Fact models

-- models/dim/fact_sales.sql
{{ config(
    materialized='incremental',
    unique_key='orderid',
    incremental_strategy='merge'
) }}

WITH source_data AS (
    SELECT
        orderid,
        productid,
        customerid,
        date,
        salesamount,
        inserted_timestamp,
        ROW_NUMBER() OVER (
            PARTITION BY orderid
            ORDER BY inserted_timestamp DESC
        ) as row_number
    FROM {{ source('raw_sales_analytics_data_layer', 'sales') }}
    {% if is_incremental() %}
    WHERE orderid NOT IN (SELECT orderid FROM {{ this }})  -- Changed condition
    {% endif %}
)

SELECT
    s.orderid,
    s.productid,
    s.customerid,
    s.date,
    s.salesamount,
    s.inserted_timestamp
FROM source_data s
WHERE s.row_number = 1

Step 5: Analytics layer

The analytics layer builds upon dimension and fact models to create more complex analyzes. In this step, we create a daily sales analysis model that combines data from fact_sales, dim_customers, and dim_products models.

We also implement a customer insights model that analyzes purchase patterns across different Regions and product categories.

These analytics models demonstrate how we can use our transformed data to generate valuable business insights. By materializing these models as Iceberg tables, we make sure that they benefit from the same ACID transactions and time travel capabilities as our raw and transformed data.

Daily sales analysis

The analytics layer introduces a fact_sales_analysis model that consolidates transactional sales data with customer and product dimensions to enable business-ready reporting. Built as an incremental model with a merge strategy, it efficiently processes data by deduplicating records using the latest inserted timestamp per order, enabling reliable downstream consumption without full table refreshes.

-- models/ads/fact_sales_analysis.sql
{{ config(
    materialized='incremental',
    unique_key='orderid',
    incremental_strategy='merge'
) }}

WITH source_data AS (
    SELECT
        s.orderid,
        s.date,
        s.salesamount,
        c.customername,
        c.region,
        p.productname,
        p.category,
        p.supplier,
        s.inserted_timestamp,
        ROW_NUMBER() OVER (
            PARTITION BY s.orderid
            ORDER BY s.inserted_timestamp DESC
        ) as row_number
    FROM {{ ref('fact_sales') }} s
    JOIN {{ ref('dim_customers') }} c ON s.customerid = c.customerid
    JOIN {{ ref('dim_products') }} p ON s.productid = p.productid
    {% if is_incremental() %}
    WHERE s.orderid NOT IN (SELECT orderid FROM {{ this }})
    {% endif %}
)

SELECT
    s.orderid,
        s.date,
        s.salesamount,
        s.customername,
        s.region,
        s.productname,
        s.category,
        s.supplier,
        s.inserted_timestamp
FROM source_data s
WHERE s.row_number = 1

Customer insights

The customer_purchase_patterns model aggregates sales activity across customer Regions and product categories to surface revenue trends and buying behavior. Materialized as an Iceberg table in the analytics schema, it provides a performant and scalable foundation for customer segmentation, Regional performance analysis, and category-level revenue attribution.

-- models/analytics/customer_purchase_patterns.sql
{{
    config(
        materialized='table',
        file_format='iceberg',
        schema='analytics'
    )
}}

SELECT
    dc.Region,
    dp.category,
    COUNT(DISTINCT fs.orderid) as total_orders,
    COUNT(DISTINCT dc.customerid) as unique_customers,
    SUM(fs.salesamount) as total_revenue,
    SUM(fs.salesamount) / COUNT(DISTINCT dc.customerid) as revenue_per_customer
FROM {{ ref('fact_sales') }} fs
JOIN {{ ref('dim_customers') }} dc ON fs.customerid = dc.customerid
JOIN {{ ref('dim_products') }} dp ON fs.productid = dp.productid
GROUP BY dc.Region, dp.category

Step 6: Transactional operations and time travel with Apache Iceberg

This section demonstrates how to use Apache Iceberg’s time travel capabilities and transactional operations using actual snapshot data from our dim_customers table. We walk through querying data at different points in time and comparing changes between snapshots.

Transactional capabilities
Let’s first look at current data:

Now, modify the raw layer data for customerid 201 and change the Region to East

Run the dbt model for dim_customers to sync the changes

Validate the data in curated layer for dim_customers dimension table
Time-travel capabilities
First, let’s fetch snapshots for customers dimension table in curated layer

Now, find the data state before and after the modification.

Step 7: Data quality tests

Data quality is a critical pillar of any reliable data pipeline. In this step, we define and enforce quality checks directly within the dbt project using schema-level test configurations. Rather than relying on one-time validation scripts, with dbt’s built-in testing framework, we can declaratively specify expectations on our models, ensuring that key fields remain unique, non-null, and consistent across the data layer before they reach downstream consumers.

Generic tests configuration

The schema.yml file serves as the central contract for model integrity. Here, we apply generic tests on the fact_sales and dim_customers models to catch data anomalies early in the pipeline.

# models/schema.yml
version: 2

models:
  - name: fact_sales
    columns:
      - name: orderid
        tests:
          - unique
          - not_null
      - name: salesamount
        tests:
          - not_null

  - name: dim_customers
    columns:
      - name: customerid
        tests:
          - unique
          - not_null

Step 8: Maintenance procedures

A well-functioning data pipeline requires ongoing maintenance to remain performant and auditable over time. This step covers two essential practices, table optimization to keep data storage efficient, and snapshot management to track historical changes in source data. Together, these procedures keep the pipeline reliable, cost-effective, and capable of supporting time-based analysis.

Table optimization
As data accumulates in Delta or Iceberg tables, small files and fragmented storage can degrade query performance. The optimize_table macro provides a reusable utility to run Databricks’ OPTIMIZE command on any target table, consolidating small files and improving read efficiency without manual intervention.
```
-- macros/optimize_table.sql
{% macro optimize_table(table_name) %}
    {% set query %}
        OPTIMIZE {{ table_name }}
    {% endset %}
    {% do run_query(query) %}
{% endmacro %}
```

Snapshot management

To maintain a historical record of customer data changes, we use dbt snapshots with a timestamp-based strategy. The customers_snapshot model captures row-level changes from the raw source layer and persists them in a dedicated snapshots schema, enabling point-in-time analysis and audit trails.

-- snapshots/customer_snapshot.sql
{% snapshot customers_snapshot %}
{{
    config(
      target_schema='snapshots',
      unique_key='CustomerID',
      strategy='timestamp',
      updated_at='inserted_timestamp'
    )
}}

SELECT * FROM {{ source('raw_sales_analytics_data_layer', 'customers') }}

{% endsnapshot %}

Step 9: Monitoring and logging

Observability is an essential aspect of any production-grade data pipeline. This step establishes logging and monitoring practices within the dbt project to track pipeline runs, capture errors, and support debugging. With structured logging enabled, teams gain visibility into model execution, test results, and runtime behavior, streamlining issue diagnosis and maintaining operational confidence.

dbt logging configuration
The dbt_project.yml logging configuration directs dbt to write logs to a dedicated path and outputs them in JSON format. JSON-structured logs are particularly useful for integration with log aggregation tools and monitoring dashboards, enabling automated alerting and audit trail management.
```
# dbt_project.yml
logs:
  path: logs
  enable_json: true
```

Step 10. Deployment and running

With the pipeline fully built, tested, and maintained, the final step covers how to deploy and execute dbt models across different scenarios. Whether running a complete refresh, processing incremental updates, or validating data quality, these commands form the operational backbone of day-to-day pipeline management.

Full refresh
A full refresh rebuilds all models from scratch, reprocessing the entire dataset. This is typically used after significant schema changes, backfills, or when incremental state needs to be reset.
```
dbt run --full-refresh
```
Incremental update
For routine pipeline runs, incremental updates process only new or changed data, significantly reducing compute time and cost. The following command targets specific models (dim_customers and fact_sales) allowing selective execution without triggering the full DAG.
```
dbt run --select dim_customers fact_sales
```
Testing
After models are run, data quality tests defined in the schema configuration are executed to validate integrity across all models. This validates that constraints such as uniqueness and non-null checks are met before data reaches downstream consumers.
```
dbt test
```

Step 11. Cleanup

Infrastructure cleanup

# Delete EMR cluster
aws emr terminate-clusters --cluster-id <cluster-id>

# Remove S3 buckets
aws s3 rb s3://amzn-s3-demo-transactional-datalake-raw-bucket-[unique-identifier] --force
aws s3 rb s3://amzn-s3-demo-transactional-datalake-curated-bucket-[unique-identifier] --force
aws s3 rb s3://amzn-s3-demo-transactional-datalake-analytics-bucket-[unique-identifier] --force

Database cleanup

DROP SCHEMA raw_sales_analytics_data_layer CASCADE;
DROP SCHEMA curated_sales_analytics_data_layer CASCADE;

Conclusion

In this post, you learned how to build a transactional data lake on Amazon EMR using dbt and Apache Iceberg, from environment setup and modeling raw data, to quality enforcing, snapshot management, and incremental pipeline deployment. The architecture brings together the scalability of Amazon EMR, dbt’s transformation capabilities, and Iceberg’s ACID-compliant table format to deliver a reliable, maintainable, and cost-efficient data platform.

To get started, see the Amazon EMR documentation to deploy this architecture in your own environment. Whether you’re modernizing a legacy data platform or building a new analytics foundation, this stack gives you the flexibility to scale with confidence.

About the authors

Reducing costs for shuffle-heavy Apache Spark workloads with serverless storage for Amazon EMR Serverless

2026-03-10 Praveen Mohan Prasad

Post Syndicated from Praveen Mohan Prasad original https://aws.amazon.com/blogs/big-data/reducing-costs-for-shuffle-heavy-apache-spark-workloads-with-serverless-storage-for-amazon-emr-serverless/

At re:Invent 2025, we announced serverless storage for Amazon EMR Serverless, eliminating the need to provision local disk storage for Apache Spark workloads. Serverless storage of Amazon EMR Serverless reduces data processing costs by up to 20% while helping prevent job failures from disk capacity constraints.

In this post, we explore the cost improvements we observed when benchmarking Apache Spark jobs with serverless storage on EMR Serverless. We take a deeper look at how serverless storage helps reduce costs for shuffle-heavy Spark workloads, and we outline practical guidance on identifying the types of queries that can benefit most from enabling serverless storage in your EMR Serverless Spark jobs.

Benchmark results for EMR 7.12 with serverless storage against standard disks

We conducted the performance and cost savings benchmarking using the TPC-DS dataset at 3TB scale, running 100+ queries that included a mix of high and low shuffle operations. The test configuration utilized Dynamic Resource Allocation (DRA) with no pre-initialized capacity. The system was set up with 20GB of disk space, and Spark configurations included 4 cores and 14GB memory for both driver and executor, with dynamic allocation starting at 3 initial executors (spark.dynamicAllocation.initialExecutors = 3). A comparative analysis was performed between local disk storage and serverless storage configurations. The aim was to assess both total and average cost implications between these storage approaches.

The following table and chart compare the cost reduction we observed in the testing environment described above. Based on us-east-1 pricing, we saw a cost savings of more than 26% when using serverless storage.

	Shuffle
	serverless storage	standard Disks	savings
Total Cost ($)	24.28	33.1	26.65%
Average Cost ($)	0.233	0.318	26.73%

Average cost comparison between standard disks and serverless storage

% Relative savings (per query) of serverless storage compared to standard disk shuffle

In this testing, we observed that serverless storage in EMR Serverless reduces cost for approximately 80% of TPC-DS queries. For the queries where it provides benefits, it delivers an average cost saving of approximately 47%, with savings of up to 85%. Queries that regress typically have low shuffle intensity, maintain high parallelism throughout execution, or complete quickly enough that executor scale-down opportunities are minimal. The following figure shows the percentage cost difference for each of the TPC-DS queries when serverless storage was enabled, compared to the baseline configuration without serverless storage. Positive values indicate cost savings (higher is better), while negative values indicate cost regressions.

Percentage cost savings per TPC-DS query with serverless storage enabled

Percentage cost savings per TPC-DS query with serverless storage enabled

Runtime comparison

There is significant cost savings due to the increased elasticity from terminating executors earlier. However, job completion time may increase because the shuffle data is stored in serverless storage rather than locally on the executors. The additional read and write latency for shuffle data contributes to the longer runtime. The following table and chart show the runtime comparison, we observed in our testing environment.

	Shuffle
	serverless storage	standard disks	runtime
Total Duration (sec)	6770.63	4908.52	-37.94%
Average Duration (sec)	65.1	47.2	-37.92%

Runtime comparison

Storing shuffle externally and decoupling from the compute allowed the flexibility for EMR Serverless to turn off unused resources dynamically as the state info has been offloaded from the compute. However, these cost savings can be realized only when DRA is on. If DRA is turned off, Spark would keep those unused resources alive adding to the total cost.

Query patterns that benefit from serverless storage

The cost savings from serverless storage depend heavily on how executor demand changes across stages of a job. In this section, we examine common execution patterns and explain which query shapes are most likely to benefit from serverless storage of EMR Serverless and which query patterns may not benefit from shuffle externalization.

Inverted triangle pattern queries

In order to understand why externalizing the shuffle data can allow such a significant cost savings, consider a simplified query. The following query calculates annual total sales from the TPC-DS dataset by joining the store_sales and date_dim tables, summing the sales amounts per year, and ordering the results.

SELECT d_year, SUM(ss_net_paid) AS total_sales
FROM store_sales
JOIN date_dim ON store_sales.ss_sold_date_sk = date_dim.d_date_sk
GROUP BY d_year
ORDER BY d_year;

This query demonstrates that high executor demand during the map phase and low executor demand in the reduce phase is an aggregation query with a high cardinality input and a low cardinality group by.

Stage 1 (High Executor Demand)

The join and read steps scan the entire store_sales and date_dim tables. This often involves billions of rows in large-scale TPC-DS datasets, so Spark will try to parallelize the scan across many executors to maximize read throughput and compute efficiency.

Stage 2 (Low Executor Demand)

The aggregation is on d_year, which typically has few unique values, such as only a handful of years in the data. This means after the shuffle stage, the reduce phase combines the partial aggregates into a number of keys equal to the number of years (often < 10). Only a few Spark tasks are needed to finish the final aggregation, so most executors become idle.

With shuffle information stored on the local disk, the compute resources associated with these idle executors would still be running in order to keep the shuffle data available. With shuffle data offloaded from the nodes running the executors, with DRA enabled, those nodes with idle executors get released immediately.

Because early stages process high-cardinality inputs and later stages collapse data into a small number of keys, these queries form an “inverted triangle” execution pattern: wide parallelism at the top and narrow parallelism at the bottom as shown in the following image:

Inverted triangle pattern queries

Hourglass pattern queries

Depending upon the complexity of the job, there can be multiple stages with varying demand on number of executors needed for the stage. Such jobs can benefit from greater elasticity obtained by offloading shuffle data to external serverless storage. One such pattern is the hour glass pattern. The following figure shows a workload pattern where executor demand expands, contracts during shuffle-heavy stages, and expands again. Serverless storage of EMR Serverless decouples shuffle data from compute, enabling more efficient scale-down during narrow stages and helping improve cost optimization for elastic workloads.

Hourglass pattern in Spark stage execution

Hourglass pattern queries

To identify queries of this category, consider the following example, The query progresses through three stages:

Stage 1: The initial join and filter between store_sales and item produces a wide, high-cardinality intermediate dataset, requiring high parallelism (many executors).
Stage 2: Aggregation groups by a small set of categories such as “Home” or “Electronics”, resulting in a drastic drop in output partitions. So this stage efficiently runs with only a few executors, as there’s little data to parallelize.
Stage 3: The small result is joined (usually a broadcast join) back to a large fact table with a date dimension, again producing a large result that is well-parallelized, causing Spark to ramp up executor usage for this stage.

WITH stage1_large_scan AS (
-- Stage 1: Scan and wide join generates lots of parallelism and needs many executors
SELECT ss_item_sk, ss_sold_date_sk, ss_net_paid, i_category
FROM store_sales
JOIN item ON store_sales.ss_item_sk = item.i_item_sk
WHERE item.i_category IN ('Home', 'Electronics')
),
stage2_small_agg AS (
-- Stage 2: Aggregate on low-cardinality column (by category), reducing to few groups, so few executors needed
SELECT i_category, SUM(ss_net_paid) AS total_cat_sales
FROM stage1_large_scan
GROUP BY i_category
),
stage3_broadcast_filter AS (
-- Stage 3: Join back to high-cardinality table, pushing parallelism up again
SELECT s.*, d.d_year
FROM store_sales s
JOIN date_dim d ON s.ss_sold_date_sk = d.d_date_sk
)

SELECT s3.d_year, s2.i_category, s2.total_cat_sales
FROM stage2_small_agg s2
JOIN stage3_broadcast_filter s3 ON s2.i_category = s3.i_category
ORDER BY s3.d_year, s2.i_category;

This pattern is common for reporting and dimensional analysis scenarios and is effective for demonstrating how Spark dynamically adjusts resource usage across job stages based on cardinality and parallelism needs. Such queries can also benefit from the elasticity enabled by external serverless storage.

Rectangle pattern queries

Not all queries benefit from externalizing the shuffle. Consider a query where the cardinality is high throughout, meaning both the stages operate on a large number of partitions and keys. Typically, queries that group by high-cardinality columns (such as item or customer) cause most stages to require similar amounts of parallelism. The following figure illustrates a Spark workload where parallelism remains consistently high across stages. In this pattern, both Stage 1 and Stage 2 operate on a large number of partitions and keys, resulting in sustained executor demand throughout the job lifecycle.

High-cardinality execution pattern with sustained parallelism

Rectangle pattern queries

The following query is the same query that we used in the inverted triangle pattern earlier, with one change. We have replaced the dim_date table (low cardinality) with item (high cardinality).

SELECT i_item_id, SUM(ss_net_paid) AS total_sales
FROM store_sales
JOIN item ON store_sales.ss_item_sk = item.i_item_sk
GROUP BY i_item_id
ORDER BY i_item_id
LIMIT 100;

Stage 1: Reads the rows from store_sales and joins with item, spreading data across many partitions—similar to the original query’s first stage.
Stage 2: The aggregation is by i_item_id, which normally has thousands to millions of distinct values in real datasets. This keeps parallelism high; many tasks handle non-overlapping keys, and shuffle outputs remain large.

There is no significant drop in cardinality: Because neither stage is reduced to a small group set, most executors stay busy throughout the job’s main phases, with little idle time even after the shuffle. This type of query results in a flatter executor utilization profile because each stage processes a similar volume of work, thus minimizing variation in resource utilization. These rectangle pattern queries will not see the cost benefit from the elasticity obtained by offloading shuffle data. However, there may still be other benefits such as reduction of job failures and performance bottlenecks from disk constraints, freedom from capacity planning and sizing, and provisioning of storage for intermediate data operations.

Conclusion

Serverless storage for Amazon EMR Serverless can deliver substantial cost savings for workloads with dynamic resource patterns, as seen in the 26% average cost savings we observed in our testing environment. By externalizing shuffle data, you can gain the elasticity to release idle executors immediately, demonstrated by the savings reaching up to 85% in our testing environment, on queries following inverted triangle and hourglass patterns when Dynamic Resource Allocation is enabled.Understanding your workload characteristics is key. While rectangle pattern queries may not see dramatic cost reductions, they can still benefit from improved reliability and removal of capacity planning overhead.

To get started: Analyze your job execution patterns, enable Dynamic Resource Allocation, and pilot serverless storage on shuffle-heavy workloads. Looking to reduce your Amazon EMR Serverless costs for Spark workloads? Explore serverless storage for EMR Serverless today.

About the authors

Optimize HBase reads with bucket caching on Amazon EMR

2026-03-10 Yu-Ting Su

Post Syndicated from Yu-Ting Su original https://aws.amazon.com/blogs/big-data/optimize-hbase-reads-with-bucket-caching-on-amazon-emr/

Apache HBase is a database system for big data applications that efficiently manages billions of rows and millions of columns. Its distributed, column-oriented structure handles both structured and unstructured data while addressing speed, flexibility, and scalability challenges. Amazon EMR HBase on Amazon S3 extends these features by storing data directly in Amazon S3, enabling data persistence and cross-zone access while supporting compute-based cluster sizing and read-only replicas.

HBase BucketCache serves as an advanced L2 caching mechanism that works alongside traditional on-heap memory cache. It stores large data volumes outside the JVM heap, reducing garbage collection overhead while maintaining fast access. When combined with Amazon EBS gp3 SSDs, it provides near-HDFS performance at lower costs.

However, implementing terabyte-scale BucketCache in production environments presents challenges: determining optimal cache sizes, balancing cost versus performance, and configuring eviction policies for S3-backed storage.

In this post, we demonstrate how to improve HBase read performance by implementing bucket caching on Amazon EMR. Our tests reduced latency by 57.9% and improved throughput by 138.8%. This solution is particularly valuable for large-scale HBase deployments on Amazon S3 that need to optimize read performance while managing costs.

The following diagram shows Amazon EMR’s integration with Apache HBase and Amazon S3 to implement a multi-tiered caching strategy.

Amazon EMR HBase multi-tiered caching architecture diagram showing client applications connecting to HBase Master nodes that route requests to RegionServers across CORE nodes. Each node implements L1 on-heap cache and L2 bucket cache layers, with Amazon S3 providing persistent storage. CloudWatch monitors performance metrics across all components.

Figure 1 – Solution Architecture

The solution implements key components:

Configure persistent bucket cache with validated parameters
Implement cache-aware load balancing
Use ZGC for improved garbage collection performance
Monitor cache effectiveness through l2CacheHitRatio using Amazon EMR metrics

In our testing with datasets in terabytes, we achieved:

Bucket cache hit ratios exceeding 95%
S3 GET requests reduced to under 1,000/hour at peak performance
Read latencies reduced to milliseconds
Zero JVM pause detection during high read workloads
138.8% improvement in read throughput

Walkthrough

Prerequisites

This section shows how we improved HBase read performance using bucket caching on Amazon EMR in our tests. Before implementing this solution, you should have:

AWS resources

An AWS account with appropriate permissions
The following IAM roles:
EMR_DefaultRole_V2
EMR_EC2_DefaultRole with Amazon S3 access
Familiarity with Apache HBase and Amazon EMR

Technical requirements:

Amazon EMR version 7.10.0 or later
Apache HBase 2.6.2 on Amazon EMR 7.10.0 or later
ycsb 0.17.0 for benchmarking
AWS CLI installed and configured

For setup instructions, refer to:

Create an EMR Cluster with optimized configuration

Create your EMR cluster using the following exampled launch command. This command is an optimization demo for terabyte-scale bucket caching:

aws emr create-cluster \
 --name "EMR HBase Bucket cache" \
 --log-uri "<your-s3-log-location>" \
 --release-label "emr-7.12.0" \
 --service-role "arn:aws:iam::<your-account-id>:role/EMR_DefaultRole_V2" \
 --ec2-attributes '{
    "InstanceProfile": "EMR_EC2_DefaultRole",
    "EmrManagedMasterSecurityGroup": "<your-primary-security-group-id>",
    "EmrManagedSlaveSecurityGroup": "<your-worker-security-group-id>",
    "KeyName": "<your-key-name>",
    "AdditionalMasterSecurityGroups": [],
    "AdditionalSlaveSecurityGroups": [],
    "SubnetIds": ["<your-subnet-id>"]
}' \
 --applications Name=AmazonCloudWatchAgent Name=HBase Name=ZooKeeper \
 --configurations '[
    {
        "Classification": "hbase",
        "Properties": {
            "hbase.emr.storageMode": "s3"
        }
    },
    {
    "Classification": "hbase-env",
    "Properties": {},
    "Configurations": [{
        "Classification": "export",
        "Properties": {
            "HBASE_HEAPSIZE": "<your-jvm-heap-size>",
            "HBASE_REGIONSERVER_GC_OPTS": "\"-XX:+UseZGC -XX:+ZGenerational -XX:+AlwaysPreTouch\"",
            "HBASE_REGIONSERVER_OPTS": "\"-Xmx<YOUR-JVM-HEAP-SIZE>m\"",
            "JAVA_HOME": "/usr/lib/jvm/jre-21"
        }
    }]
},
    {
        "Classification": "hbase-site",
        "Properties": {
            "hbase.rootdir": "<your-hbase-rootdir>",
            "hbase.bucketcache.size": "<your-bucket-cache-size-per-region-server>",
            "hbase.bucketcache.bucket.sizes": "<bucket-sizes-of-your-bucket-cache>",
            "hbase.master.loadbalancer.class": "org.apache.hadoop.hbase.master.balancer.CacheAwareLoadBalancer",
            "hbase.bucketcache.persistent.path": "/mnt/hbase/persistent_cache",
            "hbase.bucketcache.writer.threads": "<your-bucket-cache-writer-threads>",
            "hbase.bucketcache.writer.queuelength": "<your-bucket-cache-writer-queue-length>",
            "hbase.rs.prefetchblocksonopen": "true",
            "hbase.rs.cacheblocksonwrite": "true",
            "hbase.rs.cachecompactedblocksonwrite": "true",
            "hbase.block.data.cachecompressed": "true"
        }
    },
 {
    "Classification": "emr-metrics",
    "Configurations": [{
        "Classification": "emr-hbase-region-server-metrics",
        "Properties": {
            "Hadoop:service=HBase,name=RegionServer,sub=Server": "writeRequestCount,readRequestCount,l2CacheHitCount,l2CacheMissCount,l2CacheHitRatio",
            "otel.metric.export.interval": "30000"
        }
    }],
    "Properties": {}
}]' \
 --instance-groups '[{
    "InstanceCount": 6,
    "InstanceGroupType": "CORE",
    "Name": "Core",
    "InstanceType": "r8g.2xlarge",
    "EbsConfiguration": {
        "EbsBlockDeviceConfigs": [{
                "VolumeSpecification": {
                    "VolumeType": "gp3",
                    "Iops": 3000,
                    "SizeInGB": <size-depends-on-bucket-cache-size>
                },
                "VolumesPerInstance": 1
        }]
    }
}, {
    "InstanceCount": 1,
    "InstanceGroupType": "MASTER",
    "Name": "Primary",
    "InstanceType": "r8g.2xlarge",
    "EbsConfiguration": {
        "EbsBlockDeviceConfigs": [{
            "VolumeSpecification": {
                "VolumeType": "gp3",
                "SizeInGB": 64
            },
            "VolumesPerInstance": 2
        }]
    }
}]' \
 --scale-down-behavior "TERMINATE_AT_TASK_COMPLETION" \
 --ebs-root-volume-size "30" \
 --region "<region-id>"

Explain configurations for HBase optimized cache performance

In the above launch command, you can see configurations through the EMR software configurations. These settings are specifically for terabyte-scale caching scenarios. When HBase is installed on EMR, Apache YARN’s memory allocation is reduced by approximately 50% from its default configuration (68-73% of RAM) to 34-36% of physical RAM, reserving memory for HBase RegionServer operations. The cache and memstore sizes must be carefully balanced against available node memory to prevent resource contention.

The hbase.bucketcache.size parameter determines the total bucket cache size per RegionServer in megabytes, which directly affects how much data can be stored in bucket cache. If the data files are stored in compressed formats, you have to enable hbase.block.data.cachecompressed . This feature keeps blocks compressed in the cache, reducing memory footprint while maintaining quick access times. Your EBS size per RegionServer depends on the value of hbase.bucketcache.size. The configured EBS size can be the value of this feature plus a buffer for system usage. The hbase.bucketcache.bucket.sizes setting defines bucket sizes to efficiently accommodate different data block sizes, while hbase.bucketcache.writer.threads controls the number of threads used for writing to the cache, optimizing write performance.

In the above launch command, we configured ZGC settings to optimize garbage collection.

Using ZGC minimizes the need for a large JVM heap to accommodate JVM objects for large-scale bucket cache operations, resulting in fewer JVM pauses. By adjusting the heap size through increasing or decreasing the HBASE_HEAPSIZE parameter, you can optimize memory allocation for your specific workload. A key advantage of ZGC is that it keeps JVM pause times short regardless of heap size, whereas traditional garbage collectors experience longer full GC times as heap size increases. This makes ZGC particularly valuable for HBase deployments with terabyte-scale bucket caches, where maintaining consistent low-latency performance is critical.

The generational garbage collection settings efficiently manage memory by separating short-lived objects from long-lived ones, reducing collection frequency and overhead. The AlwaysPreTouch parameter improves Apache HBase responsiveness by pre-allocating memory during operation.

Explain EMR metrics collection configurations

In the above launch command, we set up configurations to publish emr metrics to CloudWatch through CloudWatch Agents. We can use these metrics to track bucket cache request amount and hit ratios. If L2CacheHitRatio is high but L2CacheMissCount is low, it means HBase can fetch most of the requested data in bucket cache. The read latencies can be shorted to milliseconds in this case.

Performance testing and results

This section details our performance testing methodology and results using a 7.9 TB dataset.

Test setup

We used ycsb to generate and test with a 7.9 TB dataset.

# generate a test dataset
bin/ycsb load hbase20 \
-p columnfamily=cf \
-p recordcount=49828500 \
-p fieldcount=2000 \
-p fieldlength=425 \
-P workloads/workloadc \
-threads 150 -s

We used the following command to run a read-only workload:

for i in {1..3}
    do
      nohup bin/ycsb.sh run hbase20 -p columnfamily=cf -p recordcount=49828500 -p operationcount=49828500 -P workloads/workloadc -threads 500 -s > /dev/null &
done

Configuration	Throughput (ops/sec)	Latency (ms)
Without Cache	371.93	2680
With Cache	888.67	1127
Improvement	138.80%	57.90%

In our read performance test using bucket cache to cache terabytes of data, we achieved a 138.8% improvement in read throughput (from 371.93 to 888.67 ops/sec) and a 57.9% reduction in read latency (from 2680ms to 1127ms) compared to a scenario without bucket cache.

Read performance improvement

As shown in the previoustable, implementing bucket cache led to improvements in both throughput and latency. The system achieved a 138.8% increase in throughput, processing 888.67 operations per second compared to the baseline 371.93 ops/sec. Similarly, latency was reduced by 57.9%, dropping from 2680ms to 1127ms, demonstrating the performance benefits of the caching solution. The following chart shows implementing bucket cache led to improvements in average throughput compared to a scenario without bucket cache.

Bar chart comparing HBase read throughput performance: BucketCache enabled achieves 888.66 operations per second versus 372.14 ops/sec when disabled, demonstrating 2.4x performance improvement. This validates multi-tiered caching effectiveness in Amazon EMR clusters for optimizing read-intensive workloads and reducing S3 access costs.

Figure 2 – Average throughput comparison

Cache hit ratio progression

The cache hit ratio data demonstrates the effectiveness of the bucket cache implementation over time. Starting from 0% at initialization, the cache hit ratio improved to 85% within 12 hours, ultimately stabilizing above 95% after 24 hours. This progression corresponded with an extensive reduction in Amazon S3 GetObject requests, from 95,000 per hour initially to fewer than 1,000 per hour at peak performance, reducing both latency and costs.

Time (hours)	Hit Ratio	S3 Requests/hour
0	0%	95,000
12	85%	15,000
24	95%+	<1,000

Amazon CloudWatch time-series graph displaying L2 cache hit ratio metrics for six Amazon EMR HBase RegionServer instances from September 25-27, 2025. All instances show improving performance from 0.60-0.75 range to 0.95, demonstrating optimal BucketCache effectiveness in reducing Amazon S3 operations and improving query latency

Figure 3 – Bucket cache hit ratio increased after we loaded data to bucket cache through read-only workload.

Amazon S3 GetObject request count time-series graph showing dramatic traffic reduction following BucketCache implementation. Requests peak at 900,000 on September 25, 2025, then decline sharply to under 50,000 within 24 hours, stabilizing near baseline by September 26, demonstrating L2 cache effectiveness in minimizing S3 operations and reducing costs.

Figure 4 – Amazon S3 GetObject request count decreased as bucket cache hit ratio increased.

Key implementation: persistent bucket cache

One of the key features introduced in HBase 2.6.0 after Amazon EMR 7.6.0 is persistent bucket cache, which maintains cache data across RegionServer restarts. This feature is particularly for production environments where maintaining consistent performance during maintenance operations is crucial. The following section demonstrate how to configure persistent bucket cache.

Configuring persistent bucket cache

Set up persistent bucket cache by implementing these configurations:

{
    "Classification": "hbase-site",
    "Properties": {
        "hbase.bucketcache.persistent.path": "/mnt/hbase/persistent_cache",
        "hbase.master.loadbalancer.class": "org.apache.hadoop.hbase.master.balancer.CacheAwareLoadBalancer",
        "hbase.master.scp.retain.assignment": "true",
        "hbase.master.scp.retain.assignment.force": "true",
        "hbase.master.scp.retain.assignment.force.retries": "10"
    }
}

Performance impact of persistent cache

The following table shows the tests demonstrated significant improvements in RegionServer restart performance. With persistent cache enabled, the HBase cluster maintained consistent read request performance and low latency after RegionServer restarts since data remained directly accessible in the bucket cache. In contrast, clusters without persistent cache required 6 hours to reload bucket cache after RegionServer restarts before achieving comparable read operation performance and latency levels. It demonstrates significant improvements from enabling persistent cache.

	Pre-restart throughput	Post-restart throughput	Recovery time
Without Persistent Cache	888.67 ops/sec	371.93 ops/sec	~6 hours
With Persistent Cache	889.08 ops/sec	886.71 ops/sec	<2 minutes

In the following graph, the RegionServer L2 cache size metrics revealed that the bucket cache size remained stable after RegionServer restart, confirming that the cached data was preserved rather than reset during the process. The metrics were unavailable between 16:30 and 16:35 because the RegionServer was stopped and restarted.

Amazon CloudWatch line graph monitoring RegionServer L2 bucket cache size growth across three Amazon EMR core nodes from 15:55 to 16:50. All instances show steady cache population from zero to approximately 18-20 GB, indicating successful BucketCache warm-up as frequently accessed HBase data loads into secondary cache for improved read performance.

Figure 5 – The bucket cache size remained stable after RegionServer restart

L2 cache miss count is a cumulative metric that tracks cache misses from RegionServer startup. When the RegionServer restarts, this metric resets to zero. In the following graph, the L2 cache miss count increased steeply at the beginning because read requests retrieved data from HFiles, as the data had not yet been loaded into bucket cache. Over time, the bucket cache was populated with data through read-only workload, and the slope of the L2 cache miss count decreased. We restarted RegionServer between 16:30 and 16:35 . Thus, L2 cache miss count reset to 0. Notably, these metrics remained at zero even during subsequent client read operations. The requests did not retrieve data from HFiles that caused an increase in L2 cache miss count. This confirmed that data persisted in the bucket cache and was directly accessible without requiring cache rebuilding.

Amazon CloudWatch line graph tracking RegionServer L2 bucket cache miss counts across three Amazon EMR HBase core nodes from 16:05 to 16:45. Graph shows initial cache misses during empty BucketCache warm-up (315k-630k), stabilized plateau after cache population, and sharp drop following RegionServer restart at 16:32, demonstrating cache behavior during operational cycles.

Figure 6 – Regionserver bucketcache miss count remained 0 after restarting RegionServer

The RegionServer read request count metrics demonstrated consistent read operation volumes following restart. This indicated that RegionServers maintained read performance levels without needing to fetch HFiles from Amazon S3, thus avoiding the increased latency and reduced throughput typically associated with S3 lookups. This persistent cache behavior directly reduces S3 costs by minimizing API calls—our above testing statistics showed S3 GET requests dropping from 95,000 per hour during initial cache warming to fewer than 1,000 per hour once the cache reached optimal performance, representing a 99% reduction in S3 API call volume.

Amazon CloudWatch line graph displaying RegionServer read request counts across three Amazon EMR HBase core nodes from 16:05 to 16:50. Graph shows consistent read traffic at 2.3M requests during initial and secondary reads, peaking at 4.6M around 16:27, followed by service interruption and recovery post-restart, illustrating typical HBase query patterns with BucketCache enabled.

Figure 7 – Regionserver read request count

Best practices and recommendations

In this section, we share guidelines to optimize HBase bucket cache performance.

cache sizing guidelines

Enable hbase.block.data.cachecompressed when working with compressed Hfiles: This setting ensures data blocks are stored in the bucket cache in compressed form, saving memory and improving efficiency.
Activate hbase.rs.cachecompactedblocksonwrite: This prevents data eviction during HBase compactions.

For optimal performance, size your bucket cache appropriately by ensuring the total cache size exceeds your target cached data volume. Insufficient bucket cache size will lead to frequent data evictions, degrading system performance. Monitor free cache space using Amazon CloudWatch metrics to prevent overflow issues. Furthermore, consistently analyze L2 cache hit ratio metrics to assess performance, and adjust bucket cache size based on your specific workload patterns and L2 hit ratio trends. These ongoing monitoring and adjustment practices will help maintain optimal cache performance and resource utilization.

Performance optimization

To further enhance HBase read performance, consider implementing the following configuration settings. These optimizations are designed to improve cache utilization, reduce disk I/O, and minimize latency for common read operations:

[{
    "Classification": "hbase-site",
    "Properties": {
        "hbase.bucketcache.persistent.path": "/mnt/hbase/persistent_cache",
        "hbase.master.loadbalancer.class": "org.apache.hadoop.hbase.master.balancer.CacheAwareLoadBalancer",
        "hbase.master.scp.retain.assignment": "true",
        "hbase.master.scp.retain.assignment.force": "true",
        "hbase.master.scp.retain.assignment.force.retries": "10",
        "hbase.rs.prefetchblocksonopen": "true",
        "hbase.rs.cacheblocksonwrite": "true",
        "hbase.rs.cachecompactedblocksonwrite": "true",
        "hbase.block.data.cachecompressed": "true"
    }
}, {
    "Classification": "emr-metrics",
    "Configurations": [{
        "Classification": "emr-hbase-region-server-metrics",
        "Properties": {
            "Hadoop:service=HBase,name=RegionServer,sub=Server": "writeRequestCount,readRequestCount,l2CacheHitCount,l2CacheMissCount,l2CacheHitRatio",
            "otel.metric.export.interval": "30000"
        }
    }]
}]

Resource monitoring

Set up Amazon CloudWatch dashboards to monitor key metrics. These dashboards should track L2 cache hit ratios, which provide insight into the effectiveness of your caching strategy. Additionally, monitor Amazon S3 request patterns to understand your data access trends and optimize accordingly. Keep a close eye on memory utilization to ensure your instances have sufficient resources to handle the workload efficiently. Finally, regularly analyze garbage collection (GC) patterns to identify and address any potential memory management issues that could impact performance.

Cleaning up

To avoid incurring unnecessary charges, clean up your resources when you’re done testing

# Terminate EMR cluster
aws emr terminate-clusters \
--cluster-id <your-cluster-id>

# Remove test data from S3
aws s3 rm s3://<your-bucket>/hbase-root/ --recursive

Conclusion

In this post, you learned how to implement and optimize HBase bucket cache with persistent storage on Amazon EMR. In our testing, we achieved 95%+ cache hit ratios with consistent millisecond latencies. The implementation reduced Amazon S3 access costs by minimizing the number of direct Amazon S3 requests required. Read performance saw 138.8% improvement in read throughput. The system maintained stable performance during maintenance windows, eliminating performance degradation during routine operations. Additionally, the solution demonstrated better resource utilization, maximizing the efficiency of the allocated infrastructure while minimizing waste.

Related resources

About the author

How Razorpay achieved 11% performance improvement and 21% cost reduction with Amazon EMR

2026-03-09 Narendra Kumar

Post Syndicated from Narendra Kumar original https://aws.amazon.com/blogs/big-data/how-razorpay-achieved-11-performance-improvement-and-21-cost-reduction-with-amazon-emr/

This is a guest post by Narendra Kumar, Head of Platform – Data at Razorpay, in partnership with AWS.

In this post, we explore how Razorpay, India’s leading FinTech company, transformed their data platform by migrating from a third-party solution to Amazon EMR, unlocking improved performance and significant cost savings. We’ll walk through the architectural decisions that guided this migration, the implementation strategy, and the measurable benefits Razorpay achieved.

Founded in 2014, Razorpay has become a powerhouse in comprehensive payment solutions, enabling businesses to accept, process, and disburse payments online. With offerings like RazorpayX for business banking and Razorpay Capital for lending solutions, the company has experienced explosive growth, now serving millions of businesses. This rapid expansion brought significant data challenges. When Razorpay’s data platform began straining under the weight of more than 1PB daily processing demands, the engineering team faced a critical decision: continue scaling their existing third-party solution or modernize with a platform offering greater flexibility and control. They chose Amazon EMR to build a comprehensive data architecture spanning batch warehousing, real-time stream processing, and interactive analytics – all running on Apache Spark with open-source Delta Lake for ACID transactions. This wasn’t simply an ETL migration; it was a complete platform transformation that gave Razorpay’s 800 daily users access to more than 60 concurrent streaming pipelines, more than 3,000 orchestrated workflows, and the ability to query 6PB of data daily. The results validated their architectural choices: 11% better overall performance, 21% cost reduction, and the operational flexibility to optimize Spark resource allocation, leverage EC2 Spot instances, and implement advanced features like liquid clustering – all without vendor lock-in.

Achieving data insights cost-effectively with AWS

The data architecture has a data ingestion layer, data processing layer, and data consumption layer. Razorpay ingests more than 20 TB of new data every day, processes more than 1 PB of daily data using more than 60 data stream processing pipelines. This data is then consumed by querying more than 6 PB of daily data through more than 3,000 scheduled workflows.

Data flows from a variety of sources such as online transaction processing (OLTP) databases – traditional transactional or entity stores, events such as clickstream and application events, and third-party events like reverse extract, transform, and load (ETL). Most of the data consumption use cases power merchant reporting and internal analytics of the organization. The architecture powers a variety of data science use cases and financial infrastructure around a reconciliation service.

Solution overview

As shown in the following diagram, in its early stages, Razorpay operated on a small scale, using Sqoop to dump transactional data daily into a data lake and managing a Presto layer for querying this data. As they grew, the demand for near real-time data increased, prompting the setup of a change data capture (CDC) collector using Maxwell to stream data manipulation language (DML) events to Kafka. To further enhance data processing, Razorpay built a processing layer that consumed data from Kafka to UPSERT information into the lake using Apache Hudi.

Architecture diagram showing a five-layer big data processing pipeline: data stores feed into Kafka for message streaming, which connects to Apache Spark, Apache Hudi, and Sqoop for stream and batch processing, followed by a data storage and query layer using Apache Hive and Apache Presto, and finally a visualization layer with Looker, redash, and Qubole.

Additionally, the company onboarded data from third-party sources such as Freshdesk and Google Sheets and automated event ingestion from frontend applications using Lumberjack, thereby streamlining their data management processes.

As Razorpay scaled its operations, the demand for multiple real-time use cases became mission-critical, prompting the development of a robust data warehouse ingestion framework to efficiently ingest data into TiDB. To enhance service reliability and support dashboard querying, a low-latency, high-throughput service called Harvester was created, which stored pre-aggregated data for effective monitoring. Over time, reporting use cases emerged, leading to the use of a warehouse service to establish a denormalized report data layer while also exploring a real-time layer for dynamic insights. Additionally, to facilitate a smooth transition to microservices, Razorpay built a unified storage layer capable of supporting data from both its existing monolithic architecture and the new microservices, ensuring seamless integration and improved data accessibility across the organization.

Razorpay implemented a comprehensive data service migration to Amazon EMR using a phased approach. The solution architecture as shown in the following diagram comprises multiple layers handling data ingestion, processing, and consumption.

Technical implementation

A modern and scalable analytics platform focuses on real-time data ingestion, petabyte-scale processing, and cost-optimized storage – all orchestrated with robust workflow management:

Data ingestion layer

To handle large-scale and diverse data sources, they implemented a combination of CDC and file ingestion patterns:

CDC using Amazon Aurora MySQL-Compatible Edition – Used Debezium and Maxwell for low-latency replication and streaming of database changes
High-volume streaming pipelines – Configured streaming pipelines capable of processing more than 20 TB of daily inbound data
Third-party data integration: Implemented secure file push mechanisms to ingest partner and software as a service (SaaS) data into the service

Data processing layer

Razorpay designed the processing stack on Amazon EMR on Amazon Elastic Compute Cloud (Amazon EC2) with Spark as the primary compute engine

Batch warehousing – Daily ETL and aggregation jobs processing more than 1 PB of data
Stream processing – Real-time analytics pipelines across more than 60 concurrent processing streams
Delta merge operations – High-performance incremental updates across more than 25 Delta Lake tables

Data storage and organization

Their data storage follows the medallion architecture pattern layered on an Amazon Simple Storage Service (Amazon S3):

Raw zone – Immutable ingestion zone for original source data
Processed and aggregated zone – Optimized datasets ready for analytics and reporting
Open source software (OSS) Delta Lake format – Implemented open source Delta Lake for ACID transactions, schema enforcement, and faster query performance

Workflow orchestration

Complex data workflows are automated and monitored using a hybrid orchestration approach:

Apache Airflow integration – Scheduling and coordinating more than 3,000 workflows per day
dbt on Amazon EMR – SQL-based transformations for business logic and metric definitions
Specialized compliance jobs – Dedicated workflows meeting the 15-minute SLA for sensitive regulatory reporting

Performance optimizations

To ensure cost efficiency and high throughput, the following optimizations were applied:

Spark tuning – Custom configurations for executor memory, shuffle partitions, and serialization to maximize hardware utilization
Liquid clustering – Implemented in delta lake tables to improve query performance over large datasets
Optimized delta merges – Reduced merge latency for incremental updates.
Auto scaling – Dynamic scaling policies based on workload patterns to balance performance and cost

To enable a secure migration, they implemented Amazon EMR security best practices following AWS guidance on encryption, authentication, and authorization as documented in the Amazon EMR security best practices.

This architecture delivers low-latency ingestion, petabyte-scale processing, and robust workflow orchestration so that analytics teams can derive faster insights while maintaining compliance and optimizing for cost.

The combination of Debezium and Maxwell for CDC, Spark on Amazon EMR, OSS Delta Lake on Amazon S3, and Airflow with dbt has proven to be a scalable and resilient approach for modern data analytics workloads

Business Impact: What Amazon EMR Enabled

11% performance improvement enabling faster insights for 800 daily active users
13-15% faster execution for large warehouse jobs, accelerating time-to-insight for critical business decisions
21% cost reduction reinvested into product innovation for merchant customers
Seamless scaling from 20 TB to 1 PB+ daily processing without performance degradation
Enterprise reliability supporting 350,000 operational reports and compliance requirements

Key learnings and best practices

Throughout their migration to Amazon EMR, Razorpay learned valuable lessons that helped optimize their data platform. We are sharing these insights to help other customers accelerate their own modernization journeys while avoiding common pitfalls.

Infrastructure Stability and Performance

Optimizing Spark Resource Allocation – Razorpay initially assumed that Spark’s dynamic allocation would automatically optimize resource utilization. However, they discovered it introduced overhead that degraded performance for certain workload patterns. To address this challenge, they took two approaches depending on workload characteristics – setting explicit maxExecutors values for predictable workloads, and enabling maximizeResourceAllocation to create “fat executors” that fully utilized available cluster resources. These targeted configurations improved job execution times by 13-15% for large-scale data processing workloads.
Ensuring Stability with Yet Another Resource Negotiator (YARN) node labels – When using EC2 Spot instances for cost optimization, Razorpay encountered a critical issue in which Spot instance interruptions occasionally terminated nodes running critical driver containers, causing entire job failures. Their solution was elegant and effective. They configured YARN node labels to ensure driver containers always spawn on On-Demand Instances, while task nodes use cost-effective Spot capacity. This architecture delivered both cost efficiency and reliability, making their jobs resilient to Spot interruptions while maintaining 21% cost savings.
Managing Spot Instances Effectively – Razorpay’s initial approach of switching entirely to On-Demand Instances during Spot availability constraints eliminated the cost benefits they were seeking. They implemented several best practices to address this such as using instance fleets with allocation strategies (price-capacity optimized and capacity optimized) to maximize Spot availability, spreading primary instances across multiple Availability Zones for fault tolerance, and accepting that heterogeneous executors create varying executor sizes while planning capacity accordingly. They maintained high Spot utilization rates while ensuring workload continuity, achieving optimal price performance.

Cost Optimization

Achieving Sustainable Cost Efficiency – As data volumes grew to more than 20 TB daily, Razorpay needed to scale infrastructure while controlling costs. They implemented a comprehensive cost optimization strategy that included multiple components. First, they right-sized primary nodes by avoiding over-provisioning and selecting instance types matching actual workload requirements. They consolidated workloads by combining multiple jobs on fewer large clusters to maximize resource utilization. For SLA-sensitive jobs, they migrated to Amazon EKS and Amazon EMR Serverless for automatic scaling and pay-per-use pricing. They adopted Graviton instances, migrating compatible workloads to AWS Graviton processors for superior price-performance. Finally, they diversified instance fleets by employing multiple instance types to reduce Spot interruption impact.

These optimizations delivered 21% cost savings while supporting 800 daily active users and processing 1 PB of data daily. This enabled Razorpay to invest savings back into product innovation for their merchant customers, demonstrating how technical optimization directly translates to business value.

Conclusion

Razorpay’s migration to Amazon EMR demonstrates how the right data processing platform can transform business outcomes at scale. By achieving 11% better performance, 13-15% faster execution times, and 21% cost savings, EMR enabled Razorpay to build an enterprise-grade data platform that supports 800 daily users, more than 3,000 dashboards, and 10 million monthly queries.

To learn more about building similar data analytics solutions on AWS, check out the following resources.

Documentation:

AWS solutions:

Get started:

Visit the Amazon EMR product page
Explore Amazon EMR Notebooks overview

About the authors

Orchestrate end-to-end scalable ETL pipeline with Amazon SageMaker workflows

2026-02-10 Shubham Kumar

Post Syndicated from Shubham Kumar original https://aws.amazon.com/blogs/big-data/orchestrate-end-to-end-scalable-etl-pipeline-with-amazon-sagemaker-workflows/

Amazon SageMaker Unified Studio serves as a collaborative workspace where data engineers and scientists can work together on end-to-end data and machine learning (ML) workflows. SageMaker Unified Studio specializes in orchestrating complex data workflows across multiple AWS services through its integration with Amazon Managed Workflows for Apache Airflow (Amazon MWAA). Project owners can create shared environments where team members jointly develop and deploy workflows, while maintaining oversight of pipeline execution. This unified approach makes sure data pipelines run consistently and efficiently, with clear visibility into the entire process, making it seamless for teams to collaborate on sophisticated data and ML projects.

This post explores how to build and manage a comprehensive extract, transform, and load (ETL) pipeline using SageMaker Unified Studio workflows through a code-based approach. We demonstrate how to use a single, integrated interface to handle all aspects of data processing, from preparation to orchestration, by using AWS services including Amazon EMR, AWS Glue, Amazon Redshift, and Amazon MWAA. This solution streamlines the data pipeline through a single UI.

Example use case: Customer behavior analysis for an ecommerce platform

Let’s consider a real-world scenario: An e-commerce company wants to analyze customer transactions data to create a customer summary report. They have data coming from multiple sources:

Customer profile data stored in CSV files
Transaction history in JSON format
Website clickstream data in semi-structured log files

The company wants to do the following:

Extract data from these sources
Clean and transform the data
Perform quality checks
Load the processed data into a data warehouse
Schedule this pipeline to run daily

Solution overview

The following diagram illustrates the architecture that you implement in this post.

This architecture diagram illustrates a comprehensive, end-to-end data processing pipeline built on AWS services, orchestrated through Amazon SageMaker Unified Studio. The pipeline demonstrates best practices for data ingestion, transformation, quality validation, advanced processing, and analytics.

The workflow consists of the following steps:

Establish a data repository by creating an Amazon Simple Storage Service (Amazon S3) bucket with an organized folder structure for customer data, transaction history, and clickstream logs, and configure access policies for seamless integration with SageMaker Unified Studio.
Extract data from the S3 bucket using AWS Glue jobs.
Use AWS Glue and Amazon EMR Serverless to clean and transform the data.
Implement data quality validation using AWS Glue Data Quality.
Load the processed data into Amazon Redshift Serverless.
Create and manage the workflow environment using SageMaker Unified Studio with Identity Center–based domains.

Note: Amazon SageMaker Unified Studio supports two domain configuration models: IAM Identity Center (IdC)–based domains and IAM role–based domains. While IAM-based domains enable role-driven access management and visual workflows, this post specifically focuses on Identity Center–based domains, where users authenticate via IdC and projects access data and resources using project roles and identity-based authorization.

Prerequisites

Before beginning, ensure you have the following resources:

An AWS account.
AWS IAM Identity Center enabled in your AWS account with users and groups created. For detailed instructions, see Getting started with IAM Identity Center and Access Amazon SageMaker Unified Studio.

Configure Amazon SageMaker Unified Studio domain

This solution requires SageMaker Unified Studio domain in the us-east-1 AWS Region. Although SageMaker Unified Studio is available in multiple Regions, this post uses us-east-1 for consistency. For a complete list of supported Regions, refer to Regions where Amazon SageMaker Unified Studio is supported.

Complete the following steps to configure your domain:

Sign in to the AWS Management Console, navigate to Amazon SageMaker, and open the Domains section from the left navigation pane.
On the SageMaker console, choose Create domain, then choose Quick setup.
If the message “No VPC has been specifically set up for use with Amazon SageMaker Unified Studio” appears, select Create VPC. The process redirects to an AWS CloudFormation stack. Leave all settings at their default values and select Create stack.
Under Quick setup settings, for Name, enter a domain name (for example, etl-ecommerce-blog-demo). Review the selected configurations.
Choose Continue to proceed.
On the Create IAM Identity Center user page, create an SSO user (account with IAM Identity Center) or select an existing SSO user to log in to the Amazon SageMaker Unified Studio. The SSO selected here is used as the administrator in the Amazon SageMaker Unified Studio.
Choose Create domain.

For detailed instructions, see Create a SageMaker domain and Onboarding data in Amazon SageMaker Unified Studio.

After you have created a domain, popup will appear with the message: “Your domain has been created! You can now log in to Amazon SageMaker Unified Studio”. You can close the popup for now.

Create a project

In this section, we create a project to serve as a collaborative workspace for teams to work on business use cases. Complete the following steps:

Choose Open Unified Studio and sign in with your SSO credentials using the Sign in with SSO option.
Choose Create project.
Name the project (for example, ETL-Pipeline-Demo) and create it using the All capabilities project profile.
Choose Continue.
Keep the default values for the configuration parameters and choose Continue.
Choose Create project.

Project creation might take a few minutes. After the project is created, the environment will be configured for data access and processing.

Integrate S3 bucket with SageMaker Unified Studio

To enable external data processing within SageMaker Unified Studio, configure integration with an S3 bucket. This section walks through the steps to set up the S3 bucket, configure permissions, and integrate it with the project.

Create and configure S3 bucket

Complete the following steps to create your bucket:

In a new browser tab, open the AWS Management Console and search for S3.
On the Amazon S3 console, choose Create Bucket .
Create a bucket named ecommerce-raw-layer-bucket-demo-<Account-ID>-us-east-1. For detailed instructions, see create a general-purpose Amazon S3 bucket for storage.
Create the following folder structure in the bucket. For detailed instructions, see Creating a folder:
- raw/customers/
- raw/transactions/
- raw/clickstream/
- processed/
- analytics/

Upload sample data

In this section, we upload sample ecommerce data that represents a typical business scenario where customer behavior, transaction history, and website interactions need to be analyzed together.

The raw/customers/customers.csv file contains customer profile information, including registration details. This structured data will be processed first to establish the customer dimension for our analytics.

customer_id,name,email,registration_date
1,John Doe,[email protected],2022-01-15
2,Jane Smith,[email protected],2022-02-20
3,Robert Johnson,[email protected],2022-01-30
4,Emily Brown,[email protected],2022-03-05
5,Michael Wilson,[email protected],2022-02-10

The raw/transactions/transactions.json file contains purchase transactions with nested product arrays. This semi-structured data will be flattened and joined with customer data to analyze purchasing patterns and customer lifetime value.

[
{"transaction_id": "t1001", "customer_id": 1, "amount": 125.99, "date": "2023-01-10", "items": ["product1", "product2"]},
{"transaction_id": "t1002", "customer_id": 2, "amount": 89.50, "date": "2023-01-12", "items": ["product3"]},
{"transaction_id": "t1003", "customer_id": 1, "amount": 45.25, "date": "2023-01-15", "items": ["product2"]},
{"transaction_id": "t1004", "customer_id": 3, "amount": 210.75, "date": "2023-01-18", "items": ["product1", "product4", "product5"]},
{"transaction_id": "t1005", "customer_id": 4, "amount": 55.00, "date": "2023-01-20", "items": ["product3", "product6"]}
]

The raw/clickstream/clickstream.csv file captures user website interactions and behavior patterns. This time-series data will be processed to understand customer journey and conversion funnel analytics.

timestamp,customer_id,page,action
2023-01-10T10:15:23,1,homepage,view
2023-01-10T10:16:45,1,product_page,view
2023-01-10T10:18:12,1,product_page,add_to_cart
2023-01-10T10:20:30,1,checkout,view
2023-01-10T10:22:15,1,checkout,purchase
2023-01-12T14:30:10,2,homepage,view
2023-01-12T14:32:20,2,product_page,view
2023-01-12T14:35:45,2,product_page,add_to_cart
2023-01-12T14:40:12,2,checkout,view
2023-01-12T14:42:30,2,checkout,purchase

raw

For detailed instructions on uploading files to Amazon S3, refer to the Uploading objects.

Configure CORS policy

To allow access from the SageMaker Unified Studio domain portal, update the Cross-Origin Resource Sharing (CORS) configuration of the bucket:

On the bucket’s Permissions tab, choose Edit under Cross-origin resource sharing (CORS).

Enter the following CORS policy and replace domainUrl with the SageMaker Unified Studio domain URL (for example, https://<domain-id>.sagemaker.us-east-1.on.aws ). The URL can be found at the top of the domain details page on the SageMaker Unified Studio console.

[
    {
        "AllowedHeaders": [
            "*"
        ],
        "AllowedMethods": [
            "PUT",
            "GET",
            "POST",
            "DELETE",
            "HEAD"
        ],
        "AllowedOrigins": [
            "domainUrl"
        ],
        "ExposeHeaders": [
            "x-amz-version-id"
        ]
    }
]

For detailed information, see Adding Amazon S3 data and gain access using the project role.

Grant Amazon S3 access to SageMaker project role

To enable SageMaker Unified Studio to access the external Amazon S3 location, the corresponding AWS Identity and Access Management (IAM) project role must be updated with the required permissions. Complete the following steps:

On the IAM console, choose Roles in the navigation pane.
Search for the project role using the last segment of the project role Amazon Resource Name (ARN). This information is located on the Project overview page in SageMaker Unified Studio (for example, datazone_usr_role_1a2b3c45de6789_abcd1efghij2kl).
Choose the project role to open the role details page.
On the Permissions tab, choose Add permissions, then choose Create inline policy.
Use the JSON editor to create a policy that grants the project role access to the Amazon S3 location
In the JSON policy below, replace the placeholder values with your actual environment details:
- Replace <BUCKET_PREFIX> with the prefix of S3 bucket name (for example, ecommerce-raw-layer)
- Replace <AWS_REGION> with the AWS Region where your AWS Glue Data Quality rulesets are created (for example, us-east-1)
- Replace <AWS_ACCOUNT_ID> with your AWS account ID

Paste the updated JSON policy into the JSON editor.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ETLBucketListAccess",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetBucketLocation"
            ],
            "Resource": "arn:aws:s3:::<BUCKET_PREFIX>-*"
        },
        {
            "Sid": "ETLObjectAccess",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
                "s3:DeleteObject"
            ],
            "Resource": "arn:aws:s3:::<BUCKET_PREFIX>-*/*"
        },
        {
            "Sid": "GlueDataQualityPublish",
            "Effect": "Allow",
            "Action": [
                "glue:PublishDataQuality"
            ],
 "Resource":"arn:aws:glue:<AWS_REGION>:<AWS_ACCOUNT_ID>:dataQualityRuleset/*"
        }
    ]
}

Choose Next.
Enter a name for the policy (for example, etl-rawlayer-access), then choose Create policy.
Choose Add permissions again, then choose Create inline policy.

In the JSON editor, create a second policy to manage S3 Access Grants:Replace <BUCKET_PREFIX> with the prefix of S3 bucket name (for example, ecommerce-raw-layer) and paste this JSON policy.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3AGLocationManagement",
            "Effect": "Allow",
            "Action": [
                "s3:CreateAccessGrantsLocation",
                "s3:DeleteAccessGrantsLocation",
                "s3:GetAccessGrantsLocation"
            ],
            "Resource": [
                "arn:aws:s3:*:*:access-grants/default/*"
            ],
            "Condition": {
                "StringLike": {
                    "s3:accessGrantsLocationScope": "s3://<BUCKET_PREFIX>-*/*"
                }
            }
        },
        {
            "Sid": "S3AGPermissionManagement",
            "Effect": "Allow",
            "Action": [
                "s3:CreateAccessGrant",
                "s3:DeleteAccessGrant"
            ],
            "Resource": [
                "arn:aws:s3:*:*:access-grants/default/location/*",
                "arn:aws:s3:*:*:access-grants/default/grant/*"
            ],
            "Condition": {
                "StringLike": {
                    "s3:accessGrantScope": "s3://<BUCKET_PREFIX>-*/*"
                }
            }
        }
    ]
}

Choose Next.
Enter a name for the policy (for example, s3-access-grants-policy), then choose Create policy.

create policy

For detailed information about S3 Access Grants, see Adding Amazon S3 data.

Add S3 bucket to project

After you add policies to the project role for access to the Amazon S3 resources, complete the following steps to integrate the S3 bucket with the SageMaker Unified Studio project:

In SageMaker Unified Studio, open the project you created under Your projects.
Choose Data in the navigation pane.
Select Add and then Add S3 location.
Configure the S3 location:
1. For Name, enter a descriptive name (for example, E-commerce_Raw_Data).
2. For S3 URI, enter your bucket URI (for example, s3://ecommerce-raw-layer-bucket-demo-<Account-ID>-us-east-1/).
3. For AWS Region, enter your Region (for this example, us-east-1).
4. Leave Access role ARN blank.
5. Click Add S3 Location
Wait for the integration to complete.
Verify the S3 location appears in your project’s data catalog (on the Project overview page, on the Data tab, locate the Buckets pane to view the buckets and folders).

add

This process connects your S3 bucket to SageMaker Unified Studio, making your data ready for analysis.

Create notebook for job scripts

Before you can create the data processing jobs, you must set up a notebook to develop the scripts that will generate and process your data. Complete the following steps:

In SageMaker Unified Studio, on the top menu, under Build, choose JupyterLab.
Choose Configure Space and choose the instance type ml.t3.xlarge. This makes sure your JupyterLab instance has at least 4 vCPUs and 4 GiB of memory.
Choose Configure and Start Space or Save and Restart to launch your environment.
Wait a few moments for the instance to be ready.
Choose File, New, and Notebook to create a new notebook.
Set Kernel as Python 3, Connection type as PySpark, and Compute as Project.spark.compatibility.
In the notebook, enter the following script to use later for your AWS Glue job. This script processes raw data from three sources in the S3 data lake, standardizes dates, and converts data types before saving the cleaned data in Parquet format for optimal storage and querying.

Replace <Bucket-Name> with the name of actual S3 bucket in script:

import sys
from awsglue.transforms import *
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions
from pyspark.sql import functions as F
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
# Customers
customer_df = (
    spark.read
    .option("header", "true")
    .csv("s3://<Bucket-Name>/raw/customers/")
    .withColumn("registration_date", F.to_date("registration_date"))
    .withColumn("processed_at", F.current_timestamp())
)
customer_df.write.mode("overwrite").parquet(
    "s3://<Bucket-Name>/processed/customers/"
)
# Transactions
transaction_df = (
    spark.read
    .json("s3://<Bucket-Name>/raw/transactions/")
    .withColumn("date", F.to_date("date"))
    .withColumn("customer_id", F.col("customer_id").cast("int"))
    .withColumn("processed_at", F.current_timestamp())
)
transaction_df.write.mode("overwrite").parquet(
    "s3://<Bucket-Name>/processed/transactions/"
)
# Clickstream
clickstream_df = (
    spark.read
    .option("header", "true")
    .csv("s3://<Bucket-Name>/raw/clickstream/")
    .withColumn("customer_id", F.col("customer_id").cast("int"))
    .withColumn("timestamp", F.to_timestamp("timestamp"))
    .withColumn("processed_at", F.current_timestamp())
)
clickstream_df.write.mode("overwrite").parquet(
    "s3://<Bucket-Name>/processed/clickstream/"
)
print("Data processing completed successfully")
job.commit()

This script processes customer, transaction, and clickstream data from the raw layer in Amazon S3 and saves it as Parquet files in the processed layer.

Choose File, Save Notebook As, and save the file as shared/etl_initial_processing_job.ipynb.

Create notebook for AWS Glue Data Quality

After you create the initial data processing script, the next step is to set up a notebook to perform data quality checks using AWS Glue. These checks help validate the integrity and completeness of your data before further processing. Complete the following steps:

Choose File, New, and Notebook to create a new notebook.
Set Kernel as Python 3, Connection type as PySpark, and Compute as Project.spark.compatibility.

In this new notebook, add the data quality check script using the AWS Glue EvaluateDataQuality method. Replace <Bucket-Name> with the name of actual S3 bucket in script:

from datetime import datetime
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsgluedq.transforms import EvaluateDataQuality
from awsglue.transforms import SelectFromCollection

# ---------------- Glue setup ----------------
sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
job = Job(glueContext)
job.init("GlueDQJob", {})

# ---------------- Constants ----------------
RUN_DATE = datetime.utcnow().strftime("%Y-%m-%d")
year, month, day = RUN_DATE.split("-")
OUTPUT_PATH = "s3://<Bucket-Name>/data-quality-results"

# ---------------- Tables and Rules ----------------
tables = {
    "customers": ["s3://<Bucket-Name>/processed/customers/",
                  ["IsComplete \"customer_id\"", "IsUnique \"customer_id\"", "IsComplete \"email\""]],
    "transactions": ["s3://<Bucket-Name>/processed/transactions/",
                     ["IsComplete \"transaction_id\"", "IsUnique \"transaction_id\""]],
    "clickstream": ["s3://<Bucket-Name>/processed/clickstream/",
                    ["IsComplete \"customer_id\"", "IsComplete \"action\""]]
}

# ---------------- Process Each Table ----------------
for table, (path, rules) in tables.items():
    df = glueContext.create_dynamic_frame.from_options("s3", {"paths":[path]}, "parquet")
    results = EvaluateDataQuality().process_rows(
        frame=df,
        ruleset=f"Rules = [{', '.join(rules)}]",
        publishing_options={"dataQualityEvaluationContext": table}
    )
    rows = SelectFromCollection.apply(results, key="rowLevelOutcomes", transformation_ctx="rows").toDF()
    rows = rows.drop("DataQualityRulesPass", "DataQualityRulesFail", "DataQualityRulesSkip")

    # Write passed/failed rows
    for status, colval in [("pass","Passed"), ("fail","Failed")]:
        tmp = rows.filter(rows.DataQualityEvaluationResult.contains(colval))
        if tmp.count() > 0:
            tmp.write.mode("append").parquet(
        f"{OUTPUT_PATH}/{table}/status=dq_{status}/Year={year}/Month={month}/Date={day}"
            )
print("Data Quality checks completed and written to S3")
job.commit()

Choose File, Save Notebook As, and save the file as shared/etl_data_quality_job.ipynb.

Create and test AWS Glue jobs

Jobs in SageMaker Unified Studio enable scalable, flexible ETL pipelines using AWS Glue. This section walks through creating and testing data processing jobs for efficient and governed data transformation.

Create initial data processing job

This job performs the first processing job in the ETL pipeline, transforming raw customer, transaction, and clickstream data and writing the cleaned output to Amazon S3 in Parquet format. Complete the following steps to create the job:

In SageMaker Unified Studio, go to your project.
On the top menu, choose Build, and under Data Analysis & Integration, choose Data processing jobs.
Choose Create job from notebooks.
Under Choose project files, choose Browse files.
Locate and select etl_initial_processing_job.ipynb (the notebook saved earlier in JupyterLab), then choose Select and Next.
Configure the job settings:
1. For Name, enter a name (for example, job-1).
2. For Description, enter a description (for example, Initial ETL job for customer data processing).
3. For IAM Role, choose the project role (default).
4. For Type, choose Spark.
5. For AWS Glue version, use version 5.0.
6. For Language, choose Python.
7. For Worker type, use G.1X.
8. For Number of Instances, set to 10.
9. For Number of retries, set to 0.
10. For Job timeout, set to 480.
11. For Compute connection, choose project.spark.compatibility.
12. Under Advanced settings, turn on Continuous logging.
Leave the remaining settings as default, then choose Submit.

After the job is created, a confirmation message will appear indicating that job-1 was created successfully.

Create AWS Glue Data Quality job

This job runs data quality checks on the transformed datasets using AWS Glue Data Quality. Rulesets validate completeness and uniqueness for key fields. Complete the following steps to create the job:

In SageMaker Unified Studio, go to your project.
On the top menu, choose Build, and under Data Analysis & Integration, choose Data processing jobs.
Choose Create job, Code-based job, and Create job from files.
Under Choose project files, choose Browse files.
Locate and select etl_glue_data_quality.ipynb, then choose Select and Next.
Configure the job settings:
For Name, enter a name (for example, job-2).
For Description, enter a description (for example, Data quality checks using AWS Glue Data Quality).
For IAM Role, choose the project role.
For Type, choose Spark.
For AWS Glue version, use version 5.0.
For Language, choose Python.
For Worker type, use G.1X.
For Number of Instances, set to 10.
For Number of retries, set to 0.
For Job timeout, set to 480.
For Compute connection, choose project.spark.compatibility.
Under Advanced settings, turn on Continuous logging.
Leave the remaining settings as default, then choose Submit.

After the job is created, a confirmation message will appear indicating that job-2 was created successfully.

Test AWS Glue jobs

Test both jobs to make sure they execute successfully:

In SageMaker Unified Studio, go to your project.
On the top menu, choose Build, and under Data Analysis & Integration, choose Data processing jobs.
Select job-1 and choose Run job.
Monitor the job execution and verify it completes successfully.
Similarly, select job-2 and choose Run job.
Monitor the job execution and verify it completes successfully.

Add EMR Serverless compute

In the ETL pipeline, we use EMR Serverless to perform compute-intensive transformations and aggregations on large datasets. It automatically scales resources based on workload, offering high performance with simplified operations. By integrating EMR Serverless with SageMaker Unified Studio, you can simplify the process of running Spark jobs interactively using Jupyter notebooks in a serverless environment.

This section walks through the steps to configure EMR Serverless compute within SageMaker Studio and use it for executing distributed data processing jobs.

Configure EMR Serverless in SageMaker Unified Studio

To use EMR Serverless for processing in the project, follow these steps:

In the navigation pane on Project Overview, choose Compute.
On the Data processing tab, choose Add compute and Create new compute resources.
Select EMR Serverless and choose Next.
Configure EMR Serverless settings:
For Compute name, enter a name (for example, etl-emr-serverless).
For Description, enter a description (for example, EMR Serverless for advanced data processing).
For Release label, choose emr-7.8.0.
For Permission mode, choose Compatibility.
Choose Add Compute to complete the setup.

After it’s configured, the EMR Serverless compute will be listed with the deployment status Active.

emr serverless

Create and run notebook with EMR Serverless

After you create the EMR Serverless compute, you can run PySpark-based data transformation jobs using a Jupyter notebook to perform large-scale data transformations. This job reads cleaned customer, transaction, and clickstream datasets from Amazon S3, performs aggregations and scoring, and writes the final analytics outputs back to Amazon S3 in both Parquet and CSV formats.Complete the following steps to create a notebook for EMR Serverless processing:

On the top menu, under Build, choose JupyterLab.
Choose File, New, and Notebook.
Set Kernel as Python 3, Connection type as PySpark, and Compute as emr-s.etl-emr-serverless.

Enter the following PySpark script to run your data transformation job on EMR Serverless. Provide the name of your S3 bucket:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("CustomerAnalytics").getOrCreate()

customers = spark.read.parquet("s3://<bucket-name>/processed/customers/")
transactions = spark.read.parquet("s3://<bucket-name>/processed/transactions/")
clickstream = spark.read.parquet("s3://<bucket-name>/processed/clickstream/")

customer_spending = transactions.groupBy("customer_id").agg(
    F.count("transaction_id").alias("total_transactions"),
    F.sum("amount").alias("total_spent"),
    F.avg("amount").alias("avg_transaction_value"),
    F.datediff(F.current_date(), F.max("date")).alias("days_since_last_purchase")
)

customer_engagement = clickstream.groupBy("customer_id").agg(
    F.count("*").alias("total_clicks"),
    F.countDistinct("page").alias("unique_pages_visited"),
    F.count(F.when(F.col("action") == "purchase", 1)).alias("purchase_actions"),
    F.count(F.when(F.col("action") == "add_to_cart", 1)).alias("add_to_cart_actions")
)

customer_analytics = customers.join(customer_spending, on="customer_id", how="left").join(
    customer_engagement, on="customer_id", how="left")

customer_analytics = customer_analytics.na.fill(0, [
    "total_transactions", "total_spent", "total_clicks", 
    "unique_pages_visited", "purchase_actions", "add_to_cart_actions"
])

customer_analytics = customer_analytics.withColumn(
    "customer_value_score",
    (F.col("total_spent") * 0.5) + (F.col("total_transactions") * 0.3) + (F.col("purchase_actions") * 0.2)
)

customer_analytics.write.mode("overwrite").parquet("s3://<bucket-name>/analytics/customer_analytics/")

customer_summary = customer_analytics.select(
    "customer_id", "name", "email", "registration_date", 
    "total_transactions", "total_spent", "avg_transaction_value",
    "days_since_last_purchase", "total_clicks", "purchase_actions",
    "customer_value_score"
)

customer_summary.write.mode("overwrite").option("header", "true").csv("s3://<bucket-name>/analytics/customer_summary/")

print("EMR processing completed successfully")

Choose File, Save Notebook As, and save the file as shared/emr_data_transformation_job.ipynb.
Choose Run Cell to run the script.
Monitor the Script execution and verify it completes successfully.
Monitor the Spark job execution and ensure it completes without errors.

emr run

Add Redshift Serverless compute

With Redshift Serverless, users can run and scale data warehouse workloads without managing infrastructure. It is ideal for analytics use cases where data needs to be queried from Amazon S3 or integrated into a centralized warehouse. In this step, you add Redshift Serverless to the project for loading and querying processed customer analytics data generated in earlier stages of the pipeline. For more information about Redshift Serverless, see Amazon Redshift Serverless.

Set up Redshift Serverless compute in SageMaker Unified Studio

Complete the following steps to set up Redshift Serverless compute:

In SageMaker Unified Studio, choose the Compute tab within your project workspace (ETL-Pipeline-Demo).
On the SQL analytics tab, choose Add compute, then choose Create new compute resources to begin configuring your compute environment.
Select Amazon Redshift Serverless.
Configure the following:
1. For Compute name, enter a name (for example, ecommerce_data_warehouse).
2. For Description, enter a description (for example, Redshift Serverless for data warehouse).
3. For Workgroup name, enter a name (for example, redshift-serverless-workgroup).
4. For Maximum capacity, set to 512 RPUs.
5. For Database name, enter dev.
Choose Add Compute to create the Redshift Serverless resource.

After the compute is created, you can test the Amazon Redshift connection.

On the Data warehouse tab, confirm that redshift.ecommerce_data_warehouse is listed.
Choose the compute: redshift.ecommerce_data_warehouse.
On the Permissions tab, copy the IAM role ARN. You use this for the Redshift COPY command in the next step.

Create and execute querybook to load data into Amazon Redshift

In this step, you create a SQL script to load the processed customer summary data from Amazon S3 into a Redshift table. This enables centralized analytics for customer segmentation, lifetime value calculations, and marketing campaigns. Complete the following steps:

On the Build menu, under Data Analysis & Integration, choose Query editor.

Enter the following SQL into the querybook to create the customer_summary table in the public schema:

-- Create customer_summary table in public schema
CREATE TABLE IF NOT EXISTS public.customer_summary (
    customer_id INT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    registration_date DATE,
    total_transactions INT,
    total_spent DECIMAL(10, 2),
    avg_transaction_value DECIMAL(10, 2),
    days_since_last_purchase INT,
    total_clicks INT,
    purchase_actions INT,
    customer_value_score DECIMAL(10, 2)
);

Choose Add SQL to add a new SQL script.
Enter the following SQL into the querybook
```
TRUNCATE TABLE customer_summary;
```
Note: We truncate the customer_summary table to remove existing records and ensure a clean, duplicate-free reload of the latest aggregated data from S3 before running the COPY command.
Choose Add SQL to add a new SQL script.

Enter the following SQL to load the data into Redshift Serverless from your S3 bucket. Provide the name of your S3 bucket and IAM role ARN for Amazon Redshift:

-- Load data from S3 (replace with your bucket name and IAM role)
COPY public.customer_summary FROM 's3://<bucket-name>/analytics/customer_summary/'
IAM_ROLE 'arn:aws:iam::<Account-ID>:role/<your-redshift-role>'
FORMAT AS CSV
IGNOREHEADER 1
REGION 'us-east-1';

In the Query Editor, configure the following:
1. Connection: redshift.ecommerce_data_warehouse
2. Database: dev
3. Schema: public
Choose Choose to apply the connection settings.
Choose Run Cell for each cell to create the customer_summary table in the public schema and then load data from Amazon S3.
Choose Actions, Save, name the querybook final_data_product, and choose Save changes.

This completes the creation and execution of the Redshift data product using the querybook.

Create and manage the workflow environment

This section describes how to create a shared workflow environment and define a code-based workflow that automates a customer data pipeline using Apache Airflow within SageMaker Unified Studio. Shared environments facilitate collaboration among project members and centralized workflow management.

Create the workflow environment

Workflow environments must be created by project owners. After they’re created, members of the project can sync and use the workflows. Only project owners can update or delete workflow environments. Complete the following steps to create the workflow environment:

Choose Compute for your project.
On the Workflow environments tab, choose Create.
Review the configuration parameters and choose Create workflow environment.
Wait for the environment to be fully provisioned before proceeding It will take around 20 minutes to provision.

workflow

Create the code-based workflow

When the workflow environment is ready, define a code-based ETL pipeline using Airflow. This pipeline automates daily processing tasks across services like AWS Glue, EMR Serverless, and Redshift Serverless.

On the Build menu, under Orchestration, choose Workflows.
Choose Create new workflow, then choose Create workflow in code editor.
Configure Space and choose the instance type ml.t3.xlarge. This ensures your JupyterLab instance has at least 4 vCPUs and 4 GiB of memory.
Choose Configure and Restart Space to launch your environment.

sample_dag

The following script defines a daily scheduled ETL workflow that automates several actions:

Initial data transformation using AWS Glue
Data quality validation using AWS Glue (EvaluateDataQuality)
Advanced data processing with EMR Serverless using a Jupyter notebook
Loading transformed results into Redshift Serverless from a querybook

Replace the default DAG template with the following definition, ensuring that job names and input paths match the actual names used in your project:

from datetime import datetime
from airflow import DAG
from airflow.decorators import dag
from airflow.utils.dates import days_ago
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from workflows.airflow.providers.amazon.aws.operators.sagemaker_workflows import NotebookOperator
from sagemaker_studio import Project
# Get SageMaker Studio project IAM role
project = Project()
default_args = {
    'owner': 'data_engineer',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1
}
@dag(
    dag_id='customer_etl_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=days_ago(1),
    is_paused_upon_creation=False,
    tags=['etl', 'customer-analytics'],
    catchup=False
)
def customer_etl_pipeline():
    # Step 1: Initial data transformation using Glue
    initial_transformation = GlueJobOperator(
        task_id='initial_transformation',
        job_name='job-1',
        iam_role_arn=project.iam_role,
    )
    # Step 2: Data quality checks using Glue DQ
    data_quality_check = GlueJobOperator(
        task_id='data_quality_check',
        job_name='job-6',
        iam_role_arn=project.iam_role,
    )
    # Step 3: EMR Serverless notebook processing
    emr_processing = NotebookOperator(
        task_id='emr_processing',
        input_config={
            "input_path": "emr_data_transformation_job.ipynb",
            "input_params": {}
        },
        output_config={"output_formats": ['NOTEBOOK']},
        poll_interval=10,
    )
    # Step 4: Load to Redshift notebook
    redshift_load = NotebookOperator(
        task_id='redshift_load',
        input_config={
            "input_path": "final_data_product.sqlnb",
            "input_params": {}
        },
        output_config={"output_formats": ['NOTEBOOK']},
        poll_interval=10,
    )
    # Task dependencies
    initial_transformation >> data_quality_check >> emr_processing >> redshift_load
# Instantiate DAG
customer_etl_dag = customer_etl_pipeline()

Choose File, Save python file, name the file shared/workflows/dags/customer_etl_pipeline.py, and choose Save.

Deploy and run the workflow

Complete the following steps to run the workflow:

On the Build menu, choose Workflows.
Choose the workflow customer_etl_pipeline and choose Run.

scheduled

Running a workflow puts tasks together to orchestrate Amazon SageMaker Unified Studio artifacts. You can view multiple runs for a workflow by navigating to the Workflows page and choosing the name of a workflow from the workflows list table.

To share your workflows with other project members in a workflow environment, refer to Share a code workflow with other project members in an Amazon SageMaker Unified Studio workflow environment.

Monitor and troubleshoot the workflow

After your Airflow workflows are deployed in SageMaker Unified Studio, monitoring becomes essential for maintaining reliable ETL operations. The integrated Amazon MWAA environment provides comprehensive observability into your data pipelines through the familiar Airflow web interface, enhanced with AWS monitoring capabilities. The Amazon MWAA integration with SageMaker Unified Studio offers real-time DAG execution tracking, detailed task logs, and performance metrics to help you quickly identify and resolve pipeline issues. Complete the following steps to monitor the workflow:

On the Build menu, choose Workflows.
Choose the workflow customer_etl_pipeline.
Choose View runs to see all executions.
Choose a specific run to view detailed task status.

workflows-run

For each task, you can view the status (Succeeded, Failed, Running), start and end times, duration, and logs and outputs. The workflow is also visible in the Airflow UI, accessible through the workflow environment, where you can view the DAG graph, monitor task execution in real time, access detailed logs, and view the status.

Go to Workflows and select the workflow named customer_etl_pipeline.
From the Actions menu, choose Open in Airflow UI.

airflow-ui-smus

After the workflow completes successfully, you can query the data product in the query editor.

On the Build menu, under Data Analysis & Integration, choose Query editor.
Run select * from "dev"."public"."customer_summary"

query-editor

Observe the contents of the customer_summary table, including aggregated customer metrics such as total transactions, total spent, average transaction value, clicks, and customer value scores. This allows verification that the ETL and data quality pipelines loaded and transformed the data correctly.

Clean up

To avoid unnecessary charges, complete the following steps:

Delete a workflow environment.
If you no longer need it, delete the project.
After you delete the project, delete the domain.

Conclusion

This post demonstrated how to build an end-to-end ETL pipeline using SageMaker Unified Studio workflows. We explored the complete development lifecycle, from setting up fundamental AWS infrastructure—including Amazon S3 CORS configuration and IAM permissions—to implementing sophisticated data processing workflows. The solution incorporates AWS Glue for initial data transformation and quality checks, EMR Serverless for advanced processing, and Redshift Serverless for data warehousing, all orchestrated through Airflow DAGs. This approach offers several key benefits: a unified interface that consolidates necessary tools, Python-based workflow flexibility, seamless AWS service integration, collaborative development through Git version control, cost-effective scaling through serverless computing, and comprehensive monitoring tools—all working together to create an efficient and maintainable data pipeline solution.

By using SageMaker Unified Studio workflows, you can accelerate your data pipeline development while maintaining enterprise-grade reliability and scalability. For more information about SageMaker Unified Studio and its capabilities, refer to the Amazon SageMaker Unified Studio documentation.

About the authors

Optimizing Flink’s join operations on Amazon EMR with Alluxio

2026-02-03 Qingyuan Tang

Post Syndicated from Qingyuan Tang original https://aws.amazon.com/blogs/big-data/optimizing-flinks-join-operations-on-amazon-emr-with-alluxio/

When you’re working with data analysis, you often face the challenge of effectively correlating real-time data with historical data to gain actionable insights. This becomes particularly critical when you’re dealing with scenarios like e-commerce order processing, where your real-time decisions can significantly impact business outcomes. The complexity arises when you need to combine streaming data with static reference information to create a comprehensive analytical framework that supports both your immediate operational needs and strategic planning

To tackle this challenge, you can employ stream processing technologies that handle continuous data flows while seamlessly integrating live data streams with static dimension tables. These solutions enable you to perform detailed analysis and aggregation of data, giving you a comprehensive view that combines the immediacy of real-time data with the depth of historical context. Apache Flink has emerged as a leading stream computing platform that offers robust capabilities for joining real-time and offline data sources through its extensive connector ecosystem and SQL API.

In this post, we show you how to implement real-time data correlation using Apache Flink to join streaming order data with historical customer and product information, enabling you to make informed decisions based on comprehensive, up-to-date analytics.

We also introduce an optimized solution to automatically load Hive dimension table data into Alluxio Universal Flash Storage (UFS) through the Alluxio cache layer. This enables Flink to perform temporal joins on changing data, accurately reflecting the content of a table at specific points in time.

Solution architecture

When it comes to joining Flink SQL tables with stream tables, the lookup join is a go-to method. This approach is particularly effective when you need to correlate streaming data with static or slowly changing data. In Flink, you can use connectors like the Flink Hive SQL connector or the FileSystem connector to archive the scenario.

The following architecture shows general approach which we describe ahead：

Here’s how we do this:

We use offline data to construct a Flink table. This data could be from an offline Hive database table or from files stored in a system like Amazon S3. Concurrently, we can create a stream table from the data flowing in through a Kafka message stream
Use a batch cluster for offline data processing. In this example, we use an Amazon EMR cluster which creates a fact table in it. It also provides a Detail Wide Data (DWD) table which has been used as a Flink dynamic table to perform consequence processing after a lookup join
- It is typically located in the middle layer of a data warehouse, between the raw data contained in the Operational Data Store (ODS) and the highly aggregated data found in the Data Warehouse (DW), or Data Mart (DM).
- The primary purpose of the DWD layer is to support complex data analysis and reporting needs by providing a detailed and comprehensive data view.
- Both the fact table and DWD table are hive tables on Hadoop
Use a streaming cluster for the real-time processing. In this example, we use an Amazon EMR cluster to stream event ingestion and analyze it using Flink, using Flink Kafka connector and Hive connector to join the streaming event data and statics dimension data (fact table)

One of the key challenges encountered with this approach is related to the management of the lookup dimension table data. Initially, when the Flink application is started, this data is stored in the task manager’s state. However, during subsequent operations like continuous queries or window aggregations, the dimension table data isn’t automatically refreshed. This means that the operator must either restart the Flink application periodically or manually refresh the dimension table data in the temporary table. This step is crucial to ensure that the join operations and aggregations are always performed with the most current dimension data.

Another significant challenge with this approach is needing to pull the entire dimension table data and perform a cold start each time. This becomes particularly problematic when dealing with a large volume of dimension table data. For instance, when handling tables with tens of millions of registered users or tens of thousands of product SKU attributes, this process generates substantial input/output (IO) overhead. Consequently, it leads to performance bottlenecks, impacting the efficiency of the system.

Flink’s checkpointing mechanism processes the data and stores checkpoint snapshots of all the states during continuous queries or window aggregations, resulting in state snapshots data bloat.

Optimizing the solution

This post includes an optimized solution to address the aforementioned challenges, by automatically loading Hive dimension table data into the Alluxio UFS via the Alluxio cache layer. We join this data with Flink’s temporal joins to create a view on a changing table. This view reflects the content of a table at a specific point in time

Alluxio is a distributed cache engine for big data technology stacks. It provides a unified UFS that can connect to the underlying Amazon S3 and HDFS data. Alluxio UFS read and write operations warm up the distributed storage layers on S3 and HDFS and thus significantly increase throughput and reducing network overhead. Deeply integrated with upper level computing engines such as Hive, Spark, and Trino, Alluxio is an excellent cache accelerator for offline dimension data.

Additionally, we utilize Flink’s temporal table function to pass a time parameter. This function returns a view of the temporal table at the specified time. By doing so, when the main table of the real-time dynamic table is correlated with the temporal table, it can be associated with a specific historical version of the dimension data

Solution implementation details

For this post, we use “user behavior” log data in Kafka as real-time stream fact table data, and user information data on Hive as offline dimension table data. A demo with Alluxio + Flink temporal join is used to verify the Flink join optimized solution.

Real-time fact tables

For this demonstration, we utilize user behavior JSON data simulated by the open-source component json-data-generator. We write the data to Amazon Managed Kafka (Amazon MSK) in real-time. Using the Flink Kafka Connector, we convert this stream into a Flink stream table for continuous queries. This served as our fact table data for real-time joins.

A sample of the user behavior simulation data in JSON format is as follows:

[{          
	"timestamp": "nowTimestamp()",
	"system": "BADGE",
	"actor": "Agnew",
	"action": "EXIT",
	"objects": ["Building 1"],
	"location": "45.5,44.3",
	"message": "Exited Building 1"
}]

It includes user behavior information such as operation time, login system, user signature, behavioral activities, and service objects, locations, and related text fields. We create a fact table in Flink SQL with the main fields as follows:

CREATE TABLE logevent_source (`timestamp`  string, 
`system` string,
 actor STRING,
 action STRING
) WITH (
'connector' = 'kafka',
'topic' = 'logevent',
'properties.bootstrap.servers' = 'b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092 (http://b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092/)',
'properties.group.id' = 'testGroup6',
'scan.startup.mode'='latest-offset',
'format' = 'json'
);

Caching dimension tables with Alluxio

Amazon EMR provides solid integration with Alluxio. You can use the Amazon EMR bootstrap startup script to automatically deploy Alluxio components and start the Alluxio master and worker processes when an Amazon EMR cluster is created. For detailed installation and deployment steps, refer to the article Integrating Alluxio on Amazon EMR.

In an Amazon EMR cluster that integrates Alluxio, you may use Alluxio to create a cache table for the Hive offline dimension table as follows:

##Set up the client jar package in hive-env.sh:
$ export HIVE_AUX_JARS_PATH=/<PATH_TO_ALLUXIO>/client/alluxio-2.2.0-client.jar:${HIVE_AU

##Make sure the UFS is configured on the EMR cluster where Alluxio is installed and that the table/db path has been created:
alluxio fs mkdir alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/customer
alluxio fs chown hadoop:hadoop alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/customer

##On the AWS EMR cluster, create a Hive table path pointing to Alluxio namespace URI:
!connect jdbc:hive2://xxx.xxx.xxx.xxx:10000/default;
hive> CREATE TABLE customer(
    c_customer_sk             bigint,
    c_customer_id             string,
    c_current_cdemo_sk        bigint,
    c_current_hdemo_sk        bigint,
    c_current_addr_sk         bigint,
    c_first_shipto_date_sk    bigint,
    c_first_sales_date_sk     bigint,
    c_salutation              string,
    c_first_name              string,
    c_last_name               string,
    c_preferred_cust_flag     string,
    c_birth_day               int,
    c_birth_month             int,
    c_birth_year              int,
    c_birth_country           string,
    c_login                   string,
    c_email_address           string
)
    ROW FORMAT DELIMITED
    FIELDS TERMINATED BY '|'
    STORED AS TEXTFILE
    LOCATION 'alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/customer';
OK
Time taken: 3.485 seconds

As shown in the previous section, the Alluxio table location alluxio://ip-xxx-xx:19998/s3/customer points to the S3 path where the Hive dimension table is located; writing to the customer dimension table is automatically synchronized to the Alluxio cache.

After creating the Alluxio Hive offline dimension table, you can view the details of the Alluxio cache table by connecting to the Hive metadata through the Hive catalog in Flink SQL:

CREATE CATALOG hiveCatalog WITH (  'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/etc/hive/conf/',
    'hive-version' = '3.1.2',
    'hadoop-conf-dir'='/etc/hadoop/conf/'
);
-- set the HiveCatalog as the current catalog of the session
USE CATALOG hiveCatalog;
show create table customer;
create external table customer(
    c_customer_sk             bigint,
    c_customer_id             string,
    c_current_cdemo_sk        bigint,
    c_current_hdemo_sk        bigint,
    c_current_addr_sk         bigint,
    c_first_shipto_date_sk    bigint,
    c_first_sales_date_sk     bigint,
    c_salutation              string,
    c_first_name              string,
    c_last_name               string,
    c_preferred_cust_flag     string,
    c_birth_day               int,
    c_birth_month             int,
    c_birth_year              int,
    c_birth_country           string,
    c_login                   string,
    c_email_address           string
) 
row format delimited fields terminated by '|'
location 'alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/30/customer' 
TBLPROPERTIES (
  'streaming-source.enable' = 'false',  
  'lookup.join.cache.ttl' = '12 h'
)

As shown in the preceding code, the location path of the dimension table is the UFS cache path Uniform Resource Identifier (URI). When the business program reads and writes the dimension table, Alluxio automatically updates the customer dimension table data in the cache and asynchronously writes it to the Alluxio backend storage path of the S3 table to achieve table data synchronization in the data lake.

Flink temporal table join

Flink temporal table is also a type of dynamic table. Each record in the temporal table is correlated with one or more time fields. When we join the fact table and the dimension table, we usually need to obtain real-time dimension table data for the lookup join. Thus, when creating or joining a table, we usually need to use the proctime() function to specify the time field of the fact table. When we join the tables, we use the syntax of FOR SYSTEM_TIME AS OF to specify the time version of the fact table that corresponds to the time of the lookup dimension table.

For this post, the customer information is a changing dimension table in the Hive offline table, whereas the customer behavior is the fact table in Kafka. We specified the time field with proctime() in the Flink Kafka source table. Then when joining the Flink Hive table, we used FOR SYSTEM_TIME AS OF to specify the time field of the lookup Kafka source table to allow us to realize the Flink temporal table join operation

As shown in the following code, a fact table of user behavior is created through the Kafka Connector in Flink SQL. The ts field refers to the timestamp when the temporal table is joined：

CREATE TABLE logevent_source (`timestamp`  string, 
`system` string,
 actor STRING,
 action STRING,
 ts as PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'logevent',
'properties.bootstrap.servers' = 'b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092,b-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092 (http://b-6.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-5.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092%2Cb-1.msk06.dr04w4.c3.kafka.ap-southeast-1.amazonaws.com:9092/)',
'properties.group.id' = 'testGroup-01',
'scan.startup.mode'='latest-offset',
'format' = 'json'
);

The Flink offline dimension table and the streaming real-time table are joined as follows:

select a.`timestamp`,a.`system`,a.actor,a.action,b.c_login from 
       (select *, proctime() as proctime from user_logevent_source) as a 
 left join customer  FOR SYSTEM_TIME AS OF a.proctime as b on a.actor=b.c_last_name;

When the fact table logevent_source joins the lookup dimension table, the proctime function ensures real-time joins by obtaining the latest dimension table version. This dimension data, cached in Alluxio, delivers significantly better read performance than direct S3 access.

At the same time, the dimension table data is already cached in Alluxio; the read performance is much better than offline data read on S3.

The comparison test shows that Alluxio cache brings a clear performance advantage by switching the S3 and Alluxio paths of the customer dimension table through Hive

You can easily switch the local and cache location paths with alter table in hive cli:

alter table customer set location "s3://xxxxxx/data/s3/30/customer";
alter table customer  set location "alluxio://ip-xxx-xxx-xxx-xxx.ap-southeast-1.compute.internal:19998/s3/30/customer";

You can also select the Task Manager log from the Flink dashboard for a split test.

The performance of the fact table load was doubled through the implementation of optimized data processing techniques.

Before caching (S3 path read): 5s load time

2022-06-29 02:54:34,791 INFO  com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem           [] - Opening 's3://salunchbucket/data/s3/30/customer/data-m-00029' for reading
2022-06-29 02:54:39,971 INFO  org.apache.flink.table.filesystem.FileSystemLookupFunction   [] - Loaded 433000 row(s) into lookup join cache

After caching (Alluxio read): 2s load time

2022-06-29 03:25:14,476 INFO  com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem           [] - Opening 's3://salunchbucket/data/s3/30/customer/data-m-00029' for reading
2022-06-29 03:25:16,397 INFO  org.apache.flink.table.filesystem.FileSystemLookupFunction   [] - Loaded 433000 row(s) into lookup join cache

The timeline on JobManager clearly shows the difference in execution duration under Alluxio and S3 paths.

For single task query ,we accelerate by more than 1 times using this solution. The overall job performance improvement is even more visible.

Other optimalizations to consider

Implementing a continuous join requires pulling dimension data every time. Does it lead to Flink’s checkpoint state bloat that can cause Flink TaskManager RocksDB to explode or memory overflow.
In Flink, the state comes with a TTL mechanism. You can set a TTL expiration policy to trigger Flink to clean up expired state data. Flink SQL can be set using the hint method.

insert into logevent_sink
select a.`timestamp`,a.`system`,a.actor,a.action,b.c_login from 
(select *, proctime() as proctime from logevent_source) as a 
  left join 
customer/*+ OPTIONS('lookup.join.cache.ttl' = '5 min')*/  FOR SYSTEM_TIME AS OF a.proctime as b 
on a.actor=b.c_last_name;

Flink Table/Streaming API is similar:

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter() 
    .build();
ValueStateDescriptor<Long> lastUserLogin = 
    new ValueStateDescriptor<>("lastUserLogin", Long.class);
lastUserLogin.enableTimeToLive(ttlConfig);
StreamTableEnvironment.getConfig().setIdleStateRetentionTime(min, max);

Restart the lookup join after the configuration. As you can see from the Flink TM log, after TTL expires, it triggers clean-up and re-pull the Hive dimension table data：

2022-06-29 04:17:09,161 INFO  org.apache.flink.table.filesystem.FileSystemLookupFunction   
[] - Lookup join cache has expired after 5 minute(s), reloading

In addition, you can reduce the number of checkpoint snapshots by configuring Flink state retention and thereby reduce the amount of space taken up by state at the time of snapshot.

Flink job configuration as follow：
-D state.checkpoints.num-retained=5

After the configuration, you can see that in the S3 checkpoint path, the Flink job automatically cleans up historical snapshots and keeps the most recent 5 snapshots, thus ensuring that checkpoint snapshots do not accumulate.

[hadoop@ip-172-31-41-131 ~]$ aws s3 ls s3://salunchbucket/data/checkpoints/7b9f2f9becbf3c879cd1e5f38c6239f8/
                           PRE chk-3/
                           PRE chk-4/
                           PRE chk-5/
                           PRE chk-6/
                           PRE chk-7/

Summary

Customers implementing Flink streaming framework to join dimension and real-time fact tables frequently encounter performance challenges. In this post, we presented an optimized solution that uses Alluxio’s caching capabilities to automatically load Hive dimension table data into the UFS cache. By integrating with Flink temporal table joins, dimension tables are transformed into time-versioned views, effectively addressing performance bottlenecks in traditional implementations.

About the author

Reduce EMR HBase upgrade downtime with the EMR read-replica prewarm feature

2026-01-29 Suthan Phillips

Post Syndicated from Suthan Phillips original https://aws.amazon.com/blogs/big-data/reduce-emr-hbase-upgrade-downtime-with-the-emr-read-replica-prewarm-feature/

HBase clusters on Amazon Simple Storage Service (Amazon S3) need regular upgrades for new features, security patches, and performance improvements. In this post, we introduce the EMR read-replica prewarm feature in Amazon EMR and show you how to use it to minimize HBase upgrade downtime from hours to minutes using blue-green deployments. This approach works well for single-cluster deployments where minimizing service interruption during infrastructure changes is important.

Understanding HBase operational challenges

HBase cluster upgrades have required complete cluster shutdowns, resulting in extended downtime while regions initialize and RegionServers come online. Version upgrades require a complete cluster switchover, with time-consuming steps that include loading and verifying region metadata, performing HFile checks, and confirming proper region assignment across RegionServers. During this critical period—which can extend to hours depending on cluster size and data volume—your applications are completely unavailable.

The challenge doesn’t stop at version upgrades. You must regularly apply security patches and kernel updates to maintain compliance. For Amazon EMR 7.0 and later clusters running on Amazon Linux 2023, instances don’t automatically install security updates after launch; they remain at the patch level from cluster creation time. AWS recommends periodically recreating clusters with newer AMIs, requiring the same hard cutover and downtime risks as a full version upgrade. Similarly, when you need to use different instance types, traditional approaches mean taking your cluster offline.

Solution overview

Amazon EMR 7.12 introduces read-replica prewarm, a new feature that tackles these challenges. This feature lets you make infrastructure changes to Apache HBase on Amazon S3 at scale while reducing downtime risk and maintaining data consistency.

With read-replica prewarm, you can prepare and validate your changes in a read-replica cluster before promoting it to active status, cutting service interruption from hours to minutes. You will learn how to prepare your read-replica cluster with the target version, execute cutover procedures that minimize downtime, and verify successful migration before completing the switchover.

Read-replica prewarm architecture

The following diagram shows the architecture and workflow. Both primary and read-replica clusters interact with the same Amazon S3 storage, accessing the same S3 bucket and root directory.

Distributed locking confirms only one HBase cluster can write at a time (for clusters version 7.12.0 and later). The read-replica cluster performs full HBase region initialization without time pressure, and after promotion, the read replica becomes the active writer as shown in the following diagram.

Implementation steps HBase cluster upgrade

Now that you understand how read-replica prewarm works and the architecture behind it, let’s put this knowledge into practice. You will follow a process that consists of three main phases: preparation, cutover, and verification. Each phase includes specific steps, shown in the following figure, that you will execute in sequence to complete the migration.

Phase 1: Preparation

Before starting the migration, prepare both your primary cluster and launch a new read-replica cluster. Each step in this phase builds toward confirming that your new cluster can properly access and serve your existing data.

Run major compactions on tables to verify regions are not in SPLIT state
Run major compactions to consolidate data files and verify regions are not in SPLIT state. Split regions can cause assignment conflicts during migration, so resolving them at the start helps maintain cluster stability throughout the transition.
```
echo “major_compact 'tablename'” | hbase shell
```
Run catalog_janitor to clean up stale regions
Execute the catalog_janitor process (HBase’s built-in maintenance tool) to remove stale region references from the metadata. Cleaning up these references prevents confusion during region assignment in the read-replica cluster.
```
echo “catalogjanitor_run” | hbase shell
```
Confirm no inconsistencies in the primary HBase cluster
Verify cluster integrity before migration:
```
sudo -u hbase hbase hbck > hbck_report.txt
```
Running the HBase Consistency Check tool version 2 (HBCK2) performs a diagnostic scan that identifies and reports problems in metadata, regions, and table states, confirming your cluster is ready for migration.
Launch HBase read-replica cluster with the target version connecting to the same HBase root directory in Amazon S3 as the primary cluster
Launch a new HBase cluster with the target version and configure it to connect to the same S3 root directory as the primary cluster. Confirm that read-only mode is enabled by default as shown in the following screenshot.

If you are using AWS Command Line Interface (AWS CLI), you can enable the read replica while launching the Amazon EMR HBase on the Amazon S3 cluster by setting the hbase.emr.readreplica.enabled.v2 parameter to true in the HBase classification as shown in the following example:
```
{
    "Classification": "hbase",
    "Properties": {
      "hbase.emr.readreplica.enabled.v2": "true",
      "hbase.emr.storageMode": "s3"
    }
}
```
Run meta refresh in this read-replica HBase cluster
```
echo "refresh_meta" | hbase shell
```
You’re creating a parallel environment with the new version that can access existing data without modification risk, allowing validation before committing to the upgrade.
Validate the read-replica and verify that regions show OPEN status and are properly assigned:
Execute sample read operations against your key tables to confirm the read replica can access your data correctly. In the HBase Master UI, verify that regions show OPEN status and are properly assigned to RegionServers. You should also confirm that the total data size matches your previous cluster to verify complete data visibility.

Prepare for cutover on primary cluster
Disable balancing and compactions on the primary cluster:

echo "balance_switch false" | hbase shell
echo "compaction_switch false" | hbase shell

Preventing background operations from changing data layout or triggering region movements maintains a consistent state during the migration window.

Take snapshots of your tables for rollback capability:

# For each table
echo "snapshot 'table_name', 'table_name_pre_migration_$(date +%Y%m%d)'" | hbase shell
# For system tables
echo "snapshot 'hbase:meta', 'meta_pre_migration_$(date +%Y%m%d)'" | hbase shell
echo "snapshot 'hbase:namespace', 'namespace_pre_migration_$(date +%Y%m%d)'" | hbase shell

These snapshots enable point-in-time recovery if you discover issues after migration.

Run meta refresh and refresh hfiles on the read replica:
```
echo "refresh_meta" | hbase shell
hbase org.apache.hadoop.hbase.client.example.RefreshHFilesClient "table_name'"
```
Refreshing confirms the read replica has the most current region assignments, table structure, and HFile references before taking over production traffic.
Check for inconsistencies in the read-replica cluster
Run the HBCK2 tool on the read-replica cluster to identify potential issues:
```
sudo -u hbase hbase hbck > hbck_report.txt
```
When a read replica is created, both the primary and replica clusters show metadata inconsistencies referencing each other’s meta folders: “There is a hole in the region chain”. The primary cluster complains about meta_<read-replica-cluster-id>, while the read replica complains about the primary’s meta folder. This inconsistency doesn’t impact cluster operations but shows up in hbck reports. For a clean hbck report after switching to the read replica and terminating the primary cluster, manually delete the old primary’s meta folder from Amazon S3 after taking a backup of it.

Additionally, check the HBase Master UI to visually confirm cluster health. Verifying the read-replica cluster has a clean, consistent state before promotion prevents potential data access issues after cutover.

Phase 2: Cutover

Perform the actual migration by shutting down the primary cluster and promoting the read replica. The steps in this phase minimize the window when your cluster is unavailable to applications.

Remove the primary cluster from DNS routing
Update DNS entries to direct traffic away from the primary cluster, preventing new requests from reaching it during shutdown.
Flush in-memory data to Amazon S3
Flush in-memory data to confirm durability in Amazon S3:
```
# Flush application data  
echo "flush 'usertable'" | hbase shell
# Flush system tables
echo "flush 'hbase:meta'" | hbase shell
echo "flush 'hbase:namespace'" | hbase shell
```
Flushing forces data still in memory (in MemStores, HBase’s write cache) to be written to persistent storage (Amazon S3), preventing data loss during the transition between clusters.
Terminate the primary cluster
Terminate the primary cluster after confirming the data is persisted to Amazon S3. This step releases resources and eliminates the possibility of split-brain scenarios where both clusters might accept writes to the same dataset.
Promote the read replica to active status
Convert the read replica to read-write mode:
```
echo "readonly_switch false" | hbase shell  
echo "readonly_state" | hbase shell  # Verify the switch was successful
```
The promotion process automatically refreshes meta and HFiles, capturing final changes from the flush operations and confirming complete data visibility.

When you promote the cluster, it transitions from read-only to read-write mode, allowing it to accept application write operations and fully replace the old cluster’s functionality.
Update DNS to point to the new active cluster
Update DNS entries to direct traffic to the new active cluster. Routing client traffic to the new cluster restores service availability and completes the migration from the application perspective.

Phase 3: Validation

With your new cluster now active, you’re ready to verify that everything is working correctly before declaring the migration complete.

Execute test write operations to confirm the cluster accepts writes properly. Check the HBase Master UI to verify regions are serving both read and write requests without errors. At this point, your migration to the new Amazon EMR release is complete, and your applications can connect to the new cluster and resume normal read-write operations.

Key benefits

The read-replica prewarm approach delivers several important advantages over traditional HBase upgrade methods. Most notably, you can reduce service interruption from hours to minutes by preparing your new cluster in parallel with your running production environment.

Before committing to the upgrade, you can thoroughly test that data is readable and accessible in the new version. The system loads and assigns regions before activation, eliminating the lengthy startup time that traditionally causes extended downtime. This pre-warming process means your new cluster is ready to serve traffic immediately upon promotion.

You also gain the ability to validate multiple aspects of your deployment before cutover, including data integrity, read performance, cluster stability, and configuration correctness. This validation happens while your production cluster continues serving traffic, reducing the risk of discovering issues during your maintenance window.

For testing and validation workflows, you can run parallel testing environment by creating multiple HBase read replicas. However, you should verify that only one HBase cluster remains in read-write mode to the Amazon S3 data store to prevent data corruption and consistency issues.

Rollback procedures

Always thoroughly test your HBase rollback procedures before implementing upgrades in production environments.

When rolling back HBase clusters in Amazon EMR, you have two primary options.

Option 1 involves launching a new cluster with the previous HBase version that points to the same Amazon S3 data location as the upgraded cluster. This approach is straightforward to implement, preserves data written before and after the upgrade attempt, and offers faster recovery with no additional storage requirements. However, it risks encountering data compatibility issues if the upgrade modified data formats or metadata structures, potentially leading to unexpected behavior.
Option 2 takes a more cautious approach by launching a new cluster with the previous HBase version and restoring from snapshots taken before the upgrade. This method guarantees a return to a known, consistent state, eliminates version compatibility risks, and provides complete isolation from corruption introduced during the upgrade process. The tradeoff is that data written after the snapshot was taken will be lost, and the restoration process requires more time and planning.

For production environments where data integrity is paramount, the snapshot-based approach (option 2) is generally preferred despite the potential for some data loss.

Considerations

Store file tracking migration: Migrating from Amazon EMR 7.3 (or earlier) requires disabling and dropping the hbase:storefile table on the primary cluster, then flushing metadata. When launching the new read-replica cluster, configure the DefaultStoreFileTracker implementation using the hbase.store.file-tracker.impl property. When operational, run change_sft commands to switch tables to FILE tracking method, providing seamless data file access during migration.
Multi-AZ deployments: Consider network latency and Amazon S3 access patterns when deploying read replicas across Availability Zones. Cross-AZ data transfer might impact read latency for the read-replica cluster.
Cost impact: Running parallel clusters during migration incurs additional infrastructure costs until the primary cluster is terminated.
Disabled tables: The disabled state of tables in the primary cluster is a cluster-specific administrative property that isn’t propagated to the read-replica cluster. If you want them disabled in the read replica, you must explicitly disable them.
Amazon EMR 5.x cluster upgrade: Direct upgrade from Amazon EMR 5.x to Amazon EMR 7.x using this feature isn’t supported because of the major HBase version change from 1.x to 2.x. For upgrading from Amazon EMR 5.x to Amazon EMR 7.x, follow the steps in our best practices: AWS EMR Best Practices – HBase Migration

Conclusion

In this post, we showed you how the read-replica prewarm feature of Amazon EMR 7.12 improves HBase cluster operations by minimizing the hard cutover constraints that make infrastructure changes challenging. This feature gives you a consistent blue-green deployment pattern that reduces risk and downtime for version upgrades and security patches.

When you can thoroughly validate changes before committing to them and reduce service interruption from hours to minutes, you can maintain HBase infrastructure more confidently and efficiently. You can now take a more proactive approach to cluster maintenance, security compliance, and performance optimization with greater confidence in your operational processes.

To learn more about Amazon EMR and HBase on Amazon S3, visit the Amazon EMR documentation. To get started with read replicas, see the HBase on Amazon S3 guide .

About the authors

Secure Apache Spark writes to Amazon S3 on Amazon EMR with dynamic AWS KMS encryption

2026-01-27 Pinxi Tai

Post Syndicated from Pinxi Tai original https://aws.amazon.com/blogs/big-data/secure-apache-spark-writes-to-amazon-s3-on-amazon-emr-with-dynamic-aws-kms-encryption/

When processing data at scale, many organizations use Apache Spark on Amazon EMR to run shared clusters that handle workloads across tenants, business units, or classification levels. In such multi-tenant environments, different datasets often require distinct AWS Key Management Service (AWS KMS) keys to enforce strict access controls and meet compliance requirements. At the same time, operational efficiency might drive these organizations to consolidate their data pipelines. Instead of running separate Spark jobs for each dataset, it could be more efficient to run a single job on Amazon EMR that processes inputs once and writes multiple outputs to Amazon Simple Storage Service (Amazon S3), each encrypted with its own KMS key.

Although consolidating multiple datasets in one Spark job reduces orchestration overhead and simplifies code maintenance, you might encounter challenges with encryption configurations. By default, the EMRFS and S3A file system clients cache their settings, which can cause encryption keys to persist incorrectly across writes. This means if you change the encryption key between writes to Amazon S3 in an Apache Hadoop environment, some output files can end up encrypted with unintended keys, leading to possible security and compliance concerns.

In this post, we show how to securely write data to Amazon S3 from Spark jobs running on Amazon EMR, while dynamically managing different KMS keys for encryption. We discuss three approaches to solve this challenge and how to choose the right solution for your use case.

Amazon S3 server-side encryption options with Amazon EMR

When writing data to Amazon S3 from Amazon EMR, you can choose from multiple server-side encryption options. The two most commonly used options are:

Server-side encryption with Amazon S3 managed keys (SSE-S3) – Amazon S3 manages the encryption keys for you
Server-side encryption with KMS keys (SSE-KMS) – AWS KMS manages the keys, and you can use custom KMS keys with fine-grained access control

When running Spark jobs on Amazon EMR, data writes to Amazon S3 occur through one of the following file system implementations:

EMRFS – The default implementation for Amazon EMR versions below 7.10.0
S3A – The default implementation starting from Amazon EMR 7.10.0

Both implementations provide configuration properties to control server-side encryption. The following tables show how to specify the KMS key for SSE-KMS encryption.

For EMRFS (default in Amazon EMR versions below 7.10.0), refer to the following table.

Property	Description
`fs.s3.enableServerSideEncryption`	Enables server-side encryption. Defaults to SSE-S3 if no KMS key is provided.
`fs.s3.serverSideEncryption.kms.keyId`	Specifies the KMS key ID or ARN for SSE-KMS encryption.

For S3A (default in Amazon EMR starting from 7.10.0), refer to the following table.

Property	Description
`fs.s3a.encryption.algorithm`	Specifies the encryption algorithm.
`fs.s3a.encryption.key`	Specifies the KMS key ID or ARN for SSE-KMS encryption.

Starting from the Amazon EMR 7.10.0 release, the S3A file system has replaced EMRFS as the default EMR S3 connector. For more information, refer to Migration Guide: EMRFS to S3A Filesystem.

Challenges to prevent encryption key reuse due to file system caching

In practice, a unified Spark job might write outputs for multiple tenants or classifications in a single run. In this situation, applying the correct encryption key for each output is critical to maintaining compliance and enforcing isolation in multi-tenant S3 buckets without the complexity of managing separate Spark jobs for each dataset.

When Spark executors write to Amazon S3, they use a file system (EMRFS or S3A) client that is cached and reused for performance optimization. The problem is that each file system instance keeps the encryption settings it was first created with. Each executor’s Java Virtual Machine (JVM) creates and caches a file system (and its underlying S3 client) for a given S3 bucket. This cached instance, along with its encryption configuration, persists throughout the executor’s lifecycle. If you change the encryption key in Spark after some data has been written, the existing cached client can’t pick up the new key.

For example, the following PySpark code first creates a Spark session with S3 server-side encryption enabled. It then writes a DataFrame to two different folders within the same S3 bucket amzn-s3-demo-bucket1, but with different KMS keys. The first write operation writes the data to folder1 using kmsKey1 for encryption, and the second write operation writes to folder2 using kmsKey2.

# Pseudo-code: setting different keys for successive writes to different folders within the same S3 bucket

spark = SparkSession.builder \
.appName("Write data to S3 with KMS") \
.config("spark.hadoop.fs.s3.enableServerSideEncryption", "true") \
.getOrCreate()

df.write.option('fs.s3.serverSideEncryption.kms.keyId', kmsKey1).save("s3://amzn-s3-demo-bucket1/folder1/")
df.write.option('fs.s3.serverSideEncryption.kms.keyId', kmsKey2).save("s3://amzn-s3-demo-bucket1/folder2/")

You might expect files in folder1/ to use kmsKey1 and files in folder2/ to use kmsKey2. But due to caching, the second write can still use the client configured with kmsKey1. This leads to mixed or incorrect encryption key usage across outputs.

Solution overview

Our objective is to achieve correct encryption of each output S3 object with its intended KMS key, even when a single Spark job writes multiple outputs. To implement this, you can use one of the following approaches:

Disable file system cache – Turn off S3 client caching so a new client is created for each write, picking up the current key
Separate Spark applications or sessions – Run a separate Spark application (or session) for each distinct encryption key, so each client is initialized fresh
Use S3 bucket default encryption – Configure bucket-level SSE-KMS with the desired key so Amazon S3 automatically applies the correct encryption key

Each method offers a different balance of implementation complexity, performance, and flexibility. The following sections provide detailed implementation steps and considerations for each approach.

Method 1: Disable file system cache

Disabling the file system cache forces Spark to create a new S3 client for each write, which applies the updated encryption settings. This can be done using a Spark configuration or EMR cluster settings.

The property name for disabling the cache depends on your URI scheme (s3:// or s3a://), not on your choice of file system (EMRFS or S3A). The following table summarizes which configuration property name you should use to disable the cache.

Properties for s3:// URI scheme	Properties for s3a:// URI scheme
`fs.s3.impl.disable.cache`	`fs.s3a.impl.disable.cache`
`spark.hadoop.fs.s3.impl.disable.cache`	`spark.hadoop.fs.s3a.impl.disable.cache`

To use this method, complete the following steps:

Disable the file system cache in Spark configuration.
For s3:// scheme, you can manually set spark.hadoop.fs.s3.impl.disable.cache=true, for example (PySpark):

# PySpark example for "s3://"
# Create Spark session
spark = SparkSession.builder \
.appName("Write data to S3 with KMS") \
.config("spark.hadoop.fs.s3.impl.disable.cache", "true") \
.getOrCreate()

Alternatively, you can use the following spark-defaults configuration classification:

[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.hadoop.fs.s3.impl.disable.cache": "true"
    }
  }
]

For s3a:// scheme, you can manually set spark.hadoop.fs.s3a.impl.disable.cache=true, for example (PySpark):

# PySpark example for "s3a://"
# Create Spark session
spark = SparkSession.builder \
.appName("Write data to S3 with KMS") \ 
.config("spark.hadoop.fs.s3a.impl.disable.cache", "true") \
.getOrCreate()

Alternatively, you can use the following spark-defaults configuration classification:

[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.hadoop.fs.s3a.impl.disable.cache": "true"
    }
  }
]

Instead of disabling the cache specifically for Spark applications, you can optionally configure the EMR cluster’s core-site.xml to disable the file system cache globally at cluster level. You must configure the /etc/hadoop/conf/core-site.xml file on the primary nodes of your EMR cluster. For example, when creating or modifying the cluster, use the following configuration.
For s3:// scheme:
```
[
  {
    "Classification": "core-site",
    "Properties": {
      "fs.s3.impl.disable.cache": "true"
    }
  }
]
```
For s3a:// scheme:
```
[
  {
    "Classification": "core-site",
    "Properties": { 
      "fs.s3a.impl.disable.cache": "true"
    }
  }
]
```
Enable SSE-KMS encryption.
For EMRFS, set spark.hadoop.fs.s3.enableServerSideEncryption=true for Spark applications only or use the following configuration to enable encryption at cluster level:
```
[
  {
    "Classification": "emrfs-site",
    "Properties": {
      "fs.s3.enableServerSideEncryption": "true"
    }
  }
]
```
For S3A, set spark.hadoop.fs.s3a.encryption.algorithm=SSE-KMS for Spark applications only or use the following configuration to enable encryption at cluster level:
```
[
  {
    "Classification": "core-site",
    "Properties": {
      "fs.s3a.encryption.algorithm": "SSE-KMS"
    }
  }
]
```
When using EMRFS with fs.s3.impl.disable.cache=true, you must also disable the EMRFS S3-optimized committer to avoid errors. You can do this by either manually setting spark.sql.parquet.fs.optimized.committer.optimization-enabled=false or using the following spark-defaults configuration classification:
```
[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.sql.parquet.fs.optimized.committer.optimization-enabled": "false"
    }
  }
]
```

For more information about configuring applications on EMR clusters, refer to Configure applications when you create a cluster and Reconfigure an instance group in a running cluster.

Considerations

Use this method when you need to write data to an S3 bucket using multiple KMS keys within a single Spark application. This is a quick, straightforward implementation that works well for the following use cases:

Testing environments and debugging sessions
Proof-of-concept demonstrations
Low-volume or one-time jobs where write performance is not critical
Workloads that frequently switch encryption keys to write data to different S3 prefixes within the same bucket

Before implementation, consider the following performance impacts:

Increased latency for each write
Additional S3 API operations
Extra connection overhead

Although this method provides a pragmatic solution when splitting work into separate Spark applications isn’t feasible, we don’t recommend it for high-throughput or latency-sensitive production workloads. The increased API traffic can lead to higher costs and potential throttling. For production implementations, consider Method 2 and Method 3.

Method 2: Use separate Spark applications or sessions

When writing data with multiple encryption keys, use a separate Spark application (or Spark session) for each distinct key. The file system needs to be initialized with the correct encryption key in a fresh JVM context when writing data with different keys. You can achieve this by either submitting a separate Spark application or starting a new Spark session. This enables the S3 client to be created with the intended encryption key.

Complete the following steps:

Divide the write tasks by KMS key. For example, prepare separate DataFrames or filter logic for each key.

Submit separate jobs. Choose either of the following options:

Use spark-submit commands. For example:

# For EMRFS
spark-submit –conf spark.hadoop.fs.s3.serverSideEncryption.kms.keyId=kmsKey1 job1.py
spark-submit --conf spark.hadoop.fs.s3.serverSideEncryption.kms.keyId=kmsKey2 job2.py

# For S3A
spark-submit –conf spark.hadoop.fs.s3a.server-side-encryption.key=kmsKey1 job1.py
spark-submit --conf spark.hadoop.fs.s3a.server-side-encryption.key=kmsKey2 job2.py

Use Spark sessions in code (PySpark example):

# For EMRFS
for kmsKey in [kmsKey1, kmsKey2]:
    spark = SparkSession.builder \
        .appName("Write data to S3 with KMS") \
        .config("spark.hadoop.fs.s3.enableServerSideEncryption", "true") \
        .config("spark.hadoop.fs.s3.serverSideEncryption.kms.keyId", kmsKey) \
        .getOrCreate()
    write_df_for_key(kmsKey)  # Pseudocode for writing data for this key
    spark.stop()

# For S3A
for kmsKey in [kmsKey1, kmsKey2]:
    spark = SparkSession.builder \
        .appName("Write data to S3 with KMS") \
        .config("spark.hadoop.fs.s3a.encryption.algorithm", "SSE-KMS") \
        .config("spark.hadoop.fs.s3a.encryption.key", kmsKey) \
        .getOrCreate()
    write_df_for_key(kmsKey)  # Pseudocode for writing data for this key
    spark.stop()

Use your preferred workflow (such as AWS Step Functions, Apache Airflow, or a wrapper script) to launch jobs in sequence or parallel.

Considerations

Use this method when you need a production-grade solution for applying different KMS keys at scale. This approach maintains file system caching benefits and works well for the following use cases:

High-throughput or latency-sensitive workloads with frequent write operations
Scenarios requiring strong isolation between different KMS keys
Multi-tenant environments with separate compliance boundaries

This method creates fresh S3 clients with each new Spark application or session. Compared to Method 1, it offers several advantages:

Avoids per-write connection and API overhead
Maintains full compatibility with the EMRFS S3-optimized committer
Enforces credential boundaries between workloads and improves operational and compliance isolation by assigning each Spark application or session a dedicated key

Before implementation, consider the following trade-offs:

Requires orchestration of multiple Spark applications and sessions or clusters
Involves higher resource overhead
Increases operational complexity

Choose this method when performance, cost predictability, and security isolation are more important than single-process simplicity.

Method 3: Use S3 bucket default encryption

Where possible, configure S3 bucket-level default encryption (SSE-KMS) with the desired KMS key to automatically encrypt objects written to that bucket.

Complete the following steps:

On the Amazon S3 console or using the AWS Command Line Interface (AWS CLI), enable default SSE-KMS for the bucket with the desired key. For instructions on enabling SSE-KMS for S3 buckets, refer to Configuring default encryption.
With default encryption enabled for the S3 bucket, you can write without specifying a KMS key in Spark. Amazon S3 can encrypt each object with the bucket’s key. Your Spark code only needs standard write operations.

Considerations

Use this method when your workloads can use a single KMS key per bucket. This approach works well for the following use cases:

Production environments prioritizing operational simplicity
Workloads where all data in a bucket shares the same security requirements
Scenarios where encryption configuration should be managed at the bucket level
Use cases that map naturally to per-bucket separation

This method provides several advantages:

Alleviates the need to configure encryption in Spark applications
Automatically applies the default KMS key for all writes
Simplifies encryption management

Before implementation, consider the following limitations:

You must have all data in one bucket to use the same encryption key
You can’t apply different keys to different prefixes within the same bucket

Choose this method when you need a simple, reliable approach that provides strong security while simplifying operational management.

Choosing the right approach

Choose the method based on your workload’s security requirements, performance needs, and operational constraints:

Method 1 – Use when you need to apply multiple KMS keys within a single Spark job and can accept some performance impact
Method 2 – Use for production workloads that require different encryption keys within the same bucket and need optimal performance
Method 3 – Use when a single KMS key per bucket meets your encryption requirements and you want simplified operations

Conclusion

In this post, we demonstrated how to handle multiple KMS keys when writing to Amazon S3 from Spark jobs on Amazon EMR. When encrypting multiple outputs with different encryption keys in a single Spark application, it’s important to consider the file system caching behavior. We presented several practical solutions with their respective trade-offs. You can start implementing these solutions in your environment by first testing the file system cache-disable method, which provides a straightforward approach to handling multiple encryption keys. As your workload grows, consider evolving to separate Spark sessions or S3 bucket default encryption based on your specific requirements. After implementing a solution, verify that each S3 object’s SSE-KMS key is the intended one (for example, by checking S3 object metadata). We also recommend measuring job performance and S3 API usage, especially for the cache-disable approach.

About the authors

Top 10 best practices for Amazon EMR Serverless

2026-01-26 Karthik Prabhakar

Post Syndicated from Karthik Prabhakar original https://aws.amazon.com/blogs/big-data/top-10-best-practices-for-amazon-emr-serverless/

Amazon EMR Serverless is a deployment option for Amazon EMR that you can use to run open source big data analytics frameworks such as Apache Spark and Apache Hive without having to configure, manage, or scale clusters and servers. EMR Serverless integrates with Amazon Web Services (AWS) services across data storage, streaming, orchestration, monitoring, and governance to provide a comprehensive serverless analytics solution.

In this post, we share the top 10 best practices for optimizing your EMR Serverless workloads for performance, cost, and scalability. Whether you’re getting started with EMR Serverless or looking to fine-tune existing production workloads, these recommendations will help you build efficient, cost-effective data processing pipelines. The following diagram illustrates an end-to-end EMR Serverless architecture, showing how it integrates into your analytics pipelines.

1. Define applications one time, reuse multiple times

EMR Serverless applications function as cluster templates that instantiate when jobs are submitted and can process multiple jobs without being recreated. This design significantly reduces startup latency for recurring workloads and simplifies operational management.

Typical workflow for EMR on EC2 transient cluster:

Typical workflow for EMR Serverless:

Applications feature a self-managing lifecycle that provisions resources to be available when needed without manual intervention. They automatically provision capacity when a job is submitted. For applications without pre-initialized capacity, resources are released immediately after job completion. For applications with pre-initialized capacity configured, those pre-initialized workers will stop after exceeding the configured idle timeout (15 minutes by default). You can adjust this timeout at the application level using AutoStopConfig configuration in the CreateApplication or UpdateApplication API. For example, if your jobs run every 30 minutes, increasing the idle timeout can eliminate startup delays between executions.

Most workloads are suited for on-demand capacity provisioning, which automatically scales resources based on your job requirements without incurring charges when idle. This approach is cost-effective and suitable for typical use cases including extract, transform, and load (ETL) workloads, batch processing jobs, and scenarios requiring maximum job resiliency.

For specific workloads with strict instant-start requirements, you can optionally configure pre-initialized capacity. Pre-initialized capacity creates a warm pool of drivers and executors that are ready to run jobs within seconds. However, this performance advantage comes with a tradeoff of added cost because pre-initialized workers incur continuous charges even when idle until the application reaches the Stopped state. Additionally, pre-initialized capacity restricts jobs to a single Availability Zone, which reduces resiliency.

Pre-initialized capacity should only be considered for:

Time-sensitive jobs with sub second service level agreement (SLA) requirements where startup latency is unacceptable
Interactive analytics where user experience depends on instant response
High-frequency production pipelines running every few minutes

In most other cases, on-demand capacity provides the best balance of cost, performance, and resiliency.

Beyond optimizing your applications’ use of resources, consider how you organize them across your workloads. For production workloads, use separate applications for different business domains or data sensitivity levels. This isolation improves governance and prevents resource contention between critical and noncritical jobs.

2. Choose AWS Graviton Processors for better price performance

Selecting the right underlying processor architecture can significantly impact both performance and cost. Graviton ARM-based processors offer significant performance improvement compared to x86_64.

EMR Serverless automatically updates to the latest instance generations as they become available, which means your applications benefit from the newest hardware improvements without requiring additional configuration.

To use Graviton with EMR Serverless, specify ARM64 with the architecture parameter during application creation using the CreateApplication or with the UpdateApplication API for existing applications:

aws emr-serverless create-application \
  --name my-spark-app \
  -- SPARK \
  --architecture ARM64 \
  --release-label emr-7.12.0

Considerations when using Graviton:

Resource availability – For large-scale workloads, consider engaging with your AWS account team to discuss capacity planning for Graviton workers.
Compatibility – Although many commonly used and standard libraries are compatible with Graviton (arm64) architecture, you will need to validate that third-party packages and libraries used are compatible.
Migration planning – Take a strategic approach to Graviton adoption. Build new applications on ARM64 architecture by default and migrate existing workloads through a phased transition plan that minimizes disruption. This structured approach will help optimize cost and performance without compromising reliability.
Perform benchmarks – It’s important to note that exact price performance will vary by workload. We recommend performing your own benchmarks to gauge specific results for your workload. For more details, refer to Achieve up to 27% better price-performance for Spark workloads with AWS Graviton2 on Amazon EMR Serverless.

3. Use defaults, right-size workers if needed

Workers are used to execute the tasks for your workload. While EMR Serverless defaults are optimized out of the box for a majority of use cases, you may need to right-size your workers to improve processing time and optimize cost efficiency. When submitting EMR Serverless jobs, it’s recommended to define Spark properties to configure workers, including memory size (in GB) and number of cores.

EMR Serverless configures the default worker size of 4 vCPUs, 16 GB memory, and 20 GB disk. Although this generally provides a balanced configuration for most jobs, you might want to adjust the size based on your performance requirements. Even when configuring pre-initialized workers with specific sizing, always set your Spark properties at job submission. This allows your job to use the specified worker sizing rather than default properties when it scales beyond pre-initialized capacity. When right-sizing your Spark workload, it’s important to identify the vCPU:memory ratio for your job. This ratio determines how much memory you allocate per virtual CPU core in your executors. Spark executors need both CPU and memory to process data effectively, and the optimal ratio varies based on your workload characteristics.

To get started, use the following guidance, then refine your configuration based on your specific workload requirements.

Executor configuration

The following table provides recommended executor configurations based on common workload patterns:

Workload type	Ratio	CPU	Memory	Configuration
Compute intensive	1:2	16 vCPU	32 GB	spark.emr-serverless.executor.cores=16spark.emr-serverless.executor.memory=32G
General purpose	1:4	16 vCPU	64 GB	spark.emr-serverless.executor.cores=16spark.emr-serverless.executor.memory=64G
Memory intensive	1:8	16 vCPU	108 GB	spark.emr-serverless.executor.cores=16spark.emr-serverless.executor.memory=108G

Driver configuration

The following table provides recommended driver configurations based on common workload patterns:

Workload type	Ratio	CPU	Memory	Configuration
General purpose	1:4	4 vCPU	16 GB	spark.emr-serverless.driver.cores=4spark.emr-serverless.driver.memory=16G
Apache Iceberg workloads	1:8(Large driver for metadata lookups)	8 vCPU	60 GB	spark.emr-serverless.driver.cores=8spark.emr-serverless.driver.memory=60G

To further monitor and tune your configuration, monitor your workload’s resource consumption using Amazon CloudWatch job worker-level metrics to identify constraints. Track CPU utilization, memory usage, and disk utilization metrics, then use the following table to fine-tune your configuration based on observed bottlenecks.

	Metrics observed	Workload type	Suggested action
1	High memory (>90%), Low CPU (<50%)	Memory-bound workload	Increase vCPU:memory ratio
2	High CPU (>85%), low memory (<60%)	CPU-bound workload	Increase vCPU count, maintain 1:4 ratio (For example, if using 8 vCPU, use 32 GB memory)
3	High storage I/O, normal CPU or memory with long shuffle operations	Shuffle-intensive	Enable serverless storage or shuffle-optimized disks
4	Low utilization across metrics	Over-provisioned	Reduce worker size or count
5	Consistent high utilization (>90%)	Under-provisioned	Scale up worker specifications
6	Frequent GC pauses**	Memory pressure	Increase memory overhead (10 –15%)

**You can identify frequent garbage collect (GC) pauses using the Spark UI under the Executors tab. There will be a GC time column that should generally be less than 10% of task time. Alternatively, the driver logs might frequently contain GC (Allocation Failure)] messages.

4. Control scaling boundary with T-shirt sizing

By default, EMR Serverless uses dynamic resource allocation (DRA), which automatically scales resources based on workload demand. EMR Serverless continuously evaluates metrics from the job to optimize for cost and speed, removing the need for you to estimate the exact number of workers required.

For cost optimization and predictable performance, you can configure an upper scaling boundary using one of the following approaches:

Setting the spark.dynamicAllocation.maxExecutors parameter at the job level
Setting the application-level maximum capacity

Rather than trying to fine-tune spark.dynamicAllocation.maxExecutors to an arbitrary value for each job, you can think about setting this configuration as t-shirt sizes that represent different workload profiles:

Workload size	Use cases	spark.dynamicAllocation.maxExecutors
Small	Exploratory queries, development	50
Medium	Regular ETL jobs, reports	200
Large	Complex transformations, large-scale processing	500

This t-shirt sizing approach simplifies capacity planning and helps you balance performance with cost efficiency based on your workload category, rather than attempting to optimize each individual job.

For EMR Serverless releases 6.10 and above, the default value for spark.dynamicAllocation.maxExecutors is infinity, but for earlier releases, it’s 100.

EMR Serverless automatically scales workers up or down based on the workload and parallelism required at every stage of the job. This automatic scaling is continuously evaluating metrics from the job to optimize for cost and speed, which removes the need for you to estimate the number of workers that the application needs to run your workloads.

However, in some cases, if you have a predictable workload, you might want to statically set the number of executors. To do so, you can disable DRA and specify the number of executors manually:

spark.dynamicAllocation.=false
spark.executor.instances=10

5. Provision appropriate storage for EMR Serverless jobs

Understanding your storage options and sizing them appropriately can prevent job failures and optimize execution times. EMR Serverless offers multiple storage options to handle intermediate data during job execution. The storage option selected will depend on the EMR release and use case. The storage options available in EMR Serverless are:

Storage type	EMR release	Disk size range	Use case	Benefits
Serverless Storage (recommended)	7.12+	N/A (auto-scaling)	Most Spark workloads, especially data-intensive workloads	No storage costs auto-scaling Reduces disk failures Up to 20% cost reduction
Standard Disks	7.11 and lower	20–200 GB per worker	Small to medium workloads processing datasets under 10 TB	Simple configuration 20 GB default suitable for most workloads, 200 GB max for optimal throughput
Shuffle-Optimized Disks	7.1.0+	20–2,000 GB per worker	Large-scale ETL workloads processing multi-TB	High IOPS and throughput Up to 2 TB capacity per worker

By matching your storage configuration to your workload characteristics, you’ll enable EMR Serverless jobs to run efficiently and reliably at scale.

6. Multi-AZ out-of-the-box with built-in resiliency

EMR Serverless applications are multi-AZ from the start when pre-initialized capacity isn’t enabled. This built-in failover capability provides resilience against Availability Zone disruptions without manual intervention. A single job will operate within a single Availability Zone to prevent cross-AZ data transfer costs and subsequent jobs will be intelligently distributed across multiple AZs. If EMR Serverless determines that an AZ is impaired, it will submit new jobs to a healthy AZ, enabling your workloads to continue running despite AZ impairment.

To fully benefit from EMR Serverless multi-AZ functionality verify the following:

Configure a network connection to your VPC with multiple subnets across Availability Zones selected
Avoid pre-initialized capacity which restricts applications to a single AZ
Make sure there are sufficient IP addresses available in each subnet to support the scaling of workers

In addition to multi-AZ, with Amazon EMR 7.1 and higher, you can enable job resiliency, which allows your jobs to be automatically retried in case errors are encountered. If there are multiple Availability Zones configured, it will also be retried in a different AZ. You can enable this feature for both batch and streaming jobs, though retry behavior differs between the two.

Configure job resiliency by specifying a retry policy that defines the maximum number of retry attempts. For batch jobs, the default is no automatic retries (maxAttempts=1). For streaming jobs, EMR Serverless retries indefinitely with built-in thrash prevention that stops retries after five failed attempts within 1 hour. You can configure this threshold between 1–10 attempts. For more information, refer to Job resiliency.

In the event that you need to cancel your job, you can specify a grace period to allow your jobs to shut down cleanly rather than the default behavior of immediate termination. This can also include custom shutdown hooks if you need to perform custom cleanup actions.

By combining multi-AZ support, automatic job retries, and graceful shutdown periods, you create a robust foundation for EMR Serverless workloads that can tolerate interruptions and maintain data integrity without manual intervention.

7. Secure and extend connectivity with VPC integration

By default, EMR Serverless can access AWS services such as Amazon Simple Storage Service (Amazon S3), AWS Glue, Amazon CloudWatch Logs, AWS Key Management Service (AWS KMS), AWS Security Token Service (AWS STS), Amazon DynamoDB, and AWS Secrets Manager. If you want to connect to data stores within your VPC, such as Amazon Redshift or Amazon Relational Database Service (Amazon RDS), you must configure VPC access for the EMR Serverless application.

When configuring VPC access for your EMR Serverless application, keep these key considerations in mind to gain optimal performance and cost efficiency:

Plan for sufficient IP addresses – Each worker uses one IP address within a subnet. This includes the workers that will be launched when your job is scaling out. If there aren’t enough IP addresses, your job might not be able to scale, which could result in job failure. Verify you have adhered to best practices for subnet planning for optimal performance.
Set up Gateway endpoints for Amazon S3 for applications in a private subnets – Running EMR Serverless in a private subnet without VPC endpoints for Amazon S3 will route your Amazon S3 traffic through NAT gateways, resulting in additional data transfer charges. VPC endpoints for S3 will keep this traffic within your VPC, reducing costs and improving performance for Amazon S3 operations.
Manage AWS Config costs for network interfaces – EMR Serverless generates an elastic network interface record in AWS Config for each worker, which can accumulate costs as your workloads scale. If you don’t require AWS Config tracking for EMR Serverless network interfaces, consider using resource-based exclusions or tagging strategies to filter them out while maintaining AWS Config coverage for other resources.

For more details, refer Configuring VPC access for EMR Serverless applications.

8. Simplify job submission and dependency management

EMR Serverless supports flexible job submission through the StartJobRun API, which accepts the full spark-submit syntax. For runtime environment configuration, use the spark.emr-serverless.driverEnv and spark.executorEnv prefixes to set environment variables for driver and executor processes. This is particularly useful for passing sensitive configuration or runtime-specific settings.

For Python applications, package dependencies using virtual environments by creating a venv, packaging it as a tar.gz archive, or uploading to Amazon S3 using spark.archives with the appropriate PYSPARK_PYTHON environment variable. This allows Python dependencies to be available across driver and executor workers.

For improved control under high load, enable job concurrency and queuing (available in EMR 7.0.0+) to limit the number of jobs that can be executed concurrently. With this feature, jobs submitted that exceed the concurrency limit are queued until resources become available.

You can configure Job concurrency and queue settings using the SchedulerConfiguration property using the CreateApplication or UpdateApplication API.

--scheduler-configuration '{"maxConcurrentRuns": 5, "queueTimeoutMinutes": 30}'

9. Use EMR Serverless configurations to enforce limits

EMR Serverless automatically scales resources based on workload demand, providing optimized defaults that work well for most use cases without requiring Spark configuration tuning. To manage costs effectively, you can configure resource limits that align with your budget and performance requirements. For advanced use cases, EMR Serverless also provides configuration options so you can fine-tune resource consumption and achieve the same efficiency as cluster-based deployments. Understanding these limits helps you balance performance with cost efficiency for your jobs.

Limit type	Purpose	How to configure
Job-level	Control resources for individual jobs	spark.dynamicAllocation.maxExecutors or spark.executor.instances
Application-level	Limit resources per application or business domain	Set maximum capacity when creating the application or while updating.
Account-level	Prevent abnormal resource spikes across all applications	Auto-adjustable service quota Max concurrent vCPUs per account; request increases via Service Quotas console

These three layers of limits work together to provide flexible resource management at different scopes. For most use cases, configuring job-level limits using the t-shirt sizing approach is sufficient, while application and account-level limits provide additional guardrails for cost control.

10. Monitor with CloudWatch, Prometheus, and Grafana

Monitoring EMR Serverless workloads simplifies the process of debugging, performing cost optimization, and performance tracking. EMR Serverless offers three tiers of monitoring that work together: Amazon CloudWatch, Amazon Managed Service for Prometheus, and Amazon Managed Grafana.

Amazon CloudWatch – CloudWatch integration is enabled by default and publishes metrics to the AWS/EMRServerless namespace. EMR Serverless sends metrics to CloudWatch every minute at the application level, as well as job, worker-type, and capacity-allocation-type levels. Using CloudWatch, you can configure dashboards for enhanced observability into workloads or configure alarms to alert for job failures, scaling anomalies, and SLA breaches. Using CloudWatch with EMR Serverless provides insights to your workloads so you can catch issues before they impact users.
Amazon Managed Service for Prometheus – With EMR Serverless release 7.1+, you can enable Prometheus for detailed Spark engine metrics to push metrics to Amazon Managed Service for Prometheus. This unlocks executor-level visibility, including memory usage, shuffle volumes, and GC pressure. You can use this to identify memory-constrained executors, detect shuffle-heavy stages, and find data skew.
Amazon Managed Grafana – Grafana connects to both CloudWatch and Prometheus data sources, providing a single pane of glass for unified observability and correlation analysis. This layered approach helps you correlate infrastructure issues with application-level performance problems.

Key metrics to track:

Job completion times and success rates
Worker utilization and scaling events
Shuffle read/write volumes
Memory usage patterns

For more details, refer to Monitor Amazon EMR Serverless workers in near real time using Amazon CloudWatch.

Conclusion

In this post, we shared 10 best practices to help you maximize the value of Amazon EMR Serverless by optimizing performance, controlling costs, and maintaining reliable operations at scale. By focusing on application design, right-sized workloads, and architectural choices, you can build data processing pipelines that are both efficient and resilient.

To learn more, refer to the Getting started with EMR Serverless guide.

About the Authors

Apache Spark 4.0.1 preview now available on Amazon EMR Serverless

2026-01-26 Al MS

Post Syndicated from Al MS original https://aws.amazon.com/blogs/big-data/apache-spark-4-0-1-preview-now-available-on-amazon-emr-serverless/

Amazon EMR Serverless now supports Apache Spark 4.0.1 in preview, making analytics accessible to more users, simplifying data engineering workflows, and strengthening governance capabilities. The release introduces ANSI SQL compliance, VARIANT data types support for JSON handling, Apache Iceberg v3 table format support, and enhanced streaming capabilities. This preview is available in all regions where EMR Serverless is available.

In this post, we explore key benefits, technical capabilities, and considerations for getting started with Spark 4.0.1 on Amazon EMR Serverless—a serverless deployment option that simplifies running open-source big data frameworks, without requiring managing clusters. With the emr-spark-8.0-preview release label, you can evaluate new SQL capabilities, Python API improvements, and streaming enhancements in your existing EMR Serverless environment.

Benefits

Spark 4.0.1 helps you solve data engineering problems with specific improvements. This section shows how new capabilities help with real-world scenarios.

Make analytics accessible to more users

Simplify Extract Transform Load (ETL) development with SQL scripting. Data engineers often switch between SQL and Python to build complex ETL logic with control flow. SQL scripting in Spark 4.0.1 enables loops, conditionals, and session variables directly in SQL, reducing context-switching and simplifying pipeline development. Use pipe syntax (|>) to chain operations for more readable, maintainable queries.

Improve data quality with ANSI SQL mode. Silent type conversion failures can introduce data quality issues. ANSI SQL mode (now default) enforces standard SQL behavior, raising errors for invalid operations instead of producing unexpected results. Important: ANSI SQL mode is now enabled by default. Test your queries thoroughly during this preview evaluation.

Simplify data engineering workflows

Process JSON data efficiently with VARIANT. Teams working with semi-structured data often see slow performance from repeated JSON parsing. The VARIANT data type stores JSON in an optimized binary format, eliminating parsing overhead. You can efficiently store and query JSON data in data lakes without schema rigidity.

Build Python data sources without Scala. Integrating custom data sources previously required Scala expertise. The Python data Source API lets you build connectors entirely in Python, using existing Python skills and libraries without learning a new language.

Debug streaming applications with queryable state. Troubleshooting stateful streaming applications has historically required indirect methods. The new state data source reader shows streaming state as queryable DataFrames. You can inspect state during debugging, test state values in unit tests, and diagnose production incidents.

Strengthen governance capabilities

Establish comprehensive audit trails with Apache Iceberg v3. The Apache Iceberg v3 table format provides transaction guarantees and tracks data changes over time, giving you the audit trails needed for regulatory compliance. When combined with VARIANT data type support, you can maintain governance controls while handling semi-structured data efficiently in data lakes.

Key capabilities

Spark 4.0.1 Preview on EMR Serverless introduces four major capability areas:

SQL enhancements – ANSI mode, pipe syntax, VARIANT type, SQL scripting, user-defined functions (UDFs)
Python API advances – custom data sources, UDF profiling
Streaming improvements – stateful processing API v2, queryable state
Table format support – Amazon S3 Tables, AWS Lake Formation integration

The following sections provide technical details and code examples for each capability.

SQL enhancements

Spark 4.0.1 introduces new SQL capabilities including ANSI mode compliance, SQL UDFs, pipe syntax for readable queries, VARIANT type for JSON handling, and SQL scripting with control flow.

ANSI SQL mode by default

ANSI SQL mode is now enabled by default, enforcing standard SQL behavior for data integrity. Silent casting of out-of-range values now raises errors rather than producing unexpected results. Existing queries may behave differently, particularly around null handling, string casting, and timestamp operations. Use spark.sql.ansi.enabled=false if you need legacy behavior during migration.

SQL pipe syntax

You can now chain SQL operations using the |> operator for improved readability. The following example shows how you can replace nested subqueries with a more maintainable pipeline:

FROM customer
|> LEFT OUTER JOIN orders ON c_custkey = o_custkey
|> AGGREGATE COUNT(o_orderkey) c_count GROUP BY c_custkey
|> AGGREGATE COUNT(*) AS custdist GROUP BY c_count
|> ORDER BY custdist DESC

This replaces nested subqueries, making complex transformations easier to understand and maintain.

VARIANT data type

The VARIANT type handles semi-structured JSON/XML data efficiently without repeated parsing. It uses an optimized binary representation internally while maintaining schema-less flexibility. Previously, JSON expressions required repeated parsing, degrading performance. VARIANT eliminates this overhead. The following snippet shows how to parse JSON into the VARIANT type:

df = spark.sql("SELECT parse_json('{\"name\":\"Alice\",\"age\":30}') as data")

Spark 4.0.1 on EMR Serverless supports Apache Iceberg v3, enabling the VARIANT data type with Iceberg tables. This combination provides efficient storage and querying of semi-structured JSON data in your data lake. Store VARIANT columns in Iceberg tables and use Iceberg’s schema evolution and time travel capabilities alongside Spark’s optimized JSON processing. The following example shows how to create an Iceberg table with a VARIANT column:

CREATE TABLE catalog.db.events (
  event_id BIGINT,
  event_data VARIANT,
  timestamp TIMESTAMP
) USING iceberg;

INSERT INTO catalog.db.events SELECT 1, parse_json('{"user":"alice","action":"login"}'), current_timestamp();

SQL scripting with session variables

Manage state and control flow directly in SQL using session variables, and IF/WHILE/FOR statements. The following example demonstrates a loop that populates a results table:

BEGIN
  DECLARE counter INT = 10;
  WHILE counter > 0 DO
    INSERT INTO results VALUES (counter);
    SET counter = counter - 1;
  END WHILE;
END

This enables complex ETL logic entirely in SQL without switching to Python.

SQL user-defined functions

Define custom functions directly in SQL. Functions can be temporary (session-scoped) or permanent (catalog-stored). The following example shows how to register and use a simple UDF:

CREATE FUNCTION plusOne(x INT) RETURNS INT RETURN x + 1;
SELECT plusOne(5);

Python API advances

This section covers new Python capabilities including custom data sources and UDF profiling tools.

Python data source API

You can now build custom data sources in Python without Scala knowledge. The following example shows how to create a simple data source that returns sample data:

from pyspark.sql.datasource import DataSource, DataSourceReader
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

class SampleDataSource(DataSource):
    def schema(self):
        return StructType([
            StructField("name", StringType()),
            StructField("age", IntegerType())
        ])
    
    def reader(self, schema):
        return SampleReader()

class SampleReader(DataSourceReader):
    def read(self, partition):
        yield ("Alice", 30)
        yield ("Bob", 25)

# Register and use
spark.dataSource.register(SampleDataSource)
spark.read.format("SampleDataSource").load().show()

Unified UDF profiling

Profile Python and Pandas UDFs for performance and memory insights. The following code enables performance profiling:spark.conf.set("spark.sql.pyspark.udf.profiler", "perf") # or "memory"

Structured streaming enhancements

This section covers improvements to stateful stream processing, including queryable state and enhanced state management APIs.

Arbitrary stateful processing API v2

The transformWithState operator provides robust state management with timer and TTL support for automatic cleanup, schema evolution capabilities, and initial state support for pre-populating state from batch DataFrames.

State data source reader

Query streaming state as a DataFrame for debugging and monitoring. Previously, state data was internal to streaming queries. Now you can verify state values in unit tests, diagnose production incidents, detect state corruption, and optimize performance. Note: This feature is experimental. Source options and behavior may change in future releases.

State store improvements

Upgraded changelog checkpointing for RocksDB removes performance bottlenecks. Enhanced checkpoint coordination and improved sorted string table (SST) file reuse management optimizes streaming operations.

Table format support

This section covers support for AWS S3 Tables and full table access (FTA) with AWS Lake Formation.

AWS S3 Tables

Use Spark 4.0.1 with AWS S3 Tables, a storage solution that provides managed Apache Iceberg tables with automatic optimization and maintenance. S3 Tables simplify data lake operations by handling compaction, snapshot management, and metadata cleanup automatically.

Full table access with Lake Formation

FTA is supported for Apache Iceberg, Delta Lake, and Apache Hive tables when using AWS Lake Formation, a managed service that simplifies data access control. FTA provides coarse-grained access control at the table level. Note that fine-grained access control (FGAC) with column-level or row-level permissions is not available in this preview.

Getting started

Follow these steps to create an EMR Serverless application, run sample code to test new features, and provide feedback on the preview.

Prerequisites

Before you begin, confirm you have the following:

AWS Command Line Interface (CLI): Install and configure the AWS CLI with credentials that have permissions to create and manage EMR Serverless applications
EMR Serverless access: Your Identity and Access Management (IAM) role must have permissions for EMR Serverless operations (emr-serverless:CreateApplication,emr-serverless:StartJobRun). See the EMR Serverless IAM policy examples for minimum required permissions
S3 bucket: An S3 bucket to store your job scripts and data
Runtime role: An IAM execution role for EMR Serverless with access to your S3 resources

Note: EMR Studio Notebooks and SageMaker Unified Studio are not supported during this preview. Use the AWS CLI or AWS SDK to submit jobs.

Step 1: Create your EMR Serverless application

Create or update your application with the emr-spark-8.0-preview release label. The following command creates a new application:

aws emr-serverless create-application --type spark \
  --release-label emr-spark-8.0-preview \
  --region us-east-1 --name spark4-test

Step 2: Test sample code

Run this PySpark job to verify setup and test Spark 4.0.1 features:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Spark 4.0.1 Test").getOrCreate()
print(f"Spark Version: {spark.version}")

# Create sample data
data = [("Alice", 34, "Engineering"), ("Bob", 45, "Sales"),
        ("Charlie", 28, "Engineering"), ("Diana", 52, "Marketing")]
df = spark.createDataFrame(data, ["name", "age", "department"])
df.createOrReplaceTempView("employees")

# Test SQL PIPE syntax
try:
    result = spark.sql("""
        FROM employees
        |> WHERE age > 30
        |> SELECT name, age, department
        |> ORDER BY age DESC
    """)
    result.show()
    print("✓ SQL pipe syntax test passed")
except Exception as e:
    print(f"✗ SQL pipe syntax test failed: {e}")

# Test VARIANT data type
try:
    json_data = spark.sql("""
        SELECT parse_json('{"name":"Alice","skills":["Python","Spark","SQL"]}') as data
    """)
    json_data.show(truncate=False)
    print("✓ VARIANT data type test passed")
except Exception as e:
    print(f"✗ VARIANT data type test failed: {e}")

Submit the job with the following command:

aws emr-serverless start-job-run \
    --application-id <your-application-id> \
    --execution-role-arn <your-execution-role-arn> \
    --job-driver '{
        "sparkSubmit": {
            "entryPoint": "s3://<your-bucket>/spark_4_test.py",
            "sparkSubmitParameters": "--conf spark.executor.cores=4 --conf spark.executor.memory=16g"
        }
    }'

Step 3: Test your workloads

Review the Spark SQL Migration Guide and PySpark Migration Guide, then test production workloads in non-production environments. Focus on queries affected by ANSI SQL mode and benchmark performance.

Step 4: Clean up resources

After testing, delete all resources created during this evaluation to avoid ongoing charges:

# Delete the EMR Serverless application
aws emr-serverless delete-application \
    --application-id spark4-test \
    --region us-east-1
# Remove the test script from S3
aws s3 rm s3://<your-bucket>/spark_4_test.py

Migration considerations

Before evaluating Spark 4.0.1, review the updated runtime requirements and behavioral changes that may affect your existing code.

Runtime requirements

Scala: Version 2.13.16 required (2.12 support dropped)
Java: JDK 17 or higher required (JDK 8 and 11 support removed)
Python: Version 3.9+ required, continued support for 3.11 and newly added 3.12 (3.8 support removed)
Pandas: Minimum version 2.0.0 (previously 1.0.5)
SparkR: Deprecated; migrate to PySpark

Behavioral changes

With ANSI SQL mode enforcement, you may see different behavior in:

Null handling: Stricter null propagation in expressions
String casting: Invalid casts now raise errors instead of returning null
Map key operations: Duplicate keys now raise errors
Timestamp conversions: Overflow returns null instead of wrapped values
CREATE TABLE statements: Now respect the spark.sql.sources.defaultconfiguration instead of defaulting to Hive format when USING or STORED AS clauses are omitted

You can control many of these behaviors via legacy configuration flags. Consult the official migration guides for details refer: Spark SQL Migration Guide: 3.5 to 4.0 and PySpark Migration Guide: 3.5 to 4.0.

Preview limitations

The following capabilities are not available in this preview:

Fine-grained access control: Fine-grained access control (FGAC) with row-level or column-level filtering is not supported in this preview. Jobs with spark.emr-serverless.lakeformation.enabled=true will fail.
Spark Connect: Not supported in this preview. Use standard Spark job submission with the StartJobRun API.
Open Table Format limitations: Hudi is not supported in this preview. Delta 4.0.0 does not support Flink connectors (deprecated in Delta 4.0.0). Delta Universal Format is not supported in this preview.
Connectors: spark-sql-kinesis, emr-dynamodb, and spark-redshift are unavailable.
Interactive applications: Livy and JupyterEnterpriseGateway are not included. Also, SageMaker Unified Studio and EMR Studio are not supported.
EMR features: Serverless Storage and Materialized Views are not supported.

This preview lets you evaluate Spark 4.0.1’s core capabilities on EMR Serverless, including SQL enhancements, Python API improvements, and streaming state management. Test your migration path, assess performance improvements, and provide feedback to shape the general availability release.

Conclusion

This post showed you how to get started with the Apache Spark 4.0.1 preview release on Amazon EMR Serverless. You explored how the VARIANT data type works with Iceberg v3 to process JSON data efficiently, how SQL scripting and pipe syntax eliminate context-switching for ETL development, and how queryable streaming state simplifies debugging stateful applications. You also learned about the preview limitations, runtime requirements, and behavioral changes to consider during evaluation.

Test the Spark 4.0.1 preview on EMR Serverless and provide feedback through AWS Support to help shape the general availability release.

To learn more about Apache Spark 4.0.1 features, see the Spark 4.0.1 Release Notes. For EMR Serverless documentation, see the EMR Release Guide.

Resources

Apache Spark Documentation

Amazon EMR Resources

Amazon EMR Spark Features

About the authors

Enterprise scale in-place migration to Apache Iceberg: Implementation guide

2026-01-20 Mihir Borkar

Post Syndicated from Mihir Borkar original https://aws.amazon.com/blogs/big-data/enterprise-scale-in-place-migration-to-apache-iceberg-implementation-guide/

Organizations managing large-scale analytical workloads increasingly face challenges with traditional Apache Parquet-based data lakes with Hive-style partitioning, including slow queries, complex file management, and limited consistency guarantees. Apache Iceberg addresses these pain points by providing ACID transactions, seamless schema evolution, and point-in-time data recovery capabilities that transform how enterprises handle their data infrastructure.

In this post, we demonstrate how you can achieve migration at scale from existing Parquet tables to Apache Iceberg tables. Using Amazon DynamoDB as a central orchestration mechanism, we show how you can implement in-place migrations that are highly configurable, repeatable, and fault-tolerant—unlocking the full potential of modern data lake architectures without extensive data movement or duplication.

Solution overview

When performing in-place migration, Apache Iceberg uses its ability to directly reference existing data files. This capability is only supported for formats such as Parquet, ORC, and Avro, because these formats are self-describing and include consistent schema and metadata information. Unlike raw formats such as CSV or JSON, they enforce structure and support efficient columnar or row-based access, which allows Iceberg to integrate them without rewriting the data.

In this post, we demonstrate how you can migrate an existing Parquet-based data lake that isn’t cataloged in AWS Glue by using two methodologies:

Apache Iceberg migrate and register_table approach. Ideal for converting existing Hive-registered Parquet tables into Iceberg-managed tables.
Iceberg add_files approach. Best suited for quickly onboarding raw Parquet data into Iceberg without rewriting files.

The solution also incorporates a DynamoDB table that acts as a scalable control plane, so you can perform in-place migration of your data lake from Parquet format to Iceberg format.

The following diagram shows different methodologies that you can use to achieve this in-place migration of your Hive-style partitioned data lake:

AWS data pipeline architecture diagram showing data flow from Amazon DynamoDB through Amazon EMR and AWS Glue to a Data Lake and Apache Iceberg Lakehouse, both using Parquet format, within an AWS Region.

You use DynamoDB to track the migration state, handling retries and recording errors and outcomes. This provides the following benefits:

Centralized control over which Amazon Simple Storage Service (Amazon S3) paths need migration.
Lifecycle tracking of each dataset through migration stages.
Capture and audit errors on a per-path basis.
Enable re-runs by updating stateful flags or clearing failure messages.

Prerequisites

Before you begin, you need:

An AWS account
AWS Command Line Interface (AWS CLI) installed
AWS Identity and Access Management (IAM) permissions to access Amazon DynamoDB, Amazon EMR, and AWS Glue
An existing or new Amazon Virtual Private Cloud (Amazon VPC) to Amazon EMR clusters
Amazon Athena access with a workgroup configured and the query results location (Amazon S3) set
An Amazon EMR cluster using Hive as the metastore, with SSH access. (See Appendix A for setup instructions.)
An Amazon EMR cluster or Amazon EMR Serverless environment using AWS Glue Data Catalog as the Spark metastore, with SSH access. (See Appendix B for setup instructions.)
For AWS Glue exchange, transform, and load (ETL), use Glue 4.0 or later.

Create sample Parquet dataset as a source

You can create the sample Parquet dataset for testing the different methodologies using the Athena query editor. Replace <amzn-s3-demo-bucket> with an available bucket in your account.

Create an AWS Glue database(test_db), if not present.
```
CREATE DATABASE IF NOT EXISTS test_db
```

Create a sample Parquet table (table1) and add to be used for testing the add_files approach.

CREATE TABLE table1
WITH (
  external_location = 's3://<amzn-s3-demo-bucket>/table1/',
  format = 'PARQUET',
  partitioned_by = ARRAY['date', 'hour']
)
AS
SELECT 
  1 as id,
  'John Doe' as name,
  25 as age,
  'Engineer' as job_title,
  current_date as created_date,
  current_date as date,
  hour(current_timestamp) as hour
UNION ALL
SELECT 2, 'Jane Smith', 30, 'Manager', current_date, current_date, hour(current_timestamp)
UNION ALL  
SELECT 3, 'Bob Johnson', 35, 'Analyst', current_date, current_date, hour(current_timestamp);

Create a sample Parquet table (table2) and add data to be used for testing the migrate and register_table approach. Replace <amzn-s3-demo-bucket> with your bucket name.

CREATE TABLE table2
WITH (
  external_location = 's3://<amzn-s3-demo-bucket>/table2/',
  format = 'PARQUET',
  partitioned_by = ARRAY['date', 'hour']
)
AS
SELECT 
  1 as id,
  'John Doe' as name,
  25 as age,
  'Engineer' as job_title,
  current_date as created_date,
  current_date as date,
  hour(current_timestamp) as hour
UNION ALL
SELECT 2, 'Jane Smith', 30, 'Manager', current_date, current_date, hour(current_timestamp)
UNION ALL  
SELECT 3, 'Bob Johnson', 35, 'Analyst', current_date, current_date, hour(current_timestamp);

Drop the tables from the Data Catalog because you only need Parquet data with the Hive-style partitioning structure.
```
DROP TABLE IF EXISTS test_db.table1
```

Create a DynamoDB control table

Before beginning the migration process, you must create a DynamoDB table that serves as the control plane. This table maps source Amazon S3 paths to their corresponding Iceberg database and table destinations, enabling systematic tracking of the migration process.

To implement this control mechanism, create a table with the following structure:

A primary key s3_path that stores the source Parquet data location
Two attributes that define the target Iceberg location:
- target_db_name
- target_table_name

To create the DynamoDB control table

Create the Amazon DynamoDB table using the following AWS CLI command:

aws dynamodb create-table \
--table-name migration-control-table \
--attribute-definitions \
AttributeName=s3_path,AttributeType=S \
--key-schema \
AttributeName=s3_path,KeyType=HASH \
--billing-mode PAY_PER_REQUEST \
--region <REGION>

Verify the table is created successfully. Replace <REGION> with the AWS Region where your data is stored:
```
aws dynamodb describe-table --table-name migration-control-table --region <REGION>
```

Create a migration_data.json file with the following contents.
In this example:

Replace <amzn-s3-demo-bucket> and <TablePrefix>with the name of your S3 bucket and prefix containing the Parquet data
Replace <DatabaseName> with the name of your target Iceberg database
Replace <TableName> with the name of your target Iceberg table

{
    "your-migration-table": [
        {
            "PutRequest": {
                "Item": {
                    "s3_path": {"S": "s3://<amzn-s3-demo-bucket>/table1/"},
                    "target_db_name": {"S": "test_db"},
                    "target_table_name": {"S": "table1"}
                }
            }
        },
        {
            "PutRequest": {
                "Item": {
                    "s3_path": {"S": "s3://<amzn-s3-demo-bucket>/table2/"},
                    "target_db_name": {"S": "test_db"},
                    "target_table_name": {"S": "table2"}
                }
            }
        },
        {
            "PutRequest": {
                "Item": {
                    "s3_path": {"S": "s3://<amzn-s3-demo-bucket>/<TablePrefix>/"},
                    "target_db_name": {"S": "<DatabaseName>"},
                    "target_table_name": {"S": "<TableName>"}
                }
            }
        }
    ]
}

This file defines the mapping between Amazon S3 paths and their corresponding Iceberg table destinations.

Run the following CLI command to load the DynamoDB control table.

aws dynamodb batch-write-item \
--request-items file://migration_data.json \
--region <REGION;>

Migration methodologies

In this section, you explore two methodologies for migrating your existing Parquet tables to Apache Iceberg format:

Apache Iceberg migrate and register_table approach – This approach first converts your Parquet table to Iceberg format using the native migrate procedure, followed by registering it in AWS Glue using the register_table procedure.
Apache Iceberg add_files approach – This method creates an empty Iceberg table and uses the add_files procedure to import existing Parquet data files without physically moving them.

Apache Iceberg migrate and register_table procedure

Use the Apache Iceberg Migrate procedure that is used for in-place conversion of an existing Hive or Parquet table into an Iceberg-managed table. Thereafter, you can use the Apache Iceberg RegisterTable procedure to register the respective table in AWS Glue.

AWS workflow diagram showing DynamoDB to Apache Iceberg migration using Amazon EMR with Hive Metastore for migration and Glue Metastore for registration, displaying configuration tables at each stage.

Migrate

In your EMR cluster with Hive as the metastore, create a PySpark session with the following Iceberg Packages:

pyspark \
--name "Iceberg Migration" \
--conf "spark.jars=/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar" \
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
--conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
--conf spark.sql.catalog.spark_catalog.type=hive

This post uses Iceberg v1.9.1 (Amazon EMR build), which is native to Amazon EMR 7.11. Always verify the latest supported version and update package coordinates accordingly.

Next, create your corresponding table in your Hive catalog (you can skip this step if you already have tables created in your hive catalog). Replace <amzn-s3-demo-bucket> with the name of your S3 bucket.
In the following snippet, change or remove the PARTITIONED BY command based on the partition strategy of your table, the MSCK Repair table command should only be run if your respective table is partitioned.

#You can automate this for production Scaling with DynamoDB as control table 
s3_path = "s3://<amzn-s3-demo-bucket>/table1/"
target_db_name = "test_db"
target_table_name = "table1"
# Read data in a dataframe to infer schema
df = spark.read.parquet(s3_path)
df.createOrReplaceTempView("temp_view")
# Get schema as string
schema = spark.table("temp_view").schema
schema_string = ", ".join([f"{field.name} {field.dataType.simpleString()}" for field in schema])
# Create Database If not exists 
spark.sql(f"CREATE DATABASE IF NOT EXISTS {target_db_name}").show()
# full_table_name= test_db.table1
full_table_name = f"{target_db_name}.{target_table_name}"
# Create table
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {full_table_name} (
    {schema_string}
)
STORED AS PARQUET
PARTITIONED BY (date, hour)
LOCATION '{s3_path}'
""")
# Refresh, repair, and validate
spark.sql(f"REFRESH TABLE {full_table_name}")
spark.sql(f"MSCK REPAIR TABLE {full_table_name}")

Convert the Parquet table to an Iceberg table in Hive

# Run migration procedure
spark.sql(f"CALL spark_catalog.system.migrate('{full_table_name}')")
# Validate that the table is successfully migrated 
spark.sql(f"DESCRIBE FORMATTED {full_table_name}").show(truncate=False)

Run the migrate command to convert the Parquet-based table to an Iceberg table, creating the metadata folder and the metadata.json file therein

You can stop at this point if you don’t intend to migrate your existing iceberg table from Hive to the Data Catalog.

Register

Create the session with the respective Iceberg Packages. Replace <amzn-s3-demo-bucket> with your bucket name, and <warehouse> with warehouse directory.

pyspark \
--conf "spark.jars=/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar" \
--conf "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
--conf "spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog" \
--conf "spark.sql.catalog.glue_catalog.warehouse= s3://<amzn-s3-demo-bucket>/<warehouse>/"  \
--conf "spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog" \
--conf "spark.sql.catalog.glue_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO"

Run the register_table command to make the Iceberg table visible in AWS Glue.

register_table registers an existing Iceberg table’s metadata file (metadata.json) with a catalog(glue_catalog) so that Spark (and other engines) can query it.
The procedure creates a Data Catalog entry for the table, pointing it to the given metadata location.

Replace <amzn-s3-demo-bucket> and <metadata-prefix> with the name of your S3 bucket and metadata prefix name.

Ensure that your EMR Spark Cluster has been configured with appropriate AWS Glue permissions

# You can automate this for production Scaling with DynamoDB as control table
metadata_location = "s3://<amzn-s3-demo-bucket>/table1/metadata/<metadata-prefix>.metadata.json"
target_db_name = "test_db"
target_table_name = "table1"
full_table_name = f"{target_db_name}.{target_table_name}"
# Register existing Iceberg table metadata in Glue Catalog
spark.sql(f"CALL glue_catalog.system.register_table('{full_table_name}', '{metadata_location}')")
# Set table properties (example: Iceberg format version 2)
spark.sql(f"ALTER TABLE glue_catalog.{full_table_name} SET TBLPROPERTIES('format-version'='2')")

Validate that the Iceberg table is now visible in the Data Catalog.

# Lookout for format as iceberg/parquet
spark.sql("SHOW TBLPROPERTIES glue_catalog.test_db.table1").show()

Apache Iceberg’s add_files procedure

AWS workflow diagram showing DynamoDB to Apache Iceberg migration using AWS Glue Add_Files procedure, displaying input configuration and output status tables with metadata location and registration confirmation.

Here, you’re going to use Iceberg’s add_files procedure to import raw data files (Parquet, ORC, Avro) into an existing Iceberg table by updating its metadata. This procedure works for both Hive and Data Catalog, it doesn’t physically move or rewrite the files—it only registers them so Iceberg can manage them.

This methodology comprises the following steps:

Create an empty Iceberg table in AWS Glue.
Because the add_files procedure expects the iceberg table to be already present, you need to create an empty Iceberg table by inferring the table schema.
Register existing data locations to the Iceberg table

Using the add_files procedure in a Glue-backed Iceberg catalog will register the target S3 path along with all its subdirectories to the empty Iceberg table created in the previous step.

You can consolidate both steps into a single Spark job. For the following AWS Glue job, you have specified iceberg as a value for the --datalake-formats job parameter. See the AWS Glue job configuration documentation for more details.

Replace <amzn-s3-demo-bucket> with your S3 bucket name and <warehouse> with warehouse directory.

from pyspark.sql import SparkSession
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
target_db_name = "test_db"
target_table_name = "table2"
s3_path = "s3://<amzn-s3-demo-bucket>/table2"
# Set to None or [] for unpartitioned
partitioned_cols = ["date", "hour"]  
spark = SparkSession.builder \
    .appName("Iceberg Add Files") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.glue_catalog.warehouse", "s3://<amzn-s3-demo-bucket>/<warehouse>/") \
    .getOrCreate()
full_table_name = f"glue_catalog.{target_db_name}.{target_table_name}"
# Read schema from one file (schema inference)
df = spark.read.parquet(s3_path)
schema = df.schema
# Create empty Iceberg table
empty_df = spark.createDataFrame([], schema)
if partitioned_cols:
    empty_df.writeTo(full_table_name).using("iceberg").partitionedBy(*partitioned_cols). .tableProperty("format-version", "2").create()
else:
    empty_df.writeTo(full_table_name).using("iceberg").tableProperty("format-version", "2").create()
logger.info(f"Created empty Iceberg table: {full_table_name}")
spark.sql(f"""
CALL glue_catalog.system.add_files(
  '{target_db_name}.{target_table_name}',
  'parquet.`{s3_path}`'
)
""")

When working with non-Hive partitioned datasets, a direct migration to Apache Iceberg using add_files might not behave as expected. See Appendix C for more information.

Considerations

Let’s explore two key considerations that you should address when implementing your migration strategy.

State management using DynamoDB control table

Use the following sample code snippet to update the state of DynamoDB table:

def update_dynamodb_record(self, s3_path, metadata_loc=None, error_msg=None):
    # Get current error message
    try:
        response = self.dynamodb.get_item(
            TableName='migration-control-table',
            Key={'s3_path': {'S': s3_path}}
        )
        current_error = response.get('Item', {}).get('error_message', {}).get('S', '')
    except:
        current_error = ""
    if error_msg:
        # Error case
        error_msg = (error_msg or "Unknown error")[:1000]
        update_expr = "SET error_message = :err"
        attr_values = {':err': {'S': error_msg}}
        if current_error:
            update_expr += ", prev_error_message = :prev"
            attr_values[':prev'] = {'S': current_error}
        update_kwargs = {'TableName': 'Iceberg_migration','Key': {'s3_path': {'S': s3_path}},'UpdateExpression': update_expr,'ExpressionAttributeValues': attr_values}
        self.logger.error(f"Set error for {s3_path}: {error_msg}")
    else:
        # Success case
        update_kwargs = {
            'TableName': 'Iceberg_migration',
            'Key': {'s3_path': {'S': s3_path}},
            'UpdateExpression': 'SET #s = :status, #m = :meta, #p = :prev, #e = :err',
            'ExpressionAttributeNames': {'#s': 'status','#p': 'prev_error_message','#e': 'error_message','#m': 'metadata_location'
            },
            'ExpressionAttributeValues': {
                ':status': {'S': 'Iceberg_Metadata_Populated and Registered'},
                ':prev': {'S': current_error},
                ':err': {'S': ''},
                ':meta': {'S': metadata_loc}
            }
        }
        self.logger.info(f"Updated DynamoDB status for {s3_path}: {metadata_loc}")

This ensures that any errors are logged and saved to DynamoDB as error_message. On successive retries, previous errors move to prev_error_message and new errors overwrite error_message. Successful operations clear error_message and archive the last error.

Protecting your data from unintended deletion

To protect your data from unintended deletion, never delete data or metadata files from Amazon S3 directly. Iceberg tables that are registered in AWS Glue or Athena are managed tables and should be deleted using the DROP TABLE command from Spark or Athena. The DROP TABLE command deletes both the table metadata and the underlying data files in S3. See Appendix D for more information.

Clean up

Complete the following steps to clean up your resources:

Delete the DynamoDB control table
Delete the database and tables
Delete the EMR clusters and AWS Glue job used for testing

Conclusion

In this post, we showed you how to modernize your Parquet-based data lake into an Apache Iceberg–powered lakehouse without rewriting or duplicating data. You learned two complementary approaches for this in-place migration:

Migrate and register – Ideal for converting existing Hive-registered Parquet tables into Iceberg-managed tables.
add_files – Best suited for quickly onboarding raw Parquet data into Iceberg without rewriting files.

Both approaches benefit from DynamoDB centralized state tracking, which enables retries, error auditing, and lifecycle management across multiple datasets.

By combining Apache Iceberg with Amazon EMR, AWS Glue, and Amazon DynamoDB, you can create a production-ready migration pipeline that is observable, automated, and straightforward to extend to future data format upgrades. This pattern forms a solid foundation for building an Iceberg-based lakehouse on AWS, helping you achieve faster analytics, better data governance, and long-term flexibility for evolving workloads.

To get started, try implementing this solution using the sample tables (table1 and table2) that you created using Athena queries. we encourage you to share your migration experiences and questions in the comments.

Appendix A — Creating an EMR cluster for Hive metastore using console and AWS CLI

Console steps:

Open AWS Management Console for Amazon EMR and choose Create cluster.
Select Spark or Hive under applications.
Under AWS Glue Data Catalog settings, make sure the following options are not selected:
- Use for Hive table metadata
- Use for Spark table metadata
Configure SSH access (KeyName).
Configure network (VPC, subnets, SGs) to allow access to S3.

AWS CLI steps:

aws emr create-cluster \
  --region us-east-1 \
  --name "IcebergHiveCluster711" \
  --release-label emr-7.11.0 \
  --applications Name=Hive Name=Spark Name=Hadoop \
  --ec2-attributes '{"KeyName":"<key-pair>","SubnetId":"<subnet-id>"}'  \
  --instance-groups '[
    {
      "Name":"Master",
      "InstanceGroupType":"MASTER",
      "InstanceType":"m5.xlarge",
      "InstanceCount":1
    },
    {
      "Name":"Workers",
      "InstanceGroupType":"CORE",
      "InstanceType":"m5.xlarge",
      "InstanceCount":2
    }
  ]' \
  --use-default-roles

Appendix B — EMR cluster with AWS Glue as Spark Metastore

Console steps:

Open the Amazon EMR console, choose Create cluster and then select EMR Serverless or provisioned EMR.
Under Software Configuration, verify that Spark is installed.
Under AWS Glue Data Catalog settings, select Use Glue Data Catalog for Spark metadata.
Configure SSH access (KeyName).
Configure network settings (VPC, subnets, and security groups) to allow access to Amazon S3 and AWS Glue.

AWS CLI (provisioned Amazon EMR):

aws emr create-cluster \
  --region us-east-1 \
  --name "IcebergGlueCluster711" \
  --release-label emr-7.11.0 \
  --applications Name=Spark Name=Hadoop \
  --ec2-attributes '{"KeyName":"<key-pair>","SubnetId":"<subnet-id>"}' \
  --instance-groups '[
    {
      "Name":"Master",
      "InstanceGroupType":"MASTER",
      "InstanceType":"m5.xlarge",
      "InstanceCount":1
    },
    {
      "Name":"Workers",
      "InstanceGroupType":"CORE",
      "InstanceType":"m5.xlarge",
      "InstanceCount":2
    }
  ]' \
 --configurations '[{"Classification":"spark-hive-site","Properties":{"hive.metastore.client.factory.class":"com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"}}]' \
 --use-default-roles

Appendix C — Non-Hive partitioned datasets and Iceberg add_files

This appendix explains why a direct in-place migration using an add_files-style procedure might not behave as expected for datasets that aren’t Hive-partitioned and shows recommended fixes and examples.

AWS Glue and Athena follow Hive-style partitioning, where partition column values are encoded in the S3 path rather than inside the data files. For example, following the Parquet dataset created in the Create Sample Parquet Dataset as a source section of this post:

s3://amzn-s3-demo-bucket/events/event_date=2024-09-01/hour=5/part-0000.parquet
s3://amzn-s3-demo-bucket/events/event_date=2024-09-02/hour=5/part-0001.parquet

Partition columns (event_date, hour) are represented in the folder structure.
Non-partition columns (for example, id, name, age) remain inside the Parquet files.
Iceberg add_files can correctly map partitions based on the folder path, even if partition columns are missing from the Parquet file itself.

Partition column	Stored in path	Stored in file	Athena or AWS Glue and Iceberg behavior
event_date	Yes	Yes	Partitions inferred correctly
hour	Yes	No	Partitions still inferred from path

Non-Hive partitioning layout (problem case)

s3://amzn-s3-demo-bucket/events/date/part-0000.parquet
s3://amzn-s3-demo-bucket/events/date/part-0001.parquet

No partition columns in the path.
File might not contain partition columns.

If you try to create an empty Iceberg table and directly load it using add_files on a non-hive layout, the following happens:

Iceberg cannot automatically map partitions, add_files operations fail or register files with incorrect or missing partition metadata.
Queries in Athena or AWS Glue will return unexpected NULLs or incomplete results.
Successive incremental writes using add_files will fail.

Recommended approaches:

Create an AWS Glue table and use the Iceberg snapshot procedure:

Create a table in AWS Glue pointing to your existing Parquet dataset.

You might need to manually provide the schema because glue crawler might fail to automatically infer it for you.

Use Iceberg’ s snapshot procedure to convert and move the AWS Glue table into your target Iceberg table.

This works because Iceberg relies on AWS Glue for schema inference, so this approach ensures correct mapping of columns and partitions without rewriting the data. For more information, see Snapshot procedure.

Appendix D — Understanding table types: Managed compared to external

By default, all non-Iceberg tables created in AWS Glue or Athena are external tables, Athena doesn’t manage the underlying data. If you use CREATE TABLE without the EXTERNAL keyword for non-Iceberg tables, Athena issues an error.

However, when dealing with Iceberg tables, AWS Glue and Athena also manage the underlying data for the respective tables, so these tables are treated as internal tables.

Running DROP TABLE on Iceberg tables will delete the table and the underlying data.

The following table describes how the effect of DELETE and DROP TABLE actions on Iceberg tables in AWS Glue and Athena:

Operation	What it does	Effect on S3 data
DELETE FROM mydb.products_iceberg WHERE date = 2025-10-06;	Creates new snapshot, hides deleted rows	Data files stay until cleanup
DROP TABLE test_db.table1;	Deletes table and all data	Files are permanently removed

About the authors

Using Amazon EMR DeltaStreamer to stream data to multiple Apache Hudi tables

2026-01-15 Gautam Bhaghavatula

Post Syndicated from Gautam Bhaghavatula original https://aws.amazon.com/blogs/big-data/using-amazon-emr-deltastreamer-to-stream-data-to-multiple-apache-hudi-tables/

In this post, we show you how to implement real-time data ingestion from multiple Kafka topics to Apache Hudi tables using Amazon EMR. This solution streamlines data ingestion by processing multiple Amazon Managed Streaming for Apache Kafka (Amazon MSK) topics in parallel while providing data quality and scalability through change data capture (CDC) and Apache Hudi.

Organizations processing real-time data changes across multiple sources often struggle with maintaining data consistency and managing resource costs. Traditional batch processing requires reprocessing entire datasets, leading to high resource usage and delayed analytics. By implementing CDC with Apache Hudi’s MultiTable DeltaStreamer, you can achieve real-time updates; efficient incremental processing with atomicity, consistency, isolation, durability (ACID) guarantees; and seamless schema evolution while minimizing storage and compute costs.

Using Amazon Simple Storage Service (Amazon S3), Amazon CloudWatch, Amazon EMR, Amazon MSK and AWS Glue Data Catalog, you’ll build a production-ready data pipeline that processes changes from multiple data sources simultaneously. Through this tutorial, you’ll learn to configure CDC pipelines, manage table-specific configurations, implement 15-minute sync intervals, and maintain your streaming pipeline. The result is a robust system that maintains data consistency while enabling real-time analytics and efficient resource utilization.

What is CDC?

Imagine a constantly evolving data stream, a river of information where updates flow continuously. CDC acts like a sophisticated net, capturing only the modifications—the inserts, updates, and deletes—happening within that data stream. Through this targeted approach, you can focus on the new and changed data, significantly improving the efficiency of your data pipelines.There are numerous advantages to embracing CDC:

Reduced processing time – Why reprocess the entire dataset when you can focus only on the updates? CDC minimizes processing overhead, saving valuable time and resources.
Real-time insights – With CDC, your data pipelines become more responsive. You can react to changes almost instantaneously, enabling real-time analytics and decision-making.
Simplified data pipelines – Traditional batch processing can lead to complex pipelines. CDC streamlines the process, making data pipelines more manageable and easier to maintain.

Why Apache Hudi?

Hudi simplifies incremental data processing and data pipeline development. This framework efficiently manages business requirements such as data lifecycle and improves data quality. You can use Hudi to manage data at the record-level in Amazon S3 data lakes to simplify CDC and streaming data ingestion and handle data privacy use cases requiring record-level updates and deletes. Datasets managed by Hudi are stored in Amazon S3 using open storage formats, while integrations with Presto, Apache Hive, Apache Spark, and Data Catalog give you near real time access to updated data. Apache Hudi facilitates incremental data processing for Amazon S3 by:

Managing record-level changes – Ideal for update and delete use cases
Open formats – Integrates with Presto, Hive, Spark, and Data Catalog
Schema evolution – Supports dynamic schema changes
HoodieMultiTableDeltaStreamer – Simplifies ingestion into multiple tables using centralized configurations

Hudi MultiTable Delta Streamer

The HoodieMultiTableStreamer offers a streamlined approach to data ingestion from multiple sources into Hudi tables. By processing multiple sources simultaneously through a single DeltaStreamer job, it eliminates the need for separate pipelines while reducing operational complexity. The framework provides flexible configuration options, and you can tailor settings for diverse formats and schemas across different data sources.

One of its key strengths lies in unified data delivery, organizing information in respective Hudi tables for seamless access. The system’s intelligent upsert capabilities efficiently handle both inserts and updates, maintaining data consistency across your pipeline. Additionally, its robust schema evolution support enables your data pipeline to adapt to changing business requirements without disruption, making it an ideal solution for dynamic data environments.

Solution overview

In this section, we show how to stream data to Apache Hudi Table using Amazon MSK. For this example scenario, there are data streams from three distinct sources residing in separate Kafka topics. We aim to implement a streaming pipeline that uses the Hudi DeltaStreamer with multitable support to ingest and process this data at 15-minute intervals.

Mechanism

Using MSK Connect, data from multiple sources flows into MSK topics. These topics are then ingested into Hudi tables using the Hudi MultiTable DeltaStreamer. In this sample implementation, we create three Amazon MSK topics and configure the pipeline to process data in JSON format using JsonKafkaSource, with the flexibility to handle Avro format when needed through the appropriate deserializer configuration

The following diagram illustrates how our solution processes data from multiple source databases through Amazon MSK and Apache Hudi to enable analytics in Amazon Athena. Source databases send their data changes—including inserts, updates, and deletes—to dedicated topics in Amazon MSK, where each data source maintains its own Kafka topic for change events. An Amazon EMR cluster runs the Apache Hudi MultiTable DeltaStreamer, which processes these multiple Kafka topics in parallel, transforming the data and writing it to Apache Hudi tables stored in Amazon S3. Data Catalog maintains the metadata for these tables, enabling seamless integration with analytics tools. Finally, Amazon Athena provides SQL query capabilities on the Hudi tables, allowing analysts to run both snapshot and incremental queries on the latest data. This architecture scales horizontally as new data sources are added, with each source getting its dedicated Kafka topic and Hudi table configuration, while maintaining data consistency and ACID guarantees across the entire pipeline.

To set up the solution, you need to complete the following high-level steps:

Set up Amazon MSK and create Kafka topics
Create the Kafka topics
Create table-specific configurations
Launch Amazon EMR cluster
Invoke the Hudi MultiTable DeltaStreamer
Verify and query data

Prerequisites

To perform the solution, you need to have the following prerequisites. For AWS services and permissions, you need:

AWS account:
- Amazon MSK. Refer to Getting started with Amazon MSK.
- Amazon EMR (version 6.15.0 or later recommended).
- Amazon S3.
- Data Catalog.
- IAM roles and policies.
IAM roles:
- Amazon EMR service role (EMR_DefaultRole) with permissions for Amazon S3, AWS Glue and CloudWatch.
- Amazon EC2 instance profile (EMR_EC2_DefaultRole) with S3 read/write access.
- Amazon MSK access role with appropriate permissions.
S3 buckets:
- Configuration bucket for storing properties files and schemas.
- Output bucket for Hudi tables.
- Logging bucket (optional but recommended).
Network configuration:
- Amazon Virtual Private Cloud (Amazon VPC) with appropriate subnets.
- Security groups allowing traffic between Amazon EMR and Amazon MSK.
- NAT gateway or VPC endpoints for Amazon S3 access.
Development tools:
- AWS CLI configured.
- Access to Amazon EMR console or AWS CLI.
- Kafka client tools for testing.

Set up Amazon MSK and create Kafka topics

In this step, you’ll create an MSK cluster and configure the required Kafka topics for your data streams.

To create an MSK cluster:

aws kafka create-cluster \
    --cluster-name hudi-msk-cluster \
    --broker-node-group-info file://broker-nodes.json \
    --kafka-version "2.8.1" \
    --number-of-broker-nodes 3 \
    --encryption-info file://encryption-info.json \
    --client-authentication file://client-authentication.json

Verify the cluster status:

aws kafka describe-cluster --cluster-arn $CLUSTER_ARN | jq '.ClusterInfo.State'

The command should return ACTIVE when the cluster is ready.

Schema setup

To set up the schema, complete the following steps:

Create your schema files.

input_schema.avsc:

{
    "type": "record",
    "name": "CustomerSales",
    "fields": [
        {"name": "Id", "type": "string"},
        {"name": "ts", "type": "long"},
        {"name": "amount", "type": "double"},
        {"name": "customer_id", "type": "string"},
        {"name": "transaction_date", "type": "string"}
    ]
}

output_schema.avsc:

{
    "type": "record",
    "name": "CustomerSalesProcessed",
    "fields": [
        {"name": "Id", "type": "string"},
        {"name": "ts", "type": "long"},
        {"name": "amount", "type": "double"},
        {"name": "customer_id", "type": "string"},
        {"name": "transaction_date", "type": "string"},
        {"name": "processing_timestamp", "type": "string"}
    ]
}

Create and upload schemas to your S3 bucket:

# Create the schema directory
aws s3 mb s3://hudi-config-bucket-$AWS_ACCOUNT_ID
aws s3api put-object --bucket hudi-config-bucket-$AWS_ACCOUNT_ID --key HudiProperties/
# Upload schema files
aws s3 cp input_schema.avsc s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/
aws s3 cp output_schema.avsc s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/

Create the Kafka topics

To create the Kafka topics, complete the following steps:

Get the bootstrap broker string:

# Get bootstrap brokers
BOOTSTRAP_BROKERS=$(aws kafka get-bootstrap-brokers --cluster-arn $CLUSTER_ARN --query 'BootstrapBrokerString' --output text)

Create the required topics:

kafka-topics.sh --create \
    --bootstrap-server $BOOTSTRAP_BROKERS \
    --replication-factor 3 \
    --partitions 3 \
    --topic cust_sales_details
kafka-topics.sh --create \
    --bootstrap-server $BOOTSTRAP_BROKERS \
    --replication-factor 3 \
    --partitions 3 \
    --topic cust_sales_appointment
kafka-topics.sh --create \
    --bootstrap-server $BOOTSTRAP_BROKERS \
    --replication-factor 3 \
    --partitions 3 \
    --topic cust_info

Configure Apache Hudi

The Hudi MultiTable DeltaStreamer configuration is divided into two major components to streamline and standardize data ingestion:

Common configurations – These settings apply across all tables and define the shared properties for ingestion. They include details such as shuffle parallelism, Kafka brokers, and common ingestion configurations for all topics.
Table-specific configurations – Each table has unique requirements, such as the record key, schema file paths, and topic names. These configurations tailor each table’s ingestion process to its schema and data structure.

Create common configuration file

Common Config: kafka-hudi config file where we specify kafka broker and common configuration for all topics as below

Create the kafka-hudi-deltastreamer.properties file with the following properties:

# Common parallelism settings
hoodie.upsert.shuffle.parallelism=2
hoodie.insert.shuffle.parallelism=2
hoodie.delete.shuffle.parallelism=2
hoodie.bulkinsert.shuffle.parallelism=2
# Table ingestion configuration
hoodie.deltastreamer.ingestion.tablesToBeIngested=hudi_sales_tables.cust_sales_details,hudi_sales_tables.cust_sales_appointment,hudi_sales_tables.cust_info
# Table-specific config files
hoodie.deltastreamer.ingestion.hudi_sales_tables.cust_sales_details.configFile=s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/tableProperties/cust_sales_details.properties
hoodie.deltastreamer.ingestion.hudi_sales_tables.cust_sales_appointment.configFile=s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/tableProperties/cust_sales_appointment.properties
hoodie.deltastreamer.ingestion.hudi_sales_tables.cust_info.configFile=s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/tableProperties/cust_info.properties
# Source configuration
hoodie.deltastreamer.source.dfs.root=s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/
# MSK configuration
bootstrap.servers=BOOTSTRAP_BROKERS_PLACEHOLDER
auto.offset.reset=earliest
group.id=hudi_delta_streamer
# Security configuration
hoodie.sensitive.config.keys=ssl,tls,sasl,auth,credentials
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
ssl.endpoint.identification.algorithm=
# Deserializer
hoodie.deltastreamer.source.kafka.value.deserializer.class=io.confluent.kafka.serializers.KafkaAvroDeserializer

Create table-specific configurations

For each topic, create its own configuration with a topic name and primary key details. Complete the following steps:

cust_sales_details.properties:

# Table: cust sales
hoodie.datasource.write.recordkey.field=Id
hoodie.deltastreamer.source.kafka.topic=cust_sales_details
hoodie.deltastreamer.keygen.timebased.timestamp.type=UNIX_TIMESTAMP
hoodie.deltastreamer.keygen.timebased.input.dateformat=yyyy-MM-dd HH:mm:ss.S
hoodie.streamer.schemaprovider.registry.schemaconverter=
hoodie.datasource.write.precombine.field=ts

cust_sales_appointment.properties:

# Table: cust sales appointment
hoodie.datasource.write.recordkey.field=Id
hoodie.deltastreamer.source.kafka.topic=cust_sales_appointment
hoodie.deltastreamer.keygen.timebased.timestamp.type=UNIX_TIMESTAMP
hoodie.deltastreamer.keygen.timebased.input.dateformat=yyyy-MM-dd HH:mm:ss.S hoodie.streamer.schemaprovider.registry.schemaconverter=
hoodie.datasource.write.precombine.field=ts

cust_info.properties:

# Table: cust info
hoodie.datasource.write.recordkey.field=Id
hoodie.deltastreamer.source.kafka.topic=cust_info
hoodie.deltastreamer.keygen.timebased.timestamp.type=UNIX_TIMESTAMP
hoodie.deltastreamer.keygen.timebased.input.dateformat= yyyy-MM-dd HH:mm:ss.S
hoodie.streamer.schemaprovider.registry.schemaconverter=
hoodie.datasource.write.precombine.field=ts
hoodie.deltastreamer.schemaprovider.source.schema.file=-$AWS_ACCOUNT_ID/HudiProperties/input_schema.avsc
hoodie.deltastreamer.schemaprovider.target.schema.file=-$AWS_ACCOUNT_ID/HudiProperties/output_schema.avsc

These configurations form the backbone of Hudi’s ingestion pipeline, enabling efficient data handling and maintaining real-time consistency. Schema configurations define the structure of both source and target data, maintaining seamless data transformation and ingestion. Operational settings control how data is uniquely identified, updated, and processed incrementally.

The following are critical details for setting up Hudi ingestion pipelines:

hoodie.deltastreamer.schemaprovider.source.schema.file – The schema of the source record
hoodie.deltastreamer.schemaprovider.target.schema.file – The schema for the target record
hoodie.deltastreamer.source.kafka.topic – The source MSK topic name
bootstap.servers – The Amazon MSK bootstrap server’s private endpoint
auto.offset.reset – The consumer’s behavior when there is no committed position or when an offset is out of range

Key operational fields to achieve in-place updates for the generated schema include:

hoodie.datasource.write.recordkey.field – The record key field. This is the unique identifier of a record in Hudi.
hoodie.datasource.write.precombine.field – When two records have the same record key value, Apache Hudi picks the one with the largest value for the pre-combined field.
hoodie.datasource.write.operation – The operation on the Hudi dataset. Possible values include UPSERT, INSERT, and BULK_INSERT.

Launch Amazon EMR cluster

This step creates an EMR cluster with Apache Hudi installed. The cluster will run the MultiTable DeltaStreamer to process data from your Kafka topics. To create the EMR cluster, enter the following:

# Create EMR cluster with Hudi installed
aws emr create-cluster \
    --name "Hudi-CDC-Cluster" \
    --release-label emr-6.15.0 \
    --applications Name=Hadoop Name=Spark Name=Hive Name=Livy \
    --ec2-attributes KeyName=myKey,SubnetId=$SUBNET_ID,InstanceProfile=EMR_EC2_InstanceProfile \
    --service-role EMR_ServiceRole \
    --instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=m5.xlarge InstanceGroupType=CORE,InstanceCount=2,InstanceType=m5.xlarge \
    --configurations file://emr-configurations.json \
    --bootstrap-actions Name="Install Hudi",Path="s3://hudi-config-bucket-$AWS_ACCOUNT_ID/bootstrap-hudi.sh"

Invoke the Hudi MultiTable DeltaStreamer

This step configures and starts the DeltaStreamer job that will continuously process data from your Kafka topics into Hudi tables. Complete the following steps:

Connect to the Amazon EMR master node:

# Get master node public DNS
MASTER_DNS=$(aws emr describe-cluster --cluster-id $CLUSTER_ID --query 'Cluster.MasterPublicDnsName' --output text)

# SSH to master node
ssh -i myKey.pem hadoop@$MASTER_DNS

Execute the DeltaStreamer job:

# 
spark-submit --deploy-mode client \
  --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog" \
  --conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" \
  --jars "/usr/lib/hudi/hudi-utilities-bundle_2.12-0.14.0-amzn-0.jar,/usr/lib/hudi/hudi-spark-bundle.jar" \
  --class "org.apache.hudi.utilities.deltastreamer.HoodieMultiTableDeltaStreamer" \
  /usr/lib/hudi/hudi-utilities-bundle_2.12-0.14.0-amzn-0.jar \
  --props s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/kafka-hudi-deltastreamer.properties \
  --config-folder s3://hudi-config-bucket-$AWS_ACCOUNT_ID/HudiProperties/tableProperties/ \
  --table-type MERGE_ON_READ \
  --base-path-prefix s3://hudi-data-bucket-$AWS_ACCOUNT_ID/hudi/ \
  --source-class org.apache.hudi.utilities.sources.JsonKafkaSource \
  --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider \
  --op UPSERT

For continuous mode, you need to add the following property:


--continuous \
--min-sync-interval-seconds 900

With the job configured and running on Amazon EMR, the Hudi MultiTable DeltaStreamer efficiently manages real-time data ingestion into your Amazon S3 data lake.

Verify and query data

To verify and query the data, complete the following steps:

# Start Spark shell
spark-shell --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog" \
  --conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" \
  --jars "/usr/lib/hudi/hudi-spark-bundle.jar"

# In Spark shell
spark.sql("CREATE DATABASE IF NOT EXISTS hudi_sales_tables")

spark.sql("""
CREATE TABLE hudi_sales_tables.cust_sales_details
USING hudi
LOCATION 's3://hudi-data-bucket-$AWS_ACCOUNT_ID/hudi/hudi_sales_tables.cust_sales_details'
""")

# Repeat for other tables

Query with Athena:

-- Sample query
SELECT * FROM hudi_sales_tables.cust_sales_details LIMIT 10;

You can use Amazon CloudWatch alarms to alert you of issues with the EMR job or data processing. To create a CloudWatch alarm to monitor EMR job failures, enter the following:

aws cloudwatch put-metric-alarm \
    --alarm-name EMR-Hudi-Job-Failure \
    --metric-name JobsFailed \
    --namespace AWS/ElasticMapReduce \
    --statistic Sum \
    --period 300 \
    --threshold 1 \
    --comparison-operator GreaterThanOrEqualToThreshold \
    --dimensions Name=JobFlowId,Value=$CLUSTER_ID \
    --evaluation-periods 1 \
    --alarm-actions $SNS_TOPIC_ARN

Real-world impact of Hudi CDC pipelines

With the pipeline configured and running, you can achieve real-time updates to your data lake, enabling faster analytics and decision-making. For instance:

Analytics – Up-to-date inventory data maintains accurate dashboards for ecommerce platforms.
Monitoring – CloudWatch metrics confirm the pipeline’s health and efficiency.
Flexibility – The seamless handling of schema evolution minimizes downtime and data inconsistencies.

Cleanup

To avoid incurring future charges, follow these steps to clean up resources:

Conclusion

In this post, we showed how you can build a scalable data ingestion pipeline using Apache Hudi’s MultiTable DeltaStreamer on Amazon EMR to process data from multiple Amazon MSK topics. You learned how to configure CDC with Apache Hudi, set up real-time data processing with 15-minute sync intervals, and maintain data consistency across multiple sources in your Amazon S3 data lake.

To learn more, explore these resources:

By combining CDC with Apache Hudi, you can build efficient, real-time data pipelines. The streamlined ingestion processes simplify management, enhance scalability, and maintain data quality, making this approach a cornerstone of modern data architectures.