AWS Big Data Blog

Access Amazon S3 data files directly using AWS Lake Formation permissions

Aarthi Srinivasan — Fri, 12 Jun 2026 16:32:19 +0000

Data scientists and ML engineers often need to access raw data files in Amazon Simple Storage Service (Amazon S3) for machine learning training, data exploration, and generative AI workflows. However, when table-level access is governed by AWS Lake Formation, accessing the underlying S3 files has required maintaining separate permission mechanisms. S3 bucket policies or AWS Identity and Access Management (IAM) role policies create operational overhead and risk of permission drift.

Lake Formation now supports direct access to S3 data file locations for tables whose permissions it manages. Previously, data scientists with Lake Formation permissions on AWS Glue Data Catalog tables could query them using spark.sql(). Now, they can also read and write the underlying S3 data files using spark.read.parquet() or spark.read.csv() from Amazon EMR Spark jobs, Amazon SageMaker Unified Studio notebooks with EMR compute, and custom applications. All access is governed by the same Lake Formation permissions.

This capability is powered by the new GetTemporaryDataLocationCredentials() API, which vends temporary credentials scoped to registered S3 locations when callers have appropriate Lake Formation permissions on the corresponding Data Catalog tables. This eliminates the need to manage separate S3 bucket policies for file-level access while maintaining fine-grained access control in Lake Formation for table-based access. It enables your data scientists to explore S3 datasets securely, accelerate machine learning pipelines, and build generative AI workflows without compromising governance.

In this post, we demonstrate reading from and writing to Lake Formation-managed S3 locations using Apache Spark jobs from EMR. Lake Formation credential vending for S3 location access is available in EMR release label 7.13 and later, Boto3 1.42.29 and later, AWS Java SDK 2.41.32 and later, and AWS Command Line Interface (AWS CLI) version 2.33.1 and later.

Key use cases for Lake Formation permissions to S3 locations

Unified permissions for Analytics and Machine Learning pipelines – Data scientists can access both structured tables through SQL queries and underlying data files through programmatic APIs for machine learning and AI workloads. They are empowered to use tools of their choice – for example, use Amazon Athena for SQL analytics with the table names while read and write to the underlying files in their SageMaker notebook or Spark application with spark.read.parquet(“s3://bucket/database_path/table_files/).
Enable AI ready data lakes – Machine learning pipelines can read training data directly from governed data lakes. Generative AI applications can access foundation model training datasets, and data exploration workflows to use native file APIs while maintaining centralized governance and compliance.
Reduced operational complexity – Operations teams don’t need to maintain separate permission policies – one in Lake Formation for table access and another in S3 bucket policies or AWS Identity and Access Management (IAM) roles for file access. This reduces the risk of permission mismatches and avoids inconsistent access control.
Unified audit capability – Auditors do not need to examine multiple log sources, such as S3 Access Logs, AWS CloudTrail events from different services, to understand who accessed what data and when. With this feature, you get a unified CloudTrail audit trail showing both table access through SQL engines and file access through direct APIs, with each access event linked to the Lake Formation permission grant.

What customers are saying

“Through our close collaboration with AWS, Lake Formation’s new S3 location-based permissions have transformed how we manage data governance at Intuit. By unifying two separate access mechanisms for the same data into one unified permission model, we’ve dramatically reduced complexity and streamlined our auditing process. This is exactly the kind of simplification that lets our teams move faster without compromising security, ensuring we maintain the strict compliance and governance standards our regulators expect.”

— Tapan Upadhyay, Group Engineering Manager, Intuit

Lake Formation Credential Vending Plugin for AWS SDK v2 for Java

Lake Formation has made available a specialized library AWS Lake Formation Credential Vending Plugin for AWS SDK V2 for Java. The Java plugin intercepts S3 requests for data, checks Lake Formation permissions for the requested location, and provides temporary scoped credentials to the client if permissions are granted in Lake Formation. If the S3 location access permissions are not managed by Lake Formation, the plugin checks for access in Amazon S3 Access Grants and lastly falls back to IAM permissions. The plugin is supported independently of Spark and comes as an enhancement to EMR Spark Full Table Access (FTA) mode, starting in EMR 7.13 and later. The plugin is integrated at the S3A level. Therefore, any client of S3A can enable it by setting the S3A configurations, in addition to the EMR Lake Formation Full Table Access (FTA) configuration as follows:

fs.s3a.lakeformation.access.grants.enabled = true
fs.s3a.lakeformation.access.grants.fallback.to.iam = true

With the Java plugin, you can enable governance for data lake resources in your custom applications with Lake Formation permissions – managing both fine grained access for users requiring restricted access on Data Catalog tables while providing direct S3 object level access to use-cases that require them.

Note: (1) The principal that will be accessing direct S3 locations of the tables will require full table access. That is, Lake Formation SELECT permission on all columns and rows of the table is required. (2) The Spark cluster needs FTA configuration. (3) Currently, Apache Iceberg table format is not supported with this plugin.

Solution overview

A financial services company runs daily ETL jobs using Spark in EMR. They process raw transaction records in S3 and store the processed records in another S3 location. The transformed Parquet data is registered with Lake Formation and cataloged as a table in Data Catalog. The ETL job will have direct IAM access to the raw data location, while it uses Lake Formation permissions to write to and read from the curated table location. Downstream, a data-analyst role will query the curated table, with restricted column access. The solution is shown in Figure 1.

Figure 1 – Architecture shows EMR Spark writing curated records to the S3 location of a table using Lake Formation permissions while Data-Analyst queries the same table with Lake Formation fine grained access control in Athena.

Prerequisites

To get started exploring this feature, we recommend you have the following setup.

An AWS account with a Lake Formation administrator configured. Refer to Data lake administrator permissions and Set up AWS Lake Formation. You can also refer to the blog Simplify data access for your enterprise using Amazon SageMaker Lakehouse for the Lake Formation admin setup in your AWS account. For ease of following along, you can use an IAM administrator role, registered as a Lake Formation administrator role.
An S3 bucket to host the sample table data.
An IAM role to register the preceding table data in your S3 with Lake Formation. Refer the permissions policy and trust policy for this role in Requirements for roles used to register locations.
To run the Spark code in EMR, you can choose to run the code in either SageMaker Unified Studio with EMR compute or use EMR cluster from EMR console. In the case of SageMaker Unified Studio domain and project, the Lake Formation permissions for the table location will be granted to the project execution role. In this post, we will illustrate using an EMR on EC2 cluster and a runtime role to submit the Spark script as a step to the cluster. For instructions to launch an EMR on EC2 cluster with Lake Formation full table access enabled, refer to instructions here – Lake Formation full table access for Amazon EMR on EC2 and Introducing runtime roles for Amazon EMR steps: Use IAM roles and AWS Lake Formation for access control with Amazon EMR. Fine Grained Access Control (FGAC) option is not supported for Spark on EMR with this feature since S3 location permission is full file path access.
An IAM role Data-Analyst, with permissions as detailed in Data analyst permissions.

Solution walkthrough

First, we will get the setup ready with S3, sample database, table, and data. We will add a raw data set to S3 location, create a table with parquet data in another S3 location that represents the curated dataset for further downstream consumption. We will register the table data location with Lake Formation and grant permissions for the EMR run time role and Data-Analyst role.

Your S3 bucket will have the following structure.

Raw data – s3://<your-bucket-name>/raw/transactions/dt=2024-03-21/

Process data for table – s3://<your-bucket-name>/processed/transactions/

Spark script – s3://<your-bucket-name>/scripts/

Logs for the EMR cluster – s3://<your-bucket-name>/logs/

Step 1 – Create a parquet table in Data Catalog

From the Athena console query editor, create a table in Data Catalog.

-- Create a database
CREATE DATABASE finance_db;

-- Create an external table pointing to the S3 location
CREATE EXTERNAL TABLE IF NOT EXISTS finance_db.transactions_processed (
    transaction_id STRING,
    merchant_name STRING,
    amount DECIMAL(18,2),
    currency STRING,
    account_number STRING,
    card_type STRING,
    status STRING,
    region STRING
)
PARTITIONED BY (transaction_date DATE)
STORED AS PARQUET
LOCATION 's3:///processed/transactions/'
TBLPROPERTIES (
    'parquet.compress'='SNAPPY'
);

Step 2 – Register S3 location and grant table permission to IAM roles in Lake Formation

2.1 Register the table data location s3://<your-bucket-name>/processed/transactions/ with Lake Formation in Lake Formation mode using the custom S3 registration IAM role. For details on how to register locations with Lake Formation, refer Adding an Amazon S3 location to your data lake.

2.2 Grant DESCRIBE permission on the database finance_db and ALL permission on the table transactions_processed to your EMR runtime role.

2.3 Grant Data location permission to EMR runtime role on the curated table’s location. This is to allow writing to that location.

2.4 Grant DESCRIBE permission on the database finance_db and SELECT permission on the table transactions_processed to your Data-Analyst role. Exclude the columns transaction_id and account_number while granting SELECT permissions on the table to the Data-Analyst role.

For details on how to grant Lake Formation permissions, refer Granting database permissions using the named resource method; Granting table permissions using the named resource method and Granting data location permissions.

Step 3 – Run ETL script in EMR

3.1 Download the script bdb-5860-script.py.

3.2 Edit the S3 bucket name placeholder in the script (RAW_PATH and TABLE_PATH) to your resource names and upload to your S3 path s3://<your-bucket-name>/scripts/.

3.3 Make sure your EMR runtime role has access to the script location in its IAM policy permissions.

3.4 Submit and run the script as a step to the EMR cluster, following instructions at Add a Spark step.

What does the script do?

It populates raw records of transaction data into a Spark data frame, writes to the raw data bucket location using IAM permissions on the EMR runtime role. We apply some transformations and write directly to the S3 location of the table that is registered with Lake Formation, from the data frame using Spark’s native Parquet writer.

The following figure shows the stdout of the step.

The Java plugin integrated into EMR 7.13 automatically handles the access for the table’s data location registered with Lake Formation, so you don’t need to manually call the GetTemporaryDataLocationCredentials() API. In this example, the table data location s3://<your-bucket-name>/processed/transactions/ is registered with Lake Formation, for which EMR runtime role is granted ALL permissions. The direct S3 location access support by Lake Formation allows reading and writing to the location directly using Spark data frame.

Step 4 – Run query as Data-Analyst using Athena

SELECT * FROM finance_db.transactions_processed WHERE status = 'DECLINED' AND transaction_date=DATE '2024-03-21';

The Data-Analyst role should see all but two columns of the table.

With these steps complete, we’ve read from and written to direct S3 locations using Spark data frames with the syntax s3://bucketname/prefix/, and accessed the same data using database_name.table_name syntax with Lake Formation permissions. This shows fine-grained access at table level and coarse-grained access at the file path level.

Clean up

To avoid incurring costs, clean up the resources you created for this post.

Delete the Data Catalog database and tables. This removes the related Lake Formation permissions too. Remove the S3 bucket registration from Lake Formation.
Delete the data files, logs, and the PySpark script of this post from your S3 bucket.
Terminate the EMR cluster.

Conclusion

In this post, we showed how to use Lake Formation’s direct S3 location access to read and write data files using Spark data frames from Amazon EMR, while maintaining unified governance through Lake Formation permissions. We walked through the GetTemporaryDataLocationCredentials() API and the AWS Lake Formation Credential Vending Plugin for AWS SDK v2 for Java, which is integrated into EMR release labels 7.13 and later.

This capability unifies permission management for both fine-grained table-based access and direct S3 file path access in Lake Formation. Your data scientists can now use spark.read.parquet() and spark.write alongside spark.sql(), governed by the same permissions, audited in the same CloudTrail logs, and managed from a single console.

To get started, launch an EMR 7.13 cluster and start exploring the feature. Here are some additional resources:

Lake Formation documentation on Accessing Amazon S3 locations
EMR documentation on S3 path-based access using Lake Formation for Amazon EMR Spark

Acknowledgements: We would like to thank all the team members who worked to launch this feature successfully – Rajas Bhate, Akhil Yendluri, Kunal Parikh, Sharda Khubchandani, Dhananjay Badaya, Santhosh Padmanabhan, Nitin Agrawal and Sandeep Adwankar.

About the authors

Building AI shopping agent using Amazon Bedrock AgentCore Runtime and Amazon OpenSearch Service

Omama Khurshid — Thu, 11 Jun 2026 15:27:16 +0000

In this post, we explore how to build an online shopping AI agent. We focus on its architecture and implementation with Amazon OpenSearch Service, Amazon Bedrock AgentCore, and Strands Agents. Amazon Bedrock AgentCore is an agentic platform for deploying and operating those agents and tools securely at scale without managing infrastructure. AgentCore Runtime is the secure, serverless runtime that hosts your Strands Agents and tools as containerized applications. Strands Agents is an open source SDK for building AI agents. In this SDK, an agent is defined by a model, tools, and a prompt. Tools are callable functions that allow agents to perform actions beyond text generation, such as API calls, database queries, and file operations. The framework lets the model autonomously plan steps and invoke tools to complete tasks.

Today’s AI shopping assistants understand natural language, context, and shopping intent, creating a more human-like interaction. These assistants handle complex shopping requirements, such as “Find me a formal dress under $200 that’s appropriate for a summer wedding.” They maintain conversation history, process follow-up questions naturally, and provide personalized recommendations based on user preferences and past interactions. Customers can use visual search to upload images of items that they want, and the AI finds similar products across multiple retailers, matching styles and patterns. The goal is to provide instant, relevant, and personalized assistance at scale, creating a more efficient shopping journey for consumers worldwide.

AI agents combined with Retrieval Augmented Generation (RAG) on Amazon OpenSearch Service represent an evolution in conversational search. This integration builds AI agents on enriched catalogs, supporting context-aware and autonomous search experiences while maintaining accuracy and relevance through grounded responses.

Solution overview

The following diagram illustrates the solution architecture of an AI-powered online shopping agent built using Strands Agents, Amazon Bedrock AgentCore Runtime, and Amazon OpenSearch Service. For simplicity, the diagram doesn’t show authentication and authorization. In a production setup, secure access to the backend by using mechanisms such as Amazon API Gateway, AWS Identity and Access Management (IAM) roles, or OAuth-based authentication.

The following is a walkthrough of the reference architecture:

The user submits a question through the front-end application. AgentCore Runtime receives the request and routes it to the Strands Retail Agent.
The Strands Agent processes the task and invokes the search_product_catalog tool.
OpenSearch Service performs semantic search and returns relevant product results.
The Strands Agent invokes Amazon Bedrock large language models (LLMs) to generate a natural language response.
The agent response is returned to the user through the front end.

Walkthrough

The following section walks you through how to build an online shopping AI agent.

Prerequisites

To implement this solution, you need an AWS account. You also need an OpenSearch Service domain with OpenSearch version 2.13 or later. You can use an existing domain or create a new domain.

To use the vector search capabilities of OpenSearch Service with Strands Agents on AgentCore, you use ingest pipelines. These ingestion pipelines apply built-in processors to pre-process your documents before you index them in OpenSearch Service.

You use the text_embedding processor, which relies on the ML Commons plugin and a registered embedding model—Amazon Nova Multimodal Embeddings on Amazon Bedrock. OpenSearch Service uses the ML Commons plugin to generate vector embedding for your data and uses the same model to convert incoming queries into vectors. This supports semantic search across your indexed content.

You extend your semantic search backend by adding an agent built with Strands Agents and deployed on Amazon Bedrock AgentCore.

Code samples provided in this post are tested in Python 3.11. You only need to install Python 3.11 in your environment to execute the python scripts. You also need Node.js 18 or later installed to use the AgentCore CLI. The provided code scripts will deploy into your AWS account so make sure your terminal has access to necessary AWS credentials.

Install AgentCore CLI

Install the AgentCore CLI globally using npm:

npm install -g @aws/agentcore

Python Dependencies

You also need to create a requirements.txt file with following dependencies in your workspace to deploy the agents.

boto3
uv
opensearch-py
requests-aws4auth
strands-agents
strands-agents-tools
bedrock-agentcore

Run pip install -r requirements.txt in your terminal to install the required dependencies. To avoid conflicts with other dependencies in your system, you can use a virtual environment.

Now, walk through each step.

Step 1: Configure IAM permissions

Complete the following steps to register the Nova Multimodal Embeddings model with OpenSearch Service and verify that your OpenSearch Service domain has permission to invoke the Amazon Bedrock API.

Go to the IAM console and create a new role with a custom trust policy. Add the following trust policy.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Statement1",
      "Effect": "Allow",
      "Principal": {
        "Service": "opensearchservice.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

Skip adding a permission policy.
Give your role a name and create it. For this post, we use OpenSearchBedrockEmbeddingRole as the role name. OpenSearch Service uses this role to invoke the Nova Multimodal Embeddings model on Amazon Bedrock.

On the Permissions tab, attach an inline policy with the following permissions. For this post, we name this policy OpenSearchBedrockEmbeddingPolicy.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Statement1",
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeAgent",
        "bedrock:InvokeModel"
      ],
      "Resource": [
        "arn:aws:bedrock:us-east-1::foundation-model/*"
      ]
    }
  ]
}

Create a passRole policy with the following JSON document and assign it to the IAM role that creates the ML connector. This lets the principal running the Python code pass the OpenSearchBedrockEmbeddingRole to OpenSearch. Replace <your-aws-account-id> with your own AWS account ID.
```
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "arn:aws:iam::\$<your-aws-account-id>:role/OpenSearchBedrockEmbeddingRole"
    }
  ]
}
```
By using fine-grained access control (FGAC), map the IAM role as a backend role for the ml_full_access role in the OpenSearch Dashboards Security plugin. This mapping lets the user create ML connectors:
1. Log in to OpenSearch Dashboards and open the Security page from the navigation menu.
2. Choose Roles and select ml_full_access.
3. Choose Mapped Users and Manage Mapping.
4. Under Backend roles, add the ARN of the IAM role that you created in the previous steps.

Step 2: Connect to the model by using OpenSearch ML Connectors

In this section, you create an ML connector to link OpenSearch Service with the Bedrock Nova Multimodal Embeddings model. You then register and deploy the model so you can use it for neural search queries.

Create a file named create-connector.py with the following code. Replace <your hostname>, <your region>, and <your account id> placeholders within the code.

import boto3
import requests
from requests_aws4auth import AWS4Auth
host = '<your hostname>'##CHANGE THIS
region = '<your region>' ##CHANGE THIS
account_id = '<your account id>' ##CHANGE THIS
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
path = '/_plugins/_ml/connectors/_create'
url = host + path
payload = {
"name": "Amazon Bedrock Nova multimodal model - text embedding",
"description": "Test connector for Amazon Bedrock Nova multimodal model - text embedding",
"version": 1,
"protocol": "aws_sigv4",
"credential": {
"roleArn": f"arn:aws:iam::{account_id}:role/OpenSearchBedrockEmbeddingRole"
},
"parameters": {
"region": region,
"service_name": "bedrock",
"model": "amazon.nova-2-multimodal-embeddings-v1:0",
"input_docs_processed_step_size": 1,
"dimensions": 1024,
"embeddingTypes": [
"float"
],
"truncationMode": "NONE"
},
"actions": [
{
"action_type": "predict",
"method": "POST",
"headers": {
"content-type": "application/json",
"x-amz-content-sha256": "required"
},
"url": "https://bedrock-runtime.${parameters.region}.amazonaws.com/model/${parameters.model}/invoke",
"request_body": "{\\n \\"taskType\\": \\"SINGLE_EMBEDDING\\",\\n \\"singleEmbeddingParams\\": {\\n \\"embeddingPurpose\\": \\"GENERIC_INDEX\\",\\n \\"embeddingDimension\\": ${parameters.dimensions},\\n \\"text\\": {\\n \\"truncationMode\\": \\"${parameters.truncationMode}\\",\\n \\"value\\": \\"${parameters.inputText}\\"\\n }\\n }\\n}",
"pre_process_function": "connector.pre_process.bedrock.nova.text_embedding",
"post_process_function": "connector.post_process.bedrock.nova.embedding"
}
]
}
headers = {"Content-Type": "application/json"}
r = requests.post(url, auth=awsauth, json=payload, headers=headers, timeout=15)
print(r.status_code)
print(r.text)

Run python create-connector.py in your terminal by using the IAM role with ml_full_access and passRole permissions created in the previous step. This script creates a connector between OpenSearch Service and the Bedrock Nova Multimodal Embeddings model.
The program responds with connector_id. Take a note of it. Then, navigate to OpenSearch Dashboards and open Dev Tools. Create a model group against which to register this model in the OpenSearch Service domain.

POST /_plugins/_ml/model_groups/_register
{
  "name": "agent-conversational-search-model-group",
  "description": "A model group for bedrock Nova embedding models used for conversational search"
}

POST /_plugins/_ml/models/_register
{
  "name": "nova-2-multimodal-embedding-v1",
  "function_name": "remote",
  "model_group_id": "<model group id>",
  "description": "Nova 2 Multimodal Embeddings Model",
  "connector_id": "<connector id>",
  "interface": {}
}

Run the following API call to deploy the model. Use the registered model ID from the previous step.

POST /_plugins/_ml/models/<registered-model-id>/_deploy

Step 3: Create an ingest pipeline for data indexing

Use the following code to create an ingest pipeline for data indexing. The pipeline establishes a connection to the embedding model, retrieves the embedding for the title field, and stores it in the OpenSearch index.

PUT /_ingest/pipeline/nova_multimodal_embedding
{
  "description": "Text embedding pipeline using nova_multimodal_embedding",
  "processors": [
    {
      "text_embedding": {
        "model_id": "<deployed model id>",
        "field_map": {
          "title": "title_vector"
        }
      }
    }
  ]
}

Step 4: Create an index for storing data

Create an index named product for storing data by using Dev Tools. This index stores raw text and 1024-dimensional embeddings of the title field, and uses the ingest pipeline you created in the previous step.

PUT /product
{
  "settings": {
    "index": {
      "default_pipeline": "nova_multimodal_embedding",
      "knn": true
    }
  },
  "mappings": {
    "properties": {
      "title": {
        "type": "text",
        "analyzer": "standard"
      },
      "title_vector": {
        "type": "knn_vector",
        "dimension": 1024,
        "method": {
          "name": "hnsw",
          "space_type": "cosinesimil",
          "engine": "lucene"
        }
      },
      "price": {
        "type": "float"
      },
      "description": {
        "type": "text"
      },
      "category": {
        "type": "keyword"
      },
      "image_url": {
        "type": "text"
      },
      "rating": {
        "properties": {
          "rate": {
            "type": "float"
          },
          "count": {
            "type": "integer"
          }
        }
      }
    }
  }
}

Step 5: Ingest sample data

Use the following code to ingest the sample product data in Dev Tools.

POST /_bulk
{"index": {"_index": "product", "_id": "2"}}
{"id":2,"title":"Mens Casual Premium Slim Fit T-Shirts","price":22.3,"description":"Slim-fitting style, contrast raglan long sleeve, three-button henley placket, light weight & soft fabric for breathable and comfortable wearing.","category":"men's clothing","image":"https://fakestoreapi.com/img/71-3HjGNDUL._AC_SY879._SX._UX._SY._UY_.jpg","rating":{"rate":4.1,"count":259}}
{"index": {"_index": "product", "_id": "3"}}
{"id":3,"title":"Mens Cotton Jacket","price":55.99,"description":"great outerwear jackets for Spring/Autumn/Winter, suitable for many occasions, such as working, hiking, camping, mountain/rock climbing, cycling, traveling or other outdoors.","category":"men's clothing","image":"https://fakestoreapi.com/img/71li-ujtlUL._AC_UX679_.jpg","rating":{"rate":4.7,"count":500}}
{"index": {"_index": "product", "_id": "4"}}
{"id":4,"title":"Mens Casual Slim Fit","price":15.99,"description":"The color could be slightly different between on the screen and in practice.","category":"men's clothing","image":"https://fakestoreapi.com/img/71YXzeOuslL._AC_UY879_.jpg","rating":{"rate":2.1,"count":430}}
{"index": {"_index": "product", "_id": "5"}}
{"id":5,"title":"John Hardy Women's Legends Naga Gold & Silver Dragon Station Chain Bracelet","price":695,"description":"From our Legends Collection, the Naga was inspired by the mythical water dragon that protects the ocean's pearl.","category":"jewelery","image":"https://fakestoreapi.com/img/71pWzhdJNwL._AC_UL640_QL65_ML3_.jpg","rating":{"rate":4.6,"count":400}}

Step 6: Query the index

Run the following API call to test semantic search by using the Nova Multimodal Embeddings model.

GET /product/_search
{
  "_source": false,
  "fields": [
    "title",
    "price",
    "category",
    "image"
  ],
  "size": 3,
  "query": {
    "neural": {
      "title_vector": {
        "query_text": "jacket",
        "model_id": "<deployed model id>",
        "k": 5
      }
    }
  }
}

The output of the preceding query should look like the following.

{
  ...
  "hits": {
    "total": {
      "value": 4,
      "relation": "eq"
    },
    "max_score": 0.8333229,
    "hits": [
      {
        "_index": "product",
        "_id": "3",
        "_score": 0.8333229,
        "fields": {
          "image": [
            "https://fakestoreapi.com/img/71li-ujtlUL._AC_UX679_.jpg"
          ],
          "title": [
            "Mens Cotton Jacket"
          ],
          "category": [
            "men's clothing"
          ],
          "price": [
            55.99
          ]
        }
      }
      ...
    ]
  }
}

Step 7: Create an agent with Strands and Bedrock AgentCore Runtime

Now, create the Strands Agent that uses Anthropic Claude Sonnet 4.6 on Amazon Bedrock to search products from the OpenSearch Service index. To do so:

Import the Runtime app with from bedrock_agentcore.runtime import BedrockAgentCoreApp.
Initialize the app in your code with app = BedrockAgentCoreApp().
Create the OpenSearch Service connection and search query with the @tool decorator.
Decorate the invocation function with the @app.entrypoint decorator.
Let AgentCore Runtime control the running of the agent with app.run().

Now, complete the following steps:

Make sure that you have installed the necessary dependencies from the Prerequisites section of this post.

Create and save a file named search_agent.py with the following code. Replace <your hostname>, <your region>, and <your account id> placeholders within the code.

from strands import Agent, tool
import argparse
import json
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands.models import BedrockModel
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth
app = BedrockAgentCoreApp()
@tool
def search_products(query: str, size: int = 5):
try:
# OpenSearch configuration
host = '' ## CHANGE THIS, DOMAIN ENDPOINT WITHOUT HTTPS!
region = '' ##CHANGE THIS
model_id= '' ##CHANGE THIS with your deployed model id in OpenSearch
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
# Create OpenSearch client
client = OpenSearch(
hosts=[{'host': host, 'port': 443}],
http_auth=awsauth,
use_ssl=True,
verify_certs=True,
connection_class=RequestsHttpConnection
)
"""Search products in OpenSearch using neural search"""
search_body = {
"_source": False,
"fields": ["title", "price", "category", "image"],
"size": size,
"query": {
"neural": {
"title_vector": {
"query_text": query,
"model_id": model_id,
"k": 3
}
}
}
}
response = client.search(
body=search_body,
index="product"
)
products = []
for hit in response['hits']['hits']:
fields = hit.get('fields', {})
product = {
'title': fields.get('title', [''])[0] if fields.get('title') else '',
'price': fields.get('price', [''])[0] if fields.get('price') else '',
'category': fields.get('category', [''])[0] if fields.get('category') else '',
'image': fields.get('image', [''])[0] if fields.get('image') else ''
}
products.append(product)
return f"Found {len(products)} products: {json.dumps(products, indent=2)}"
except Exception as e:
return f"Search error: {str(e)}"
model_id = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
model = BedrockModel(
model_id=model_id,
)
agent = Agent(
model=model,
tools=[search_products],
system_prompt="You're a helpful assistant. You can do product search, and tell the product details."
)
@app.entrypoint
def strands_agent_bedrock(payload):
"""
Invoke the agent with a payload
"""
user_input = payload.get("prompt")
print("User input:", user_input)
response = agent(user_input)
return response.message['content'][0]['text']
if __name__ == "__main__":
#strands_agent_bedrock({"prompt": "Search jacket"}) ##UNCOMMENT THIS FOR TESTING
#app.run() ##UNCOMMENT THIS FOR DEPLOYMENT, MAKE SURE THE ABOVE LINE IS COMMENTED WHEN YOU ARE DEPLOYING TO AGENTCORE

This deploys your agent locally for testing purposes.

Navigate to the IAM console and add the AmazonBedrockLimitedAccess permission policy to the principal running the code.
Navigate to OpenSearch Dashboards and, from the left menu, choose Security plugin, then choose Roles.
Choose Create Role.
Name the role agentcore-permissions.
Under cluster permissions, add cluster:admin/opensearch/ml/models/get and cluster:admin/opensearch/ml/predict.
Under index permissions, enter product* as the index pattern. Add search and get permissions.
Create the role.
Choose the role you created, switch to the Mapped Users tab, choose Manage mapping, and add the role that you use for running the Python code as a backend role.
Uncomment the line strands_agent_bedrock({"prompt": "Search jacket"}) and make sure the app.run() line is commented in the code.

Run python search_agent.py in your terminal to start the shopping agent. The output should look similar to the following.

"Here are the jacket search results:\n\n1. **Mens Cotton Jacket** - $55.99\n2. **Mens Casual Slim Fit** - $15.99\n3. **Mens Casual Premium Slim Fit T-Shirts** - $22.30\n4. **John Hardy Women's Legends Naga Gold & Silver Dragon Station Chain Bracelet** - $695.00\n\nThe most relevant jacket option is the **Mens Cotton Jacket** at $55.99. Would you like to know more about any of these products?

Comment strands_agent_bedrock({"prompt": "Search jacket"}) and uncomment the app.run() line in the code before going into the next step.

Step 8: Configure and launch your agent to Bedrock AgentCore Runtime

The AgentCore CLI is a command-line tool provided by AWS that simplifies deployment of agents to Amazon Bedrock AgentCore Runtime. When you run the CLI deployment command, it automates the entire deployment workflow: it creates the necessary IAM execution role with proper permissions, packages your Python application code along with its dependencies, uses AWS CodeBuild to build an optimized Docker container image, pushes that container image to Amazon Elastic Container Registry (ECR), and finally provisions the AgentCore Runtime environment that hosts your containerized agent. This eliminates the need for manual Dockerfile creation, container builds, or infrastructure management.

Before you start this step, make sure you have gone through section 7 and installed the AgentCore CLI and Python dependencies listed in the Prerequisites section.
Create a policy named AgentCoreAccessPolicy with the following permissions and attach it to the role running the code. Replace <ACCOUNT_ID> and <REGION> placeholders.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "BedrockAgentCoreServiceAccess",
      "Effect": "Allow",
      "Action": [
        "bedrock-agentcore:*"
      ],
      "Resource": "*"
    },
    {
      "Sid": "ECRAuthorizationToken",
      "Effect": "Allow",
      "Action": [
        "ecr:GetAuthorizationToken"
      ],
      "Resource": "*"
    },
    {
      "Sid": "ECRRepositoryAccess",
      "Effect": "Allow",
      "Action": [
        "ecr:BatchCheckLayerAvailability",
        "ecr:GetDownloadUrlForLayer",
        "ecr:BatchGetImage",
        "ecr:PutImage",
        "ecr:InitiateLayerUpload",
        "ecr:UploadLayerPart",
        "ecr:CompleteLayerUpload",
        "ecr:DescribeRepositories",
        "ecr:CreateRepository",
        "ecr:ListImages",
        "ecr:DescribeImages"
      ],
      "Resource": [
        "arn:aws:ecr:<REGION>:<ACCOUNT_ID>:repository/bedrock-agentcore-*"
      ]
    },
    {
      "Sid": "CodeBuildProjectAccess",
      "Effect": "Allow",
      "Action": [
        "codebuild:CreateProject",
        "codebuild:UpdateProject",
        "codebuild:StartBuild",
        "codebuild:BatchGetBuilds",
        "codebuild:DeleteProject"
      ],
      "Resource": "*"
    },
    {
      "Sid": "IAMRoleManagement",
      "Effect": "Allow",
      "Action": [
        "iam:CreateRole",
        "iam:AttachRolePolicy",
        "iam:PutRolePolicy",
        "iam:GetRole",
        "iam:GetRolePolicy",
        "iam:PassRole",
        "iam:DeleteRole",
        "iam:DeleteRolePolicy",
        "iam:DetachRolePolicy",
        "iam:CreateServiceLinkedRole"
      ],
      "Resource": [
        "arn:aws:iam::<ACCOUNT_ID>:role/BedrockAgentCoreExecutionRole-*",
        "arn:aws:iam::<ACCOUNT_ID>:role/AmazonBedrockAgentCoreSDKCodeBuild-*",
        "arn:aws:iam::<ACCOUNT_ID>:role/aws-service-role/runtime-identity.bedrock-agentcore.amazonaws.com/AWSServiceRoleForBedrockAgentCoreRuntimeIdentity"
      ]
    },
    {
      "Sid": "CloudWatchLogsAccess",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents",
        "logs:PutResourcePolicy"
      ],
      "Resource": "arn:aws:logs:<REGION>:<ACCOUNT_ID>:log-group:/aws/bedrock-agentcore/*"
    },
    {
      "Sid": "CloudWatchLogsResourcePolicy",
      "Effect": "Allow",
      "Action": [
        "logs:PutResourcePolicy"
      ],
      "Resource": "*"
    },
    {
      "Sid": "S3BucketManagement",
      "Effect": "Allow",
      "Action": [
        "s3:CreateBucket",
        "s3:PutBucketPolicy",
        "s3:PutBucketVersioning",
        "s3:PutBucketPublicAccessBlock",
        "s3:PutLifecycleConfiguration",
        "s3:PutObject",
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::bedrock-agentcore-codebuild-sources-<ACCOUNT_ID>-*",
        "arn:aws:s3:::bedrock-agentcore-codebuild-sources-<ACCOUNT_ID>-*/*"
      ]
    }
  ]
}

Create a file named agentcore.yaml in your project directory with the following configuration. Replace <REGION>, <ACCOUNT_ID>, and <OPENSEARCH_DOMAIN_NAME> placeholders:

# AgentCore Runtime Configuration
  runtime:
    name: shopping-search-agent-runtime
    entrypoint: search_agent.py:strands_agent_bedrock
    region: <REGION>  # CHANGE THIS - Your AWS region (e.g., us-east-1)

    execution_role:
      create: true
      name: BedrockAgentCoreExecutionRole-shopping-agent
      policies:
        - policy_name: BedrockAndOpenSearchAccess
          policy_document:
            Version: "2012-10-17"
            Statement:
              - Sid: BedrockModelAccess
                Effect: Allow
                Action:
                  - bedrock:InvokeModel
                  - bedrock:InvokeModelWithResponseStream
                Resource:
                  - arn:aws:bedrock:*::foundation-model/*
                  - arn:aws:bedrock:<REGION>:<ACCOUNT_ID>:inference-profile/*  # CHANGE THIS - Replace <REGION> and <ACCOUNT_ID>
              - Sid: OpenSearchAccess
                Effect: Allow
                Action:
                  - es:ESHttpGet
                  - es:ESHttpPost
                  - es:ESHttpPut
                Resource: arn:aws:es:<REGION>:<ACCOUNT_ID>:domain/<OPENSEARCH_DOMAIN_NAME>/*  # CHANGE THIS - Replace all three placeholders
              - Sid: CloudWatchLogsAccess
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: arn:aws:logs:<REGION>:<ACCOUNT_ID>:log-group:/aws/bedrock-agentcore/runtimes/*  # CHANGE THIS - Replace <REGION> and <ACCOUNT_ID>
              - Sid: ECRImageAccess
                Effect: Allow
                Action:
                  - ecr:GetAuthorizationToken
                  - ecr:BatchGetImage
                  - ecr:GetDownloadUrlForLayer
                Resource: "*"

    container:
      architecture: arm64  # Options: arm64 or x86_64
      requirements_file: requirements.txt

    ecr:
      auto_create: true

Run the following command in your terminal to deploy the agent to AgentCore Runtime:

Create the AgentCore project and replace with your search agent.

agentcore create --name ShoppingAgent --defaults
cd ShoppingAgent
cp ../search_agent.py app/ShoppingAgent/main.py

Add OpenSearch and other dependencies:

cd app/ShoppingAgent
uv add opensearch-py requests-aws4auth boto3
cd ../..

Deploy the agent to Agentcore Runtime. This process takes approximately 5-10 minutes.
```
agentcore deploy
```

Once deployment completes, verify the runtime status:

agentcore status

You should see:

ShoppingAgent: Deployed - Runtime: READY (arn:aws:bedrock-agentcore:<REGION>:<ACCOUNT_ID>:runtime/ShoppingAgent_...)

URL: https://bedrock-agentcore.<REGION>.amazonaws.com/runtimes/.../invocations

Step 9: Configure OpenSearch Service access

Map your AgentCore execution role to an OpenSearch backend role so the agent can access your data.

Navigate to OpenSearch Dashboards. From the left menu, choose the Security plugin, then choose Roles.
Search for agentcore-permissions and choose the role. Then, navigate to the Mapped Users tab, choose Manage mapping, and add arn:aws:iam::<ACCOUNT_ID>:role/AmazonBedrockAgentCoreSDKRuntime-us-east-1-custom as a backend role. Replace the <ACCOUNT_ID> placeholder with your account ID.

Step 10: Invoke the Bedrock AgentCore Runtime

You can test the agent in Agent Sandbox. Enter the prompt Search jacket less than 50$, and the agent returns the relevant result from the OpenSearch Service index with a summary.

In real-world scenarios, you can design a search application with a Strands Agent deployed in AgentCore Runtime. You can add AgentCore Memory, which gives your AI agents the ability to remember past interactions and provide more context-aware, personalized conversations.

Cleanup

To avoid incurring future charges, delete the resources created while building this solution:

Delete the OpenSearch Service domain.
Delete the Amazon Bedrock AgentCore Runtime resources.

Conclusion

In this post, you saw how to create a conversational search with Amazon OpenSearch Service and Strands Agents. You also learned how to deploy the agent on Amazon Bedrock AgentCore Runtime. You can further enhance this shopping agent by using other AgentCore capabilities. For example, AgentCore Memory retains user preferences and past interactions across sessions, AgentCore Identity manages shopper authentication and access control, and AgentCore Observability helps you monitor and debug agent behavior in production. Together, these services help you build shopping experiences that deliver instant, relevant assistance at scale.

Now it’s your turn. Build your own conversational search experience by integrating OpenSearch Service and Strands Agents with your product catalog. To learn more, see the Amazon OpenSearch Service and Amazon Bedrock AgentCore detail pages.

About the authors

Choosing the right workflow orchestration service for your use case: Amazon MWAA and AWS Step Functions

Rajkumar Raghuwanshi — Wed, 10 Jun 2026 15:32:13 +0000

Whether you’re processing financial data, managing e-commerce orders, or training machine learning (ML) models, efficiently coordinating complex processes is essential. Amazon Web Services (AWS) offers two services for workflow orchestration: Amazon Managed Workflows for Apache Airflow (Amazon MWAA) and AWS Step Functions.

This post explores how to select the right workflow orchestration service based on your specific use case requirements. We’ll examine key workflow characteristics, present real-world scenarios, and provide practical guidance to help you make an informed decision for your particular needs.

Understanding workflow orchestration requirements

Before exploring specific services, consider the key dimensions that influence workflow orchestration needs:

Data statefulness: Does your workflow process independent units of work (stateless) or create dependencies where each step modifies data from previous steps (stateful)?
Execution duration: Are your workflows short-lived (seconds to minutes) or long-running (hours to days)?
Scheduling requirements: Do you need built-in time-based execution or rely primarily on event triggers?
Recovery capabilities: How critical is the ability to restart from specific failure points rather than reprocessing entirely?
Integration complexity: What systems, services, and data sources need to be coordinated?
Security and access control: Do you need fine-grained permissions for different workflow components?

Let’s explore how these requirements map to real-world use cases and the appropriate orchestration solutions.

Use case: Enterprise data analytics pipeline

This scenario illustrates how Amazon MWAA handles complex, stateful data pipelines with built-in scheduling and granular recovery.

Business challenge

A global financial services company processes massive volumes of transaction data daily, requiring sophisticated data analytics capabilities. Their requirements include:

Designed to process 5-10 TB of financial transaction data daily
Running complex extract, transform, and load (ETL) jobs with multiple transformation stages
Generating regulatory reports for compliance use cases
Supporting both scheduled batch processing and event-driven workflows
Capable of handling long-running jobs that can take up to 12 hours
Ensuring data consistency and integrity throughout the pipeline

Workflow characteristics

Data statefulness: Highly stateful workflows where each processing step modifies transaction data, creating dependencies throughout the pipeline
Execution duration: Supports long-running processes extending 2-12 hours
Scheduling needs: Mixed time-based and event-driven patterns
Recovery requirements: Critical ability to resume from specific failure points
Integration complexity: Orchestrates multiple AWS services and external systems

Solution: Amazon Managed Workflows for Apache Airflow (Amazon MWAA)

For this enterprise data analytics scenario, Amazon MWAA provides capabilities that align well with these requirements:

Stateful workflow management

MWAA excels at managing complex, stateful data pipelines where data consistency is critical. When processing terabytes of financial data, MWAA’s ability to resume from the last successful checkpoint helps prevent costly reprocessing and maintain data integrity.

The following code example demonstrates how to structure a complex financial ETL pipeline in MWAA:

# Example: Complex ETL pipeline with proper dependency management
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

dag = DAG(
	'financial_etl_pipeline',
	schedule_interval='0 2 * * *',  # Daily at 2 AM
	start_date=datetime(2024, 1, 1),
	catchup=False
)

# Define tasks
extract_transactions = PythonOperator(task_id='extract_transactions', ...)
extract_market_data = PythonOperator(task_id='extract_market_data', ...)
transform_data = PythonOperator(task_id='transform_data', ...)
load_warehouse = PythonOperator(task_id='load_warehouse', ...)
generate_reports = PythonOperator(task_id='generate_reports', ...)

# Express complex dependencies clearly
[extract_transactions, extract_market_data] >> transform_data >> [load_warehouse, generate_reports]

This Directed Acyclic Graph (DAG) shows how to define task dependencies for parallel data extraction followed by sequential transformation and loading operations. The >> operator clearly defines the workflow dependencies. Transformation only begins after both extraction tasks complete successfully.

Built-in scheduling capabilities

MWAA includes native scheduling capabilities, making it straightforward to set up recurring workflows without additional services. The schedule_interval parameter in the DAG definition provides flexible scheduling options using cron syntax.

Granular recovery and resume control

During production incidents, operations teams can use the MWAA web interface to restart or bypass specific steps with a few clicks. This capability is important for stateful applications where restarting the entire workflow could compromise data consistency.

The MWAA web interface provides a visual representation of the workflow execution, allowing operators to:

Identify failed tasks – Examine task logs for troubleshooting – Clear the status of specific tasks – Restart execution from specific points

Figure 1: A Directed Acyclic Graph (DAG) in MWAA showing parallel execution ofAmazon Redshift Data APItasks. If any task fails, you can re-run specific tasks rather than restarting from the beginning.

Comprehensive monitoring and operational control

MWAA’s metadata server maintains comprehensive execution logs, enabling organizations to build operational dashboards for: – Real-time workflow monitoring – Task completion rate tracking – Pipeline execution pattern analysis – Optimization opportunity identification

Implementation considerations

Infrastructure planning: While MWAA requires capacity planning, the automatic scaling capabilities effectively handle variable workloads by setting minimum and maximum worker counts.
Security model: MWAA uses a shared execution role across DAGs, but you can implement additional security through resource-level policies and separate environments for different teams.
Cost predictability: The worker-hour pricing model provides predictable costs for long-running jobs, making budget planning more straightforward.

Use case: Real-time serverless application orchestration

This scenario shows how AWS Step Functions handles event-driven, serverless workflows that need to scale automatically with unpredictable traffic.

Business challenge

An e-commerce platform needs to orchestrate real-time order processing workflows that can handle thousands of concurrent orders during peak shopping periods. Their requirements include:

Designed for processing customer orders in real-time (targeting sub-second response times)
Coordinating payment validation, inventory checks, and fulfillment
Integrating with multiple AWS services (AWS Lambda, Amazon Simple Queue Service (Amazon SQS), Amazon Simple Notification Service (Amazon SNS), Amazon DynamoDB)
Designed to handle traffic spikes during promotional events
Implementing approval workflows for high-value orders
Maintaining cost efficiency during variable load periods

Workflow characteristics

Data statefulness: Primarily stateless processing where each customer order represents an independent transaction
Execution duration: Supports rapid, real-time processing with sub-second to few-minute response times.
Event-driven nature: Core architectural pattern where workflows are triggered by specific customer actions
Integration requirements: Extensive coordination with AWS serverless services
Scalability needs: Highly unpredictable traffic patterns requiring automatic scaling

Solution: AWS Step Functions

For this real-time e-commerce scenario, AWS Step Functions provides capabilities that align well with these requirements:

Serverless architecture and automatic scaling

Step Functions automatically scales to handle traffic spikes without infrastructure management. During peak shopping events like Black Friday, the service handles increased load without manual intervention.

Event-driven workflow execution

Step Functions is designed for order-triggered workflows that need immediate execution. The following JSON definition shows how to structure an e-commerce order processing workflow:

{
  "Comment": "E-commerce Order Processing Workflow",
  "StartAt": "ValidatePayment",
  "States": {
    "ValidatePayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:region:account:function:ValidatePayment",
      "Retry": [
        {
          "ErrorEquals": ["States.TaskFailed"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ],
      "Next": "CheckInventory"
    },
    "CheckInventory": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "CheckWarehouse1",
          "States": {
            "CheckWarehouse1": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:region:account:function:CheckWarehouse",
              "End": true
            }
          }
        },
        {
          "StartAt": "CheckWarehouse2", 
          "States": {
            "CheckWarehouse2": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:region:account:function:CheckWarehouse",
              "End": true
            }
          }
        }
      ],
      "Next": "ProcessOrder"
    },
    "ProcessOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:region:account:function:ProcessOrder",
      "End": true
    }
  }
}

This Step Functions definition demonstrates several key capabilities: – The ValidatePayment state includes built-in retry logic with exponential backoff – The CheckInventory state uses parallel execution to simultaneously check multiple warehouses – Each Lambda function is called via its Amazon Resource Name (ARN), providing direct integration with AWS services

Figure 2: A complex workflow in AWS Step Functions, involving multiple stages of data processing. The parallel execution doesn’t allow resuming from a specific mid-execution step, but the branching structure provides automated error handling and recovery.

Native AWS service integration

Step Functions provides direct integration with Lambda functions, SQS queues, SNS topics, and DynamoDB, eliminating the need for custom connectors or additional infrastructure components.

Cost-effective pay-per-use model

The pay-per-execution pricing model aligns with variable order volumes, keeping costs minimal during slow periods while scaling automatically during busy times.

Human approval workflow support

Step Functions supports human approval steps, making it suitable for high-value order workflows that require manual review or approval processes.

Implementation considerations

Error handling: Built-in retry mechanisms and error handling patterns help provide reliable order processing with configurable retry policies.
Visual monitoring: The Step Functions console provides real-time visibility into order processing status, enabling quick identification of bottlenecks.
Security model: Fine-grained AWS Identity and Access Management (IAM) roles per step so that payment processing functions have different permissions than inventory management functions.

Choosing the right workflow orchestration service

When selecting between Amazon MWAA and AWS Step Functions, consider these workflow characteristics:

Consider Amazon MWAA when your use case involves:

Complex stateful data processing where workflows modify data state and require recovery mechanisms to maintain consistency
Long-running batch jobs executing for hours or days where computational investment is substantial
Built-in scheduling requirements where regular batch processing needs time-based orchestration
Granular recovery needs where resuming from specific failure points is business-critical
Complex task dependencies involving sophisticated relationships between workflow tasks
Existing Apache Airflow expertise where teams have substantial investment in Apache Airflow knowledge

Consider AWS Step Functions when your use case involves:

Event-driven serverless workflows triggered by external events requiring immediate response
Stateless processing where each workflow execution operates independently
Short to medium duration tasks completing within minutes to hours
Heavy AWS service integration involving extensive coordination with Lambda functions and other AWS services
Human approval workflows requiring manual intervention or decision-making
Variable load patterns with unpredictable traffic requiring automatic scaling

Decision framework

To help guide your decision process, consider the following questions:

Figure 3: Decision tree guiding through key considerations for choosing between Amazon MWAA and AWS Step Functions based on workflow characteristics.

Figure 4: Comprehensive comparison between Amazon MWAA and AWS Step Functions, highlighting decision factors for choosing the right workflow orchestration service.

Conclusion

Both Amazon Managed Workflows for Apache Airflow and AWS Step Functions are workflow orchestration services, each designed to address specific use case requirements. By understanding your workflow characteristics and aligning them with the strengths of each service, you can make an informed decision that supports your business needs.

For complex, stateful workflows with long execution times and sophisticated recovery requirements, Amazon MWAA provides robust capabilities. For event-driven, serverless workflows with tight AWS integration and variable load patterns, AWS Step Functions is a strong fit.

Remember that these services are not mutually exclusive. Many organizations use both to address different workflow orchestration needs across their application portfolio. By focusing on your specific use case requirements, you can select the right tool for each job and build resilient, efficient workflow orchestration solutions on AWS.

If you have questions or feedback about choosing between these services, leave a comment.

About the authors

Real-time CDC from Aurora PostgreSQL to Amazon S3 Tables using Debezium and Firehose

Chintan Agrawal — Wed, 10 Jun 2026 15:29:26 +0000

Enterprises running transactional workloads on Amazon Aurora PostgreSQL-Compatible Edition (Aurora PostgreSQL) need their operational data available for analytics. However, analytical queries and cross-database joins compete for resources on OLTP-optimized clusters. Batch exports introduce latency, and when data spans multiple Aurora clusters, there’s no straightforward way to join datasets or run cross-domain analytics. Real-time change data capture (CDC) addresses this by streaming row-level changes into a separate analytics layer. However, most CDC approaches write append-only records that require downstream consumers to reconstruct current state from the change log.

In this post, we show you how to build a CDC pipeline that delivers query-ready Iceberg tables directly. The pipeline captures inserts, updates, and deletes from Aurora PostgreSQL and applies them as row-level operations in Amazon S3 Tables, a capability of Amazon Simple Storage Service (Amazon S3). The destination tables always reflect the current state of the source database. You use Debezium on Amazon MSK Connect for change capture and Amazon Managed Streaming for Apache Kafka (Amazon MSK) for streaming. You also use AWS Lambda to transform CDC events and resolve operation semantics, and Amazon Data Firehose to deliver records into Iceberg tables. You deploy the infrastructure using the AWS Cloud Development Kit (AWS CDK).

Apache Iceberg supports row-level updates, deletes, ACID transactions, schema evolution, and time travel natively. S3 Tables handles Iceberg snapshot management and compaction automatically. With AWS Lake Formation for access control, multiple teams can query the tables through Amazon Athena, Amazon Redshift, or Amazon SageMaker Unified Studio.

Solution overview

The following diagram shows the architecture of the CDC pipeline.

Figure 1. CDC pipeline architecture from Aurora PostgreSQL to Amazon S3 Tables.

The pipeline uses six components:

Aurora PostgreSQL to Debezium. Debezium runs on MSK Connect in your VPC and uses PostgreSQL’s native logical replication to stream row-level changes from the write-ahead log (WAL), with minimal impact on query performance.
Debezium to Amazon MSK. The ByLogicalTableRouter SMT reroutes CDC events from multiple tables into a single topic (aurora.cdc.all-tables), retaining the source table name in each message.
Amazon MSK to Firehose. Firehose connects to the MSK cluster using the IAM access control over AWS PrivateLink and continuously polls the topic for new messages.
Firehose to Lambda. For each batch, Firehose invokes the Lambda function to decode the Kafka message, flatten the Debezium envelope, and set otfMetadata routing with the destination table and operation type.
Firehose to S3 Tables. Firehose reads the otfMetadata, routes each record to the correct Iceberg table, and performs the appropriate row-level operation using configured unique keys (for example, order_id for orders). S3 Tables handles compaction and snapshot management automatically.
Query and access control. After data lands in S3 Tables, you can query the Iceberg tables with Amazon Athena, Amazon Redshift, or Amazon SageMaker Unified Studio, with AWS Lake Formation managing fine-grained access control.

Firehose supports one MSK topic per delivery stream. The single-topic routing pattern uses a Debezium SMT to consolidate multiple tables into one topic, and a Lambda function to route records to the correct destination. With this, you can serve multiple tables through one Firehose stream, reducing cost and operational complexity.

Debezium event transformation

Debezium produces CDC events in an envelope structure containing both the previous and current state of a row, along with metadata about the source database, table, and operation type. However, Firehose expects records in a flattened JSON format with routing metadata that indicates the target table and operation type.

The Lambda function bridges this gap by performing three operations on each record:

Decode. When Firehose uses Amazon MSK as a source, it delivers the Kafka message value as a base64-encoded string in the kafkaRecordValue field. The function base64-decodes this field to obtain the raw Debezium JSON payload.
Flatten and extract. Pulls the row data from the Debezium envelope. For inserts and updates, the function uses the after field (the row after the change). For deletes, it uses the before field, because the after field is null when a row is removed.
Route. Sets the otfMetadata block with destinationTableName (extracted from the Debezium source.table field) and operation (mapped from Debezium’s single-character codes to Firehose’s operation types).

The following table shows how Debezium operation codes map to Firehose Iceberg operations:

Debezium code	Meaning	Firehose operation
c	Row created (insert)	insert
u	Row updated	update
d	Row deleted	delete
r	Snapshot read (initial load)	insert

When Debezium starts with snapshot.mode=initial, it reads all existing rows and emits them as r (read) events. These represent rows that existed before CDC began, so they are mapped to insert to establish the baseline state in the destination tables.

For example, the function transforms this Debezium envelope:

{
"op": "c",
"before": null,
"after": {"order_id": 1, "customer_id": 1, "total_amount": 299.99},
"source": {"table": "orders", "db": "cdcdemo"}
}

Into a response record with routing metadata:

{
"recordId": "<original-record-id>",
"result": "Ok",
"kafkaRecordValue": "<base64-encoded flattened row JSON>",
"metadata": {
"otfMetadata": {
"destinationDatabaseName": "aurora_cdc",
"destinationTableName": "orders",
"operation": "insert"
}
}
}

The kafkaRecordValue contains the base64-encoded flattened row data (for example, {"order_id": 1, "customer_id": 1, "total_amount": 299.99}), and the otfMetadata block tells Firehose which table to write to and which operation to perform.

With this routing metadata, a single Firehose stream can write to multiple destination tables. For more information, see Route incoming records to different Iceberg tables.

Walkthrough

The following sections walk you through building the CDC pipeline end to end. Before you begin, complete the prerequisites.

Prerequisites

Before you begin, make sure you have the following:

An AWS account with permissions to create the resources described in this post.
An existing Amazon Virtual Private Cloud (Amazon VPC) with at least two subnets in different Availability Zones.
An Aurora PostgreSQL cluster in the same VPC with logical replication enabled (rds.logical_replication = 1).
Aurora database credentials stored in AWS Secrets Manager. Note the secret ARN for the CDK configuration.
AWS CDK v2 installed (npm install -g aws-cdk).
Node.js 18+ and npm.
AWS Command Line Interface (AWS CLI) v2 installed and configured with appropriate credentials.
An Amazon S3 general purpose bucket for the Debezium plugin upload and Firehose failed record backup.
S3 Tables integration with AWS analytics services enabled in your AWS Region (one-time setup).

Step 1: Enable CDC in Aurora PostgreSQL

PostgreSQL supports change data capture through its logical replication framework, which allows database changes to be streamed from the write-ahead log (WAL). Debezium uses this mechanism to continuously read row-level changes and publish them to Kafka topics.

To enable logical replication in Aurora PostgreSQL, configure a custom DB cluster parameter group:

Create a custom parameter group and set the following parameter: rds.logical_replication = 1.
Apply the parameter group to your Aurora cluster and reboot the cluster for the change to take effect.
Connect to your Aurora PostgreSQL cluster and create the source tables:

CREATE TABLE public.orders (
    order_id SERIAL PRIMARY KEY,
    customer_id INTEGER,
    order_date VARCHAR(50),
    total_amount DECIMAL(12,2),
    status VARCHAR(50),
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);
CREATE TABLE public.products (
    product_id SERIAL PRIMARY KEY,
    product_name VARCHAR(255),
    category VARCHAR(100),
    price DECIMAL(10,2),
    stock_quantity INTEGER,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

Create a publication that defines which tables are included in the change stream. Debezium automatically creates the logical replication slot when the connector starts for the first time, so you don’t need to create one manually.

CREATE PUBLICATION dbz_publication FOR TABLE public.orders, public.products;

Verify the publication was created:

SELECT * FROM pg_publication WHERE pubname = 'dbz_publication';

You should see one row returned, confirming the publication is active.

Important: When the Debezium connector starts (Step 6), it creates a replication slot named debezium_slot. This slot retains WAL segments until consumed. If the connector is stopped for an extended period, WAL segments can accumulate and increase storage usage on the Aurora cluster. Monitor the ReplicationSlotDiskUsage Amazon CloudWatch metric for your Aurora cluster.

Step 2: Build and register the Debezium plugin

MSK Connect runs connectors using custom plugins that you upload to Amazon S3. In this step, you download the Debezium PostgreSQL connector, package it as a ZIP file, upload it to S3, and register it with MSK Connect.

First, create an S3 bucket for the plugin, or use an existing metadata management bucket:

aws s3 mb s3://<your-plugin-bucket> --region <your-region>

Download and package the Debezium connector:

DEBEZIUM_VERSION=2.7.3.Final
curl -LO "https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/${DEBEZIUM_VERSION}/debezium-connector-postgres-${DEBEZIUM_VERSION}-plugin.tar.gz"
mkdir -p debezium-plugin
tar -xzf debezium-connector-postgres-${DEBEZIUM_VERSION}-plugin.tar.gz -C debezium-plugin/
cd debezium-plugin && zip -r ../debezium-postgres-connector.zip . && cd ..
aws s3 cp debezium-postgres-connector.zip s3://<your-plugin-bucket>/plugins/

aws kafkaconnect create-custom-plugin \
    --custom-plugin-name debezium-postgres-connector \
    --content-type ZIP \
    --location "s3Location={bucketArn=arn:aws:s3:::<your-plugin-bucket>,fileKey=plugins/debezium-postgres-connector.zip}"

Create a worker configuration that tells MSK Connect to serialize Kafka messages as JSON without schemas:

aws kafkaconnect create-worker-configuration \
    --name debezium-worker-config \
    --properties-file-content "$(echo -n 'key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false' | base64)"

Note the customPluginArn and workerConfigurationArn from the output. You need these for the CDK configuration in the next step.

Note: The custom plugin and worker configuration are created through the AWS CLI because the Debezium connector JARs must be downloaded from the Debezium project and packaged manually. The remaining infrastructure is deployed using the AWS CDK in the following steps.

Step 3: Configure the CDK project

Clone the sample repository and install dependencies:

git clone https://github.com/aws-samples/sample-aurora-cdc-s3tables.git
cd sample-aurora-cdc-s3tables/cdk
npm install

Open cdk/lib/v2/config.ts and update the configuration values to match your environment:

export const CONFIG = {
account: '<your-account-id>',
region: '<your-region>',
// VPC - must match your Aurora cluster's VPC
vpcId: '<your-vpc-id>',
subnetIds: ['<subnet-1>', '<subnet-2>'],
auroraSecurityGroupId: '<aurora-security-group-id>',
// Aurora connection details
auroraEndpoint: '<aurora-cluster-endpoint>',
auroraPort: '5432',
auroraDbName: '<database-name>',
auroraUser: '<db-user>',
auroraSecretArn: '<secrets-manager-arn>',
// Debezium - use the ARNs from Step 2
debeziumPluginArn: '<customPluginArn-from-step-2>',
debeziumWorkerConfigArn: '<workerConfigurationArn-from-step-2>',
debeziumPluginBucket: '<your-plugin-bucket-name>',
debeziumTopicPrefix: 'aurora.cdc',
debeziumTables: 'public.orders,public.products',
// S3 Tables - the table bucket name must be globally unique
s3TablesBucketName: '<your-table-bucket-name>',
s3TablesNamespace: 'aurora_cdc',
tables: ['orders', 'products'],
tableKeys: { orders: 'order_id', products: 'product_id' },
// Firehose - general purpose S3 bucket for failed record backup
firehoseBackupBucket: '<your-backup-bucket-name>',
};

Key configuration notes:

auroraSecurityGroupId. The security group attached to your Aurora cluster. The CDK creates an MSK security group with ingress rules allowing traffic from this security group, and a reverse rule allowing MSK Connect workers to reach Aurora on port 5432.
tableKeys. The primary key column for each table. Firehose uses these to match incoming records against existing rows for update and delete operations in the Iceberg tables.
s3TablesBucketName. The name for your S3 table bucket. Table bucket names must be unique for your account in the chosen Region.

Step 4: Deploy the CDK stacks

Deploy all six stacks with a single command. The CDK resolves the dependency order automatically:

npx cdk --app "npx ts-node bin/app-v2.ts" deploy --all

When prompted, review the AWS Identity and Access Management (IAM) changes and confirm the deployment. The CDK deploys the following stacks:

Stack	What it creates
`CdcMskCluster`	Amazon MSK cluster (2x kafka.m5.large brokers) with dual authentication (IAM for Firehose, unauthenticated for Debezium), custom configuration with `auto.create.topics.enable=true`, security groups with ingress rules for Aurora and MSK Connect workers
`CdcMskConnectIam`	MSK Connect service execution role with permissions for Kafka cluster operations, VPC networking, S3 plugin access, and AWS Secrets Manager; Amazon CloudWatch Logs group for connector logs
`CdcS3Tables`	S3 table bucket, `aurora_cdc` namespace, two Iceberg tables (`orders`, `products`) with column schemas
`CdcLambdaTransform`	Lambda function for CDC event transformation and multi-table routing
`CdcFirehoseRole`	Firehose IAM role with permissions for Amazon MSK, S3 Tables, AWS Glue Data Catalog, AWS Lake Formation, VPC networking, and Lambda invocation
`CdcFirehose`	Firehose delivery stream with MSK as source (private connectivity through AWS PrivateLink), Lambda processing, Apache Iceberg Tables as destination with two table configurations, and S3 backup bucket for failed records

The MSK cluster takes approximately 25 minutes to create. The Debezium connector takes approximately 5 minutes after the cluster is ready. You can monitor the deployment progress in the AWS CloudFormation console.

After the deployment completes, you can verify the resources in the AWS console. The S3 table bucket shows the two Iceberg tables in the aurora_cdc namespace.

Figure 2. S3 table bucket showing the orders and products Iceberg tables in the aurora_cdc namespace.

The Firehose delivery stream shows the MSK source, Lambda transformation, and Apache Iceberg Tables destination.

Figure 3. Amazon Data Firehose delivery stream with MSK source, Lambda transformation, and Apache Iceberg Tables destination.

The MSK cluster uses dual authentication (IAM for Firehose, unauthenticated for Debezium through TLS_PLAINTEXT), multi-VPC private connectivity for Firehose PrivateLink access, and auto.create.topics.enable=true so Debezium can create topics on first connect. VPC connectivity and the cluster resource policy are configured as CLI steps in Step 5.

Step 5: Enable MSK VPC connectivity, grant Lake Formation permissions, and apply MSK cluster policy

After the CDK deployment completes, enable multi-VPC private connectivity with IAM on the MSK cluster. Firehose requires this to create an AWS PrivateLink endpoint to the MSK brokers. This setting can’t be configured during cluster creation and must be applied as an update, which triggers a rolling broker restart (approximately 20–30 minutes).

# Get the cluster ARN and current version from the CdcMskCluster stack outputs
MSK_ARN=<msk-cluster-arn>
CLUSTER_VERSION=$(aws kafka describe-cluster-v2 \
    --cluster-arn $MSK_ARN \
    --region <your-region> \
    --query 'ClusterInfo.CurrentVersion' --output text)
# Enable VPC connectivity with IAM
aws kafka update-connectivity \
    --cluster-arn $MSK_ARN \
    --current-version $CLUSTER_VERSION \
    --connectivity-info '{"VpcConnectivity":{"ClientAuthentication":{"Sasl":{"Iam":{"Enabled":true}}}}}' \
    --region <your-region>

Wait for the cluster state to return to ACTIVE before proceeding:

aws kafka describe-cluster-v2 \
    --cluster-arn $MSK_ARN \
    --region <your-region> \
    --query 'ClusterInfo.State'

Next, grant the Firehose IAM role permissions through AWS Lake Formation. S3 Tables uses a sub-catalog format for the CatalogId parameter, which differs from the standard AWS Glue Data Catalog. These permissions require a data lake administrator identity.

Grant database-level and table-level permissions to the Firehose role:

# Grant database-level permissions
aws lakeformation grant-permissions \
    --region <your-region> \
    --principal '{"DataLakePrincipalIdentifier": "<firehose-role-arn>"}' \
    --resource '{"Database": {"CatalogId": "<account-id>:s3tablescatalog/<table-bucket-name>", "Name": "aurora_cdc"}}' \
    --permissions '["ALL"]'
# Grant table-level permissions (wildcard for the tables in the namespace)
aws lakeformation grant-permissions \
    --region <your-region> \
    --principal '{"DataLakePrincipalIdentifier": "<firehose-role-arn>"}' \
    --resource '{"Table": {"CatalogId": "<account-id>:s3tablescatalog/<table-bucket-name>", "DatabaseName": "aurora_cdc", "TableWildcard": {}}}' \
    --permissions '["ALL"]'

Note the CatalogId format: <account-id>:s3tablescatalog/<table-bucket-name>. This is specific to S3 Tables and tells Lake Formation to look up permissions in the S3 Tables catalog rather than the default Glue Data Catalog. For more information, see Integrating Amazon S3 Tables with AWS analytics services.

Next, attach a resource-based policy to the MSK cluster that grants the Firehose service principal permission to create VPC connections:

aws kafka put-cluster-policy \
    --cluster-arn <msk-cluster-arn> \
    --region <your-region> \
    --policy '{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "firehose.amazonaws.com"},
"Action": ["kafka:CreateVpcConnection", "kafka:GetBootstrapBrokers", "kafka:DescribeClusterV2"],
"Resource": "<msk-cluster-arn>"
}]
}'

You can find the <msk-cluster-arn> in the CdcMskCluster stack outputs from Step 4, and the <firehose-role-arn> in the CdcFirehoseRole stack outputs.

Step 6: Create the Debezium connector

With the MSK cluster running and Lake Formation permissions in place, create the Debezium connector using the MSK Connect API. The connector reads changes from Aurora PostgreSQL and publishes them to the MSK topic.

Firehose supports only one MSK topic per delivery stream, so each source table would otherwise need its own Firehose stream and VPC connection. To avoid this, the connector uses the Debezium ByLogicalTableRouter Single Message Transform (SMT) to route changes from multiple tables into a single topic (aurora.cdc.all-tables). The Lambda function then uses the source table name in each message to direct records to the correct Iceberg table. This single-topic pattern uses one Firehose stream for multiple tables, reducing cost and operational complexity.

First, retrieve the MSK bootstrap servers from the cluster:

aws kafka get-bootstrap-brokers \
    --cluster-arn <msk-cluster-arn> \
    --region <your-region>

Note the BootstrapBrokerString value (the PLAINTEXT brokers). Then create the connector:

aws kafkaconnect create-connector --cli-input-json '{
"connectorName": "aurora-postgres-debezium-connector",
"kafkaCluster": {
"apacheKafkaCluster": {
"bootstrapServers": "<bootstrap-servers>",
"vpc": {
"subnets": ["<subnet-1>", "<subnet-2>"],
"securityGroups": ["<msk-security-group-id>"]
}
}
},
"kafkaClusterClientAuthentication": {"authenticationType": "NONE"},
"kafkaClusterEncryptionInTransit": {"encryptionType": "PLAINTEXT"},
"kafkaConnectVersion": "2.7.1",
"plugins": [{"customPlugin": {"customPluginArn": "<custom-plugin-arn>", "revision": 1}}],
"serviceExecutionRoleArn": "<msk-connect-service-role-arn>",
"capacity": {"provisionedCapacity": {"mcuCount": 2, "workerCount": 2}},
"workerConfiguration": {"workerConfigurationArn": "<worker-config-arn>", "revision": 1},
"connectorConfiguration": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"database.hostname": "<aurora-cluster-endpoint>",
"database.port": "5432",
"database.user": "<db-user>",
"database.password": "<db-password>",
"database.dbname": "<database-name>",
"database.server.name": "aurora_cdc",
"plugin.name": "pgoutput",
"slot.name": "debezium_slot",
"publication.name": "dbz_publication",
"table.include.list": "public.orders,public.products",
"topic.prefix": "aurora.cdc",
"schema.history.internal.kafka.topic": "schema-changes.aurora",
"schema.history.internal.kafka.bootstrap.servers": "<bootstrap-servers>",
"decimal.handling.mode": "string",
"time.precision.mode": "adaptive_time_microseconds",
"tombstones.on.delete": "false",
"snapshot.mode": "initial",
"publication.autocreate.mode": "filtered",
"transforms": "Reroute",
"transforms.Reroute.type": "io.debezium.transforms.ByLogicalTableRouter",
"transforms.Reroute.topic.regex": "aurora\\\\\\\\.cdc\\\\\\\\.public\\\\\\\\.(.*)",
"transforms.Reroute.topic.replacement": "aurora.cdc.all-tables"
},
"logDelivery": {
"workerLogDelivery": {
"cloudWatchLogs": {
"enabled": true,
"logGroup": "/aws/msk-connect/aurora-cdc-debezium"
}
}
}
}'

The <msk-security-group-id> and <msk-connect-service-role-arn> can be found in the CdcMskCluster and CdcMskConnectIam stack outputs respectively. The ByLogicalTableRouter Single Message Transform routes CDC events from the monitored tables into a single topic (aurora.cdc.all-tables).

Step 7: Verify the Debezium connector

After creating the connector, verify that it is running and has completed its initial snapshot.

aws kafkaconnect list-connectors --region <your-region> \
    --query 'connectors[?connectorName==`aurora-postgres-debezium-connector`].{Name:connectorName,State:connectorState}' \
    --output table

The connector state should show RUNNING, as shown in the following figure.

Figure 4. Debezium connector running on Amazon MSK Connect.

Check the CloudWatch Logs to confirm the snapshot completed:

aws logs tail /aws/msk-connect/aurora-cdc-debezium --follow --region <your-region>

You should see messages indicating the transition to streaming mode:

Finished exporting 0 records for table 'public.orders' (1 of 2 tables)
Finished exporting 0 records for table 'public.products' (2 of 2 tables)
Snapshot completed
Starting streaming

If the tables were empty when the connector started, the export count is 0. If you had existing data, the snapshot captures the existing rows as r (read) operations, which the Lambda function maps to insert operations in the Iceberg tables.

Verify that the Firehose delivery stream is active:

aws firehose describe-delivery-stream \
    --delivery-stream-name msk-to-s3tables-firehose \
    --region <your-region> \
    --query 'DeliveryStreamDescription.DeliveryStreamStatus'

The status should return ACTIVE.

Step 8: Test the pipeline

Insert test data into the Aurora PostgreSQL source tables. Each insert triggers a CDC event that flows through the pipeline: Aurora WAL to Debezium to MSK topic to Firehose to Lambda transform to S3 Tables.

-- Insert orders
INSERT INTO public.orders (customer_id, order_date, total_amount, status)
VALUES
(1, '2026-01-20', 299.99, 'shipped'),
(2, '2026-01-21', 149.50, 'processing'),
(1, '2026-01-22', 89.99, 'delivered');
-- Insert products
INSERT INTO public.products (product_name, category, price, stock_quantity)
VALUES
('Wireless Headphones', 'Electronics', 79.99, 150),
('Running Shoes', 'Sports', 129.99, 75),
('Coffee Maker', 'Kitchen', 49.99, 200);

This creates six records across two tables. Each record generates a Debezium CDC event with operation type c (create), which the Lambda function maps to an insert operation in the corresponding Iceberg table.

Step 9: Verify data delivery

Check the Firehose IncomingRecords metric to confirm records are flowing through the delivery stream:

aws cloudwatch get-metric-statistics \
    --namespace AWS/Firehose \
    --metric-name IncomingRecords \
    --dimensions Name=DeliveryStreamName,Value=msk-to-s3tables-firehose \
    --start-time $(date -u -v-10M +%Y-%m-%dT%H:%M:%S) \
    --end-time $(date -u +%Y-%m-%dT%H:%M:%S) \
    --period 60 --statistics Sum \
    --region <your-region>

You should see a Sum value of 6 or more. If the value is 0, wait another minute and retry. There can be a short delay between MSK topic delivery and Firehose metric reporting.

If records aren’t appearing, check the Firehose error output in the backup S3 bucket and the Lambda function’s CloudWatch Logs for transformation errors.

Step 10: Query data using Amazon Athena

With data delivered to S3 Tables, you can query the Iceberg tables using Amazon Athena. S3 Tables integrates with the AWS Glue Data Catalog as a sub-catalog, so you reference tables using the S3 Tables catalog format.

Tip: If records aren’t appearing in Athena, check the Firehose IncomingRecords CloudWatch metric and the Lambda function’s CloudWatch Logs for transformation errors.

Open the Athena console, select the AwsDataCatalog data source, and run the following queries:

SELECT * FROM "s3tablescatalog/<table-bucket-name>"."aurora_cdc"."products" LIMIT 10;
SELECT * FROM "s3tablescatalog/<table-bucket-name>"."aurora_cdc"."orders" LIMIT 10;

Replace <table-bucket-name> with your S3 table bucket name. You should see the records from the initial snapshot that Debezium captured when the connector started.

The following figures show the initial state of both tables as queried through Athena. At this point, the products table contains seven records and the orders table contains seven records, captured during the Debezium initial snapshot.

Figure 5. Initial state of the products table in Amazon Athena, showing seven records captured from Aurora PostgreSQL through the CDC pipeline.

Figure 6. Initial state of the orders table in Amazon Athena, showing seven records captured from Aurora PostgreSQL through the CDC pipeline.

Now test that update and delete operations propagate correctly. Run the following statements in Aurora:

-- Insert new records
INSERT INTO public.products (product_name, category, price, stock_quantity)
VALUES ('Bluetooth Speaker', 'Electronics', 129.99, 90), ('Standing Desk', 'Furniture', 799.99, 20);
INSERT INTO public.orders (customer_id, order_date, total_amount, status)
VALUES (201, '2026-04-03', 149.99, 'NEW'), (202, '2026-04-03', 249.50, 'NEW'), (203, '2026-04-03', 79.90, 'NEW');
-- Update existing records
UPDATE public.products SET stock_quantity = 30, price = 549.99 WHERE product_name = 'Ergonomic Chair';
UPDATE public.orders SET status = 'DELIVERED' WHERE order_id = 201;
-- Delete a record
DELETE FROM public.products WHERE product_name = 'Test Widget';

Wait for the changes to propagate through the pipeline, then query Athena again. The following figures show the results after the insert, update, and delete operations have been applied.

In the products table, the Test Widget record (product_id 100) is no longer present because it was removed by the delete operation. The Ergonomic Chair row now reflects the updated price (549.99) and stock quantity (30). Two new records, Bluetooth Speaker and Standing Desk, appear with a later created_at timestamp, confirming they were inserted after the initial snapshot.

Figure 7. Products table after CDC operations. The Ergonomic Chair, Headphones, and Desk Lamp rows reflect updated values. Bluetooth Speaker and Standing Desk are newly inserted records. The Test Widget record has been removed by the delete operation.

In the orders table, order 100 now shows a status of SHIPPED and order 201 shows DELIVERED, reflecting the update operations. Three new orders (301, 302, 303) appear with status NEW and a later timestamp, confirming they were inserted after the initial load.

Figure 8. Orders table after CDC operations. Orders 100 and 201 reflect updated status values. Orders 301, 302, and 303 are newly inserted records.

This confirms that the pipeline correctly handles the three CDC operation types: inserts, updates, and deletes are captured from the Aurora WAL by Debezium, routed through the single MSK topic, transformed by the Lambda function, and applied as row-level Iceberg operations by Firehose.

S3 Tables handles compaction and snapshot management for Iceberg tables automatically, including compaction of small data files and expiration of old snapshots. You don’t need to run manual maintenance operations.

You can also use Iceberg’s time travel capability to query the table as it existed before the updates:

SELECT * FROM "s3tablescatalog/<table-bucket-name>"."aurora_cdc"."orders"
FOR TIMESTAMP AS OF current_timestamp - interval '5' minute;

This returns the original data before the update, demonstrating the time travel capability that Apache Iceberg provides through S3 Tables.

Cleaning up

To avoid ongoing charges, delete the resources in reverse dependency order.

Delete the CDK stacks:

cd cdk
npx cdk --app "npx ts-node bin/app-v2.ts" destroy --all

Delete the Debezium custom plugin and worker configuration that were created through the AWS CLI in Step 2:

aws kafkaconnect delete-custom-plugin --custom-plugin-arn <plugin-arn>
aws kafkaconnect delete-worker-configuration --worker-configuration-arn <worker-config-arn>

Clean up the Aurora PostgreSQL replication resources:

SELECT pg_drop_replication_slot('debezium_slot');
DROP PUBLICATION dbz_publication;

Important: The replication slot (debezium_slot) was created automatically by Debezium. If you plan to redeploy the pipeline later, you don’t need to drop the slot and publication. However, the replication slot continues to retain WAL segments while the connector isn’t running, which can increase storage usage on the Aurora cluster. The MSK cluster is the largest cost component of this solution and can’t be paused. It can only be deleted and recreated.

Conclusion

In this post, we showed you how to build a near real-time CDC pipeline from Aurora PostgreSQL to Apache Iceberg tables in Amazon S3 Tables. The key architectural decisions include:

Single-topic routing with multi-table delivery. The Debezium ByLogicalTableRouter SMT routes CDC events from multiple tables through one MSK topic, and the Lambda otfMetadata routing directs each record to the correct Iceberg table. This reduces VPC connection costs by using a single Firehose stream for inserts, updates, and deletes across multiple destination tables.
Fully managed CDC pipeline. MSK Connect runs Debezium, Firehose handles delivery with automatic retries, and S3 Tables manages Iceberg compaction and snapshots. The Lambda transform preserves CDC semantics by mapping Debezium operations to Iceberg row-level operations.
Governed lakehouse access. Lake Formation controls fine-grained access to the Iceberg tables, and data from multiple isolated Aurora clusters can be unified in a single S3 Tables namespace for cross-domain analytics.
Infrastructure as code. Six AWS CDK stacks deploy the core pipeline, with Lake Formation permissions, MSK cluster policy, and Debezium connector configured through documented CLI steps.

To get started, clone the sample repository and follow the walkthrough steps. For more information about the services used in this solution, see the Amazon MSK Developer Guide, Amazon Data Firehose Developer Guide, and Amazon S3 Tables User Guide.

We encourage you to try this solution and adapt it to your own CDC workloads. If you have questions or feedback, leave a comment on this post.

About the author

Beyond JSON blobs: Implementing the VARIANT data type in Apache Iceberg V3

Arun Shanmugam — Tue, 09 Jun 2026 17:01:16 +0000

Apache Iceberg V3 introduces the VARIANT data type. VARIANT provides data engineers with a high-performance, native solution for managing semi-structured data within the data lake. Consider a massive fleet of IoT sensors: street-level temperature probes, air quality monitors, and vehicle telemetry. Each device emits data in unique JSON structures that constantly evolve with firmware updates.

Historically, engineers were forced to store these payloads as STRING blobs. This legacy approach mandates expensive CPU-intensive parsing at runtime and inflates storage costs with redundant raw text. VARIANT solves these inefficiencies by employing a shredded, binary-encoded format. This allows query engines to skip irrelevant data and access specific nested fields with columnar speed, effectively bridging the gap between the flexibility of JSON and the performance of a structured schema.

VARIANT is stored in Parquet as a three-part group: binary metadata (type and dictionary info), a binary value (the full variant for fallback), and a typed_value group where individual JSON fields are shredded into separate Parquet columns. When you query a specific field, Spark prunes the typed_value group to include only the requested sub-columns. It always retains metadata and the value fallback, so it avoids reading the entire document. This approach delivers two concrete benefits:

Reduced query processing time: Queries access only the fields they need without deserializing entire JSON documents. This reduces the amount of data scanned and the time spent on deserialization.
Lower storage footprint: Binary encoding compresses more efficiently than raw text, reducing storage costs.

Fields inside the JSON become individually accessible columns under the hood. A query that needs one value out of a deeply nested document no longer must read and deserialize the entire thing. You maintain schema flexibility while gaining the performance characteristics of structured columnar storage.

This post is part 1 of a two-part series. We walk through the basics: creating an Iceberg V3 table with a VARIANT column, inserting semi-structured data, and querying it with variant_get(). In Part 2, we scale to millions of rows and benchmark VARIANT against traditional string storage. We measure the difference in query performance and storage footprint.

Solution overview

This walkthrough demonstrates an end-to-end workflow for working with semi-structured data using the VARIANT data type in Apache Iceberg V3 on Amazon EMR Serverless. Raw JSON payloads are ingested and converted to binary VARIANT format using parse_json(). The data is stored in an Iceberg V3 table where the engine shreds the structure into columnar Parquet sub-columns. You can then query the data efficiently using variant_get() to extract specific fields without deserializing the entire document. AWS Glue Data Catalog manages the table metadata. Amazon Simple Storage Service (Amazon S3) provides the underlying storage.

Note: Check the Apache Iceberg documentation for the latest information on specification status and engine compatibility. Additionally, Fine-Grained Access Control (FGAC) through AWS Lake Formation is not currently supported for the VARIANT data type.

How VARIANT works

When you insert a JSON document into a VARIANT column, Spark converts it from a JSON string into the Variant binary format. During writes, the engine can shred the structure. It extracts individual fields and stores them as native Parquet-typed sub-columns within the VARIANT column’s typed_value group. Fields that are not shredded remain in the binary value column as a fallback. This is conceptually similar to how a columnar table stores each column independently. The difference is that the sub-columns live within a single VARIANT column, and the engine handles the shredding schema automatically.

At query time, when you ask for a specific field using variant_get(), Spark reads only the sub-column that contains that field. It does not need to load or parse the rest of the document. For workloads that repeatedly query a handful of fields out of large, complex JSON payloads, this can significantly reduce the amount of data scanned. It also reduces the time spent deserializing it.

The variant_get() function uses JSON path syntax to navigate the structure. You can extract scalar values with an explicit type (optional), access nested objects, and reach into arrays by index. The function signature is the following.

variant_get(column, '$.path.to.field', 'type')

Where column is the VARIANT column name, the second argument is a JSON path expression, and the optional third argument specifies the expected return type (such as 'string', 'int', or 'double'). When the type argument is omitted, the function returns a VARIANT value that preserves the original encoding.

Running Iceberg V3 on Amazon EMR Serverless

Amazon EMR Serverless 8.0 ships with Apache Spark 4.0.1, which includes native support for Iceberg V3 and the VARIANT data type. You do not need to install additional libraries or configure custom JARs. Amazon EMR Serverless manages the compute infrastructure and scales resources up and down based on workload demand. You can focus on the data rather than the cluster.

While this post uses Amazon EMR Serverless, Iceberg V3 VARIANT support is also available on Amazon EMR on EC2 and Amazon EMR on EKS. You can choose the deployment model that fits your environment.

Getting started

The following walkthrough creates an Iceberg V3 table with a VARIANT column, inserts a set of IoT sensor events, and runs queries to extract fields from the semi-structured payload. Each step includes the code you need to run it on Amazon EMR Serverless.

Prerequisites

Before you begin, verify you have the following:

An AWS account with permissions to create Amazon EMR Serverless applications and access Amazon Simple Storage Service (Amazon S3).
An Amazon S3 bucket for storing Iceberg table data and scripts.
AWS Glue Data Catalog configured for metadata management.
An IAM execution role with permissions for Amazon EMR Serverless, Amazon S3, AWS Glue, and Amazon CloudWatch Logs.
AWS Command Line Interface (AWS CLI) installed and configured.Note: Running this solution in your AWS account might incur charges for Amazon EMR Serverless, Amazon S3, and AWS Glue. Refer to the respective pricing pages for cost details.

Step 1: Initialize a Spark session with Iceberg V3

Start by creating a Spark session configured to use the Iceberg catalog backed by AWS Glue. The key settings are the Iceberg Spark extensions and the AWS Glue catalog implementation. Replace <YOUR_S3_BUCKET> with your bucket name.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, parse_json

spark = SparkSession.builder \
    .appName("IcebergV3VariantDemo") \
    .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.warehouse",
            "s3://<YOUR_S3_BUCKET>/warehouse/") \
    .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") \
    .getOrCreate()

When running on Amazon EMR Serverless, some Spark configurations might be set at the application or job level. The configuration shown here is included in the script for completeness. Depending on your Amazon EMR Serverless application settings, you might not need to specify all these properties in the script.

Step 2: Create an Iceberg V3 table with a VARIANT column

Create a namespace and table. The format version must be set to 3 for VARIANT data type support. The following table models IoT sensor events with a few standard columns and a VARIANT column for the semi-structured payload.

spark.sql("CREATE NAMESPACE IF NOT EXISTS glue_catalog.iceberg_v3_demo")

spark.sql("""
CREATE TABLE IF NOT EXISTS glue_catalog.iceberg_v3_demo.sensor_events (
    event_id STRING,
    device_id STRING,
    event_timestamp TIMESTAMP,
    event_data VARIANT
)
USING iceberg
TBLPROPERTIES (
    'format-version' = '3'
)
""")

The event_data column is declared as VARIANT. Iceberg stores it in Parquet as a binary-encoded VARIANT structure (metadata, value, and optional shredded sub-columns) rather than as a plain text string.

Step 3: Insert semi-structured data

To insert JSON data into a VARIANT column, use the parse_json() function. This converts a JSON string into the binary VARIANT format at write time. The following example creates a small DataFrame of IoT events and appends them to the table.

import json
from pyspark.sql.functions import current_timestamp
from pyspark.sql.types import StructType, StructField, StringType

# Sample IoT events with nested JSON payloads
events = [
    ("evt_001", "sensor_001", json.dumps({
        "device": {"manufacturer": "SensorTech", "model": "ST-200",
                   "firmware_version": "3.1.4"},
        "sensors": {"temperature": 22.5, "humidity": 61.3,
                    "air_quality": {"pm25": 12.4, "co2": 415}},
        "network": {"connection": "WiFi", "latency_ms": 42},
        "alerts": [{"severity": "low", "message": "Calibration due"}]
    })),
    ("evt_002", "sensor_002", json.dumps({
        "device": {"manufacturer": "IoTCorp", "model": "IC-500",
                   "firmware_version": "2.8.1"},
        "sensors": {"temperature": 34.1, "humidity": 78.9,
                    "air_quality": {"pm25": 142.7, "co2": 1850}},
        "network": {"connection": "LTE", "latency_ms": 210},
        "alerts": [{"severity": "critical",
                    "message": "Temperature threshold exceeded"},
                   {"severity": "high",
                    "message": "Poor air quality detected"}]
    })),
    ("evt_003", "sensor_003", json.dumps({
        "device": {"manufacturer": "SmartDevices", "model": "SD-100",
                   "firmware_version": "1.5.9"},
        "sensors": {"temperature": 18.7, "humidity": 45.2,
                    "air_quality": {"pm25": 8.1, "co2": 390}},
        "network": {"connection": "Ethernet", "latency_ms": 5},
        "alerts": []
    })),
]

schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("device_id", StringType(), False),
    StructField("event_data", StringType(), False),
])

df = spark.createDataFrame(events, schema)
df = df.withColumn("event_timestamp", current_timestamp())

# Convert JSON string to VARIANT using parse_json
df = df.withColumn("event_data", parse_json(col("event_data")))

df.writeTo("glue_catalog.iceberg_v3_demo.sensor_events").append()
print("Data inserted successfully.")

The parse_json() call is the key step. It takes the raw JSON string and encodes it into the binary VARIANT format before writing to the Iceberg table.

Step 4: Query VARIANT data with variant_get()

Once the data is in the table, you can extract individual fields from the VARIANT column using variant_get(). The following queries demonstrate three common patterns: simple field extraction, deep nested access with filtering, and array element access.

The following queries are shown as raw SQL for readability. To run them in your PySpark script, wrap each query in a spark.sql() call. For example: spark.sql("SELECT ...").show().

Query 1: Simple field extraction

Extract top-level sensor readings from the payload.

SELECT
    event_id,
    device_id,
    variant_get(event_data, '$.sensors.temperature', 'double') AS temperature,
    variant_get(event_data, '$.sensors.humidity', 'double') AS humidity
FROM glue_catalog.iceberg_v3_demo.sensor_events

This query reads only the temperature and humidity sub-columns from the VARIANT data. It does not parse or load the rest of the JSON document.

Query 2: Deep nested access with filtering

Reach into nested objects and filter on a value buried inside the structure.

SELECT
    device_id,
    variant_get(event_data, '$.sensors.air_quality.pm25', 'double') AS pm25,
    variant_get(event_data, '$.sensors.air_quality.co2', 'int') AS co2_level,
    variant_get(event_data, '$.device.manufacturer', 'string') AS manufacturer
FROM glue_catalog.iceberg_v3_demo.sensor_events
WHERE variant_get(event_data, '$.sensors.air_quality.pm25', 'double') > 100.0

The WHERE clause filters directly on a nested VARIANT field. Spark evaluates the predicate against the shredded sub-column without deserializing the full payload.

Query 3: Array element access

Access elements inside a JSON array stored within the VARIANT column.

SELECT
    event_id,
    device_id,
    variant_get(event_data, '$.alerts[0].severity', 'string') AS first_alert_severity,
    variant_get(event_data, '$.alerts[0].message', 'string') AS first_alert_message
FROM glue_catalog.iceberg_v3_demo.sensor_events
WHERE variant_get(event_data, '$.alerts[0].severity', 'string') = 'critical'

Array indexing uses standard bracket notation in the JSON path. This query finds events where the first alert has critical severity and returns the alert details.

Figure 1: Query results showing simple field extraction, nested access with filtering, and array element access from the VARIANT column.

Submitting the job to Amazon EMR Serverless

To run this on Amazon EMR Serverless, save the preceding code as a single PySpark script (for example, iceberg_v3_variant_demo.py), upload it to Amazon S3, and submit it as a job. Replace the placeholder values with your own.

Before submitting the job, make sure you have created an Amazon EMR Serverless application. For instructions, see Getting started with Amazon EMR Serverless in the Amazon EMR documentation.

# Upload script to S3
aws s3 cp iceberg_v3_variant_demo.py \
    s3://<YOUR_S3_BUCKET>/scripts/ \
    --region <REGION>

# Submit the job
aws emr-serverless start-job-run \
    --application-id <APPLICATION_ID> \
    --execution-role-arn arn:aws:iam::<ACCOUNT_ID>:role/EMRServerlessExecutionRole \
    --job-driver '{
        "sparkSubmit": {
            "entryPoint": "s3://<YOUR_S3_BUCKET>/scripts/iceberg_v3_variant_demo.py"
        }
    }' \
    --configuration-overrides '{
        "monitoringConfiguration": {
            "cloudWatchLoggingConfiguration": {
                "enabled": true,
                "logGroupName": "/aws/emr-serverless/applications/<APPLICATION_ID>"
            }
        }
    }' \
    --region <REGION>

Use cases

VARIANT fits naturally into workloads where the data is semi-structured and the schema is not fully known in advance. Some use cases include the following:

IoT and sensor data: Device fleets produce telemetry in varying JSON formats that evolve with firmware updates. VARIANT stores these payloads without requiring a fixed schema, and queries can extract specific readings without scanning the entire document.
Clickstream analytics: User behavior events on websites and mobile apps carry different attributes depending on the action. Page views, clicks, form submissions, and purchases each have their own structure. VARIANT accommodates these data types in a single column.
Log analytics: Application logs, infrastructure metrics, and audit trails often arrive as unstructured or loosely structured JSON. VARIANT lets you ingest them as is and query specific fields on demand, without defining a schema up front.

Clean up

To avoid ongoing charges, delete the resources you created:

Drop the Iceberg table and namespace using Spark SQL.

spark.sql("DROP TABLE IF EXISTS glue_catalog.iceberg_v3_demo.sensor_events")
spark.sql("DROP NAMESPACE IF EXISTS glue_catalog.iceberg_v3_demo")

Stop and delete the Amazon EMR Serverless application.

aws emr-serverless delete-application --application-id <APPLICATION_ID> --region <REGION>

Delete the S3 objects and bucket used for table data, scripts, and logs.

aws s3 rm s3://<YOUR_S3_BUCKET>/warehouse/ --recursive
aws s3 rm s3://<YOUR_S3_BUCKET>/scripts/ --recursive

Conclusion

Apache Iceberg V3’s VARIANT type provides an efficient way to store and query semi-structured data in your data lake. Columnar storage and shredding reduce storage costs, and direct field access through variant_get() removes the need to parse JSON strings at query time. On Amazon EMR Serverless, you get this capability without managing infrastructure.

In Part 2 of this series, we scale to millions of rows and benchmark VARIANT against traditional string storage. We measure query performance and storage footprint under realistic workloads.

To learn more about Apache Iceberg on AWS, see Apache Iceberg on AWS prescriptive guidance. For more information about Amazon EMR Serverless, see the Amazon EMR Serverless documentation.

About the authors

Upgrade PySpark from Spark 3.5 to Spark 4.0 with AWS Spark Upgrade Agent

Prasad Nadig — Tue, 09 Jun 2026 16:17:23 +0000

Upgrading Apache Spark applications across major versions means tracking down breaking changes, manually debugging failures from log files, and running repeated test cycles. This process can stretch across weeks for complex code bases.

In this post, we walk through a hands-on PySpark migration from Spark 3.5 to Spark 4.0 on Amazon EMR Serverless, using the AWS Spark Upgrade Agent. You’ll see how the agent iteratively validates your application on a live Amazon EMR Serverless application, automatically diagnosing and resolving failures from Amazon CloudWatch logs until the job succeeds. By the end, you have a multi-pipeline PySpark application running on Spark 4.0 with four distinct breaking changes resolved. The fixes include configuration key removals, codec renames, and stricter charset validation, all driven through natural language interaction in the Integrated Development Environment (IDE).

This is part 2 of a three-part series on how the AWS Spark Upgrade Agent can automate and simplify Spark upgrades.

In Part 1, we introduced the agent’s architecture and capabilities. This post walks through a complete PySpark migration from Spark 3.5 to Spark 4.0 on Amazon EMR Serverless.

In the sections that follow, you will set up the prerequisites and infrastructure, explore the sample application, run the iterative validation workflow on EMR Serverless, review data quality results, and generate a comprehensive upgrade summary.

Note: Because this upgrade is performed using the AWS Spark Upgrade Agent Model Context Protocol (MCP) server, an agentic artificial intelligence (AI) system, the agent might take different paths to reach the same successful outcome. The workflow demonstrated here represents one successful upgrade path. The key takeaway is the end-to-end workflow: generating an upgrade plan, iteratively validating on Amazon EMR Serverless, and producing a comprehensive upgrade summary.

1. Prerequisites and setup

This section covers the tools, infrastructure, and IDE configuration you need before starting the upgrade. To follow along, you need an AWS account with an AWS Identity and Access Management (AWS IAM) user or role that has permissions to deploy AWS CloudFormation stacks, create AWS IAM roles and policies, and create Amazon EMR Serverless applications. Intermediate knowledge of AWS Command Line Interface (AWS CLI), AWS CloudFormation, and Python is helpful.

1.1 Install Kiro CLI and local tools

In this post, we use Kiro CLI to demonstrate the upgrade workflow. You can use an MCP-compatible IDE or framework. Examples include VS Code with Cline, Cursor, Windsurf, and Claude Desktop, among others. To follow along with Kiro CLI, install it on your workstation. For more details on the installation and setup, refer to Setup for Upgrade Agent:

curl -fsSL https://cli.kiro.dev/install | bash

Run the following command and use your builder ID to log in:

kiro-cli login --use-device-flow

With the Kiro CLI installed and logged in, rather than installing the remaining tools manually, use Kiro CLI to set up and verify your prerequisites with the following prompt:

kiro-cli chat
> Install AWS CLI, Python 3.10, and uv on my system if they are not already installed

Output of AWS CLI and local tools install step.

These tools are needed for the upgrade workflow:

AWS CLI: Configured with a profile that has permissions to assume the AWS Identity and Access Management (AWS IAM) role created following.
Python 3.10+: Required to match the EMR 8.0 runtime.
uv package manager: Required by the MCP Proxy for AWS.

1.2 Infrastructure setup (AWS CloudFormation)

Two AWS CloudFormation stacks create the required resources: an AWS IAM role, an Amazon Simple Storage Service (Amazon S3) staging bucket, an Amazon EMR Serverless application (Spark 4.0.1), and its execution role.

Stack 1 – AWS IAM role and Amazon S3 staging bucket:

The spark-upgrade-mcp-setup template creates the AWS IAM role and Amazon S3 staging bucket required by the upgrade agent. For additional Regions, see the full region list.

After deployment, open the AWS CloudFormation Outputs tab, copy the ExportCommand value, and run it in your terminal. This sets SMUS_MCP_REGION, IAM_ROLE, and STAGING_BUCKET_PATH automatically.

Outputs tab of the CloudFormation stack.

# Sets SMUS_MCP_REGION, IAM_ROLE, and STAGING_BUCKET_PATH
export SMUS_MCP_REGION=<YOUR-REGION> && export IAM_ROLE=arn:aws:iam::<YOUR-ACCOUNT-ID>:role/spark-upgrade-role-* && export STAGING_BUCKET_PATH=<YOUR-BUCKET>

Then configure the AWS CLI profile:

aws configure set profile.spark-upgrade-profile.role_arn ${IAM_ROLE}
aws configure set profile.spark-upgrade-profile.source_profile default
aws configure set profile.spark-upgrade-profile.region ${SMUS_MCP_REGION}

Stack 2 – Amazon EMR Serverless target application and execution role:

git clone https://github.com/aws-samples/sample-amazon-emr-spark4-examples
cd sample-amazon-emr-spark4-examples/pyspark/AWSSpark4AutoUpgradeDemo

The PySpark sample lives at resources/global_logistics_platform/. The AWS CloudFormation template lives at resources/cloudformation/.

Deploy the AWS CloudFormation template to create the source and target Amazon EMR Serverless applications and a shared execution role:

aws cloudformation deploy \
  --template-file resources/cloudformation/emr-serverless-target-setup.yaml \
  --stack-name spark-emr-serverless-upgrade \
  --region ${SMUS_MCP_REGION} \
  --capabilities CAPABILITY_NAMED_IAM \
  --parameter-overrides \
    StagingBucketName=${STAGING_BUCKET_PATH} \
    SourceReleaseLabel=emr-7.0.0 \
    TargetReleaseLabel=emr-spark-8.0-preview \
    SourceApplicationName=spark-upgrade-source \
    TargetApplicationName=spark-upgrade-target

This creates two Amazon EMR Serverless applications: a source (Spark 3.5.0) for data quality baseline and a target (Spark 4.0.1) for upgrade validation, with a shared execution role. Both applications auto-stop after 15 minutes of idle time, so there is no cost when not in use. To upgrade between different Spark versions, override SourceReleaseLabel and TargetReleaseLabel with your target Amazon EMR release labels.

After the stack completes deployment, note the outputs:

aws cloudformation describe-stacks \
  --stack-name spark-emr-serverless-upgrade \
  --region ${SMUS_MCP_REGION} \
  --query "Stacks[0].Outputs" --output table

This gives you the SourceApplicationId, TargetApplicationId, and ExecutionRoleArn needed for the upgrade prompt. Make a note of them.

1.3 IDE and MCP server configuration

Configure the spark-upgrade MCP server. For Kiro CLI:

kiro-cli-chat mcp add \
    --name "spark-upgrade" \
    --command "uvx" \
    --args '[
      "mcp-proxy-for-aws@latest",
      "https://sagemaker-unified-studio-mcp.'${SMUS_MCP_REGION}'.api.aws/spark-upgrade/mcp",
      "--service", "sagemaker-unified-studio-mcp",
      "--profile", "spark-upgrade-profile",
      "--region", "'${SMUS_MCP_REGION}'",
      "--read-timeout", "180"
    ]' \
    --timeout 180000 \
    --scope global

For other MCP clients, refer to your IDE’s MCP configuration documentation and use the same server parameters shown previously.

Verify the connection: Start Kiro CLI and confirm the spark-upgrade tools are loaded:

$ kiro-cli chat
...
spark-upgrade (MCP):
- generate_spark_upgrade_plan          * not trusted
- update_build_configuration           * not trusted
- fix_upgrade_failure                  * not trusted
- run_validation_job                   * not trusted
- check_job_status                     * not trusted
...

Tip: After Kiro CLI and the MCP server are configured, you can ask the agent to verify your setup. For example: “Check if I have AWS CLI, Python 3.10+, and uv installed, and confirm the spark-upgrade MCP server is connected.”

Output showing the status of each tool, AWS CLI, and MCP server.

Tip: Trust mode vs. confirm mode: When running the upgrade agent in Kiro CLI, you have two options:

Trust mode: Type t when prompted to approve a tool. The agent auto-approves subsequent uses of that tool without asking for confirmation. You can also use /tools trust-all to trust every tool at once for a fully autonomous experience.

Confirm mode: Type y for each individual tool invocation. This lets you review, verify, and approve every action before the agent runs it. If this is your first time using the agent, use confirm mode for full visibility.

2. Hands-on PySpark upgrade from Spark 3.5 to Spark 4.0

This section walks through the complete migration of a representative PySpark application from Amazon EMR Serverless 7.0.0 (Spark 3.5.0) to EMR Serverless with the emr-spark-8.0-preview release label (Spark 4.0.1), using the global_logistics_platform sample.

2.1 Sample project: global logistics platform

The sample application is a multi-domain PySpark data processing application with three pipelines:

Fleet management: Processes vehicle telemetry data (GPS tracking, fuel consumption, driver behavior scoring) using window functions, lag/lead operations, and statistical aggregations. Writes Parquet with lz4raw compression.
International shipping: Handles cross-border shipment documents with multi-language address standardization using character encoding functions (encode/decode with charsets like Shift_JIS, GB2312, EUC-KR), and processes carrier manifests with ISO-8859-1 encoding.
Historical compliance: Processes regulatory audit records spanning centuries (including pre-1582 Julian calendar dates), requiring legacy datetime rebasing for Parquet writes.

Project structure:

global_logistics_platform/
├── main.py                          # Orchestrator - runs all 3 pipelines
├── src/
│   ├── utils/
│   │   └── spark_config.py          # Spark session config & logging
│   └── domain/                      # Application code that needs migration
│       ├── fleet_management/
│       │   └── telemetry_processor.py
│       ├── international_shipping/
│       │   └── shipment_processor.py
│       └── historical_compliance/
│           └── compliance_processor.py
└── data/                             # Sample dataset for the workflow
    └── sample/
        ├── fleet_telemetry.csv
        ├── international_shipments.csv
        └── compliance_records.csv

2.2 The four Spark 4.0 incompatibilities

Before diving into the upgrade, here are the four specific breaking changes present in this code base that the agent discovers and resolves entirely through runtime validation:

#	Incompatibility	File(s)
1	Legacy Parquet configuration key removed: `spark.sql.legacy.parquet.datetimeRebaseModeInWrite` removed in Spark 4.0. Must use `spark.sql.parquet.datetimeRebaseModeInWrite`.	`spark_config.py`
2	Parquet compression codec rename: `lz4raw` codec renamed to `lz4_raw` in Spark 4.0.	`telemetry_processor.py`
3	Stricter charset encoding validation: Spark 4.0 tightened `encode()` behavior. Encoding CJK (Chinese, Japanese, Korean) characters to `ISO-8859-1` now throws `MALFORMED_CHARACTER_CODING`. In Spark 3.x this silently replaced unmappable chars with `?`. Restored via `spark.sql.legacy.codingErrorAction`.	`spark_config.py`
4	Character encoding restrictions: `encode()`/`decode()` in Spark 4.0 supports US-ASCII, ISO-8859-1, UTF-8, UTF-16BE, UTF-16LE, UTF-16, and UTF-32. Code uses `Shift_JIS`, `GB2312`, `EUC-KR`.	`shipment_processor.py`

The agent resolves each of these through iterative runtime validation on EMR Serverless: submitting the job, diagnosing failures from Amazon CloudWatch logs, applying fixes, and resubmitting until the job succeeds.

2.3 Step 1: Invoke the upgrade agent

Open the project in Kiro CLI and enter the following prompt:

Upgrade my Spark application in the current directory from EMR serverless version 7.0.0 to EMR serverless version 8.0.0.
Use Amazon EMR Serverless target app-id <YOUR-TARGET-APP-ID> and execution role
<YOUR-EXECUTION-ROLE-ARN> for validation.
Use source Amazon EMR Serverless app-id <YOUR-SOURCE-APP-ID> for data quality baseline.
Store artifacts at s3://${STAGING_BUCKET_PATH}/spark4-upgrade/python/
Enable data quality validation

Tip: The SourceApplicationId, TargetApplicationId, and ExecutionRoleArn are in the Outputs of the spark-emr-serverless-upgrade AWS CloudFormation stack you deployed in Section 1.2.

The agent invokes generate_spark_upgrade_plan, scans the project structure, identifies the Spark version mapping (EMR 7.0.0 → Spark 3.5.0, EMR 8.0.0 → Spark 4.0.1), and produces a structured upgrade plan with an Analysis ID for traceability.

The agent presents the plan and asks for confirmation. Type y to approve the tool invocation, or t to trust that tool for the rest of the session.

You have an option to save the plan as a local JSON file for future reference or to resume the upgrade at a later point, so go ahead and ask Kiro to save it locally. Provide the AWS CLI profile that you have configured on your system. Use the following prompt to provide these inputs:

Yes I would like to save the plan to a local file and use spark-upgrade-profile

2.4 Step 2: Build and package

The agent validates the Python project compiles successfully, then packages it for Amazon EMR Serverless deployment:

Runs py_compile on each .py file to verify syntax.
Creates src.zip containing the src/ directory (preserving the import structure used by from src.utils import ...).
Uploads src.zip, main.py, and sample input data to the Amazon S3 staging path.

# What the agent does behind the scenes:
zip -r src.zip src/
aws s3 cp main.py s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/source/main.py
aws s3 cp src.zip s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/source/src.zip
aws s3 cp data/sample/ s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/input/ --recursive

No external dependencies (no requirements.txt), so no virtual environment is needed. If your project has external dependencies in a requirements.txt, the agent will package them into a virtual environment archive and include it in the EMR Serverless submission parameters.

2.5 Step 3: Data quality baseline on source application

Before migrating the code, the agent establishes a data quality baseline by running the original (pre-upgrade) code on the source Amazon EMR Serverless application (Spark 3.5.0 / EMR 7.0.0). This captures the expected output that the upgraded application must match.

The agent submits the job to the source application with data quality check enabled:

{
  "executionRoleArn": "arn:aws:iam::<YOUR-ACCOUNT-ID>:role/<YOUR-EXECUTION-ROLE>",
  "jobDriver": {
    "sparkSubmit": {
      "entryPoint": "s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/source/main.py",
      "entryPointArguments": [
        "s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/input/",
        "s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/output/source/"
      ],
      "sparkSubmitParameters": "--py-files s3://<YOUR-BUCKET>/spark4-upgrade/python/<ANALYSIS-ID>/source/src.zip"
    }
  },
  "configurationOverrides": {
    "monitoringConfiguration": {
      "cloudWatchLoggingConfiguration": {
        "enabled": true,
        "logGroupName": "/aws/emr-serverless"
      }
    }
  }
}

The agent monitors the source run via check_job_status until it completes successfully. This baseline output is stored for comparison after the target validation succeeds.

2.6 Step 4: Iterative runtime validation on target application

This is the core of the upgrade. The agent submits the unmodified application to the target Amazon EMR Serverless application (Spark 4.0.1), and every incompatibility is discovered, diagnosed, and fixed through runtime failures. The agent drives the entire fix cycle by submitting to EMR, reading errors from Amazon CloudWatch logs, applying fixes, rebuilding, and resubmitting.

The agent presents the proposed Amazon EMR Serverless job configuration for your review before each submission. Type y to approve.

2.6.1 Fix 1: Legacy Parquet configuration key removed (iteration 1)

The first submission fails immediately at SparkSession initialization:

org.apache.spark.sql.AnalysisException:
The SQL config 'spark.sql.legacy.parquet.datetimeRebaseModeInWrite' was removed
in the version 4.0.0. Use 'spark.sql.parquet.datetimeRebaseModeInWrite' instead.

The Historical Compliance pipeline configures spark.sql.legacy.parquet.datetimeRebaseModeInWrite for handling pre-1582 Julian calendar dates. Spark 4.0 removed the legacy. prefix from this configuration key.

The agent calls fix_upgrade_failure, which identifies the migration rule and recommends the fix:

File: src/utils/spark_config.py

# Before
.config("spark.sql.legacy.parquet.datetimeRebaseModeInWrite", "LEGACY")

# After
.config("spark.sql.parquet.datetimeRebaseModeInWrite", "LEGACY")

After applying the fix, the agent rebuilds src.zip, re-uploads to Amazon S3, and resubmits the job.

2.6.2 Fix 2: Parquet compression codec rename (iteration 2)

The resubmitted job fails with a new error, which confirms progress:

pyspark.errors.exceptions.captured.IllegalArgumentException:
[CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION]
The codec lz4raw is not available.
Available codecs are brotli, uncompressed, lzo, snappy, lz4_raw, none, zstd, lz4, gzip.
SQLSTATE: 56038

The Fleet Management pipeline’s telemetry_processor.py uses lz4raw as the Parquet compression codec. Spark 4.0 renamed this to lz4_raw (with an underscore).

The recommended fix:

File: src/domain/fleet_management/telemetry_processor.py

# Before
.option("compression", "lz4raw")

# After
.option("compression", "lz4_raw")

The agent applies the change, rebuilds, and resubmits.

2.6.3 Fix 3: Stricter charset encoding validation (iteration 3)

The next submission surfaces a different failure:

org.apache.spark.SparkRuntimeException:
[MALFORMED_CHARACTER_CODING]
Invalid value found when performing `encode` with ISO-8859-1
SQLSTATE: 22000

The International Shipping pipeline’s process_carrier_manifests() method uses encode(..., 'ISO-8859-1') on data containing CJK (Chinese, Japanese, Korean) characters. Although ISO-8859-1 is in Spark 4.0’s supported charset list, it is a single-byte encoding that cannot represent CJK characters. In Spark 3.x, the Java charset encoder silently replaced unmappable characters with ?. Spark 4.0 tightened this behavior to throw MALFORMED_CHARACTER_CODING for unmappable characters.

The agent identifies the migration rule and adds a legacy compatibility configuration:

File: src/utils/spark_config.py

# Added to SparkSession builder
.config("spark.sql.legacy.codingErrorAction", "true")

This restores the Spark 3.x behavior where unmappable characters are silently replaced instead of throwing errors.

With the configuration added, the agent rebuilds and resubmits.

2.6.4 Fix 4: Character encoding restrictions (iteration 4)

The fourth submission fails with yet another encoding error:

org.apache.spark.SparkIllegalArgumentException:
[INVALID_PARAMETER_VALUE.CHARSET]
The value of parameter(s) `charset` in `encode` is invalid:
expects one of the iso-8859-1, us-ascii, utf-16, utf-16be, utf-16le, utf-32, utf-8,
but got Shift_JIS. SQLSTATE: 22023

The International Shipping pipeline’s standardize_addresses_with_charset() method uses Shift_JIS, GB2312, and EUC-KR charsets in encode()/decode() calls. Spark 4.0 restricts these functions to seven standard charsets. These regional charsets are not in the supported list.

The agent replaces the unsupported charsets with UTF-8:

File: src/domain/international_shipping/shipment_processor.py

Before (Spark 3.5.0):

df = df.withColumn(
    "shipper_address_normalized",
    when(col("origin_country") == "JP",
         expr("decode(encode(shipper_address, 'Shift_JIS'), 'UTF-8')"))
    .when(col("origin_country") == "CN",
         expr("decode(encode(shipper_address, 'GB2312'), 'UTF-8')"))
    .when(col("origin_country") == "KR",
         expr("decode(encode(shipper_address, 'EUC-KR'), 'UTF-8')"))
    .otherwise(col("shipper_address"))
)

After (Spark 4.0.1):

df = df.withColumn(
    "shipper_address_normalized",
    when(col("origin_country") == "JP",
         expr("decode(encode(shipper_address, 'UTF-8'), 'UTF-8')"))
    .when(col("origin_country") == "CN",
         expr("decode(encode(shipper_address, 'UTF-8'), 'UTF-8')"))
    .when(col("origin_country") == "KR",
         expr("decode(encode(shipper_address, 'UTF-8'), 'UTF-8')"))
    .otherwise(col("shipper_address"))
)

The same transformation is applied to consignee_address_normalized.

The agent rebuilds and resubmits one final time.

2.6.5 Final submission: success

The fifth submission completes successfully:

{"success": true, "message": "EMR SERVERLESS job completed successfully",
"compute_run_id": "<JOB-RUN-ID>", "status": "SUCCESS",
"application_type": "EMR-Serverless"}

The three pipelines (Fleet Management, International Shipping, and Historical Compliance) complete on EMR Serverless with the emr-spark-8.0-preview release label (Spark 4.0.1).

2.7 Summary of the iterative runtime validation

The runtime validation loop is the core value of the upgrade agent. Here’s the complete iteration history:

Each iteration follows the same cycle:

Failures that would normally require manual log analysis, root cause investigation, and code patching are resolved automatically by the agent in this workflow.

3. Data quality validation

With both the source baseline (Section 2.5) and the upgraded target run (Section 2.6) completed successfully, the agent performs data quality validation to verify the migration hasn’t changed your application’s output. This is the key advantage of including the source application in your upgrade prompt: the agent can compare outputs from both Spark versions side by side.

3.1 Data quality comparison

The agent invokes get_data_quality_summary to compare the outputs across four dimensions:

Schema validation: Confirms column names, data types, and column ordering match between source and target outputs.
Row count validation: Verifies no data loss or duplication during migration.
Nullability validation: Detects changes in null handling.
Statistical summary validation: Compares numeric and string column distributions (min, max, mean, count, distinct values).

The agent presents the comparison results:

The preceding image shows the data quality summary.

Three of four checks pass cleanly. The statistical summary validation detects a mismatch in the shipper_address column of the customs_declarations output: the max and min summary values differ between source and target.

3.2 Understanding and resolving the mismatch

This mismatch is a direct consequence of Fix 4 (Section 2.6.4). The original code ran addresses through a Shift_JIS/GB2312/EUC-KR → UTF-8 roundtrip that produced garbled text, because the intermediate regional charset corrupted multi-byte UTF-8 characters. The upgraded code uses UTF-8 → UTF-8, preserving addresses faithfully. The mismatch reflects improved data quality, not a regression.

Schema, row counts, and nullability matched exactly: the difference is limited to string values that were previously garbled. No further action is needed. The upgraded application is production-ready.

Expected behavior: Character encoding migrations might change string values, although they preserve semantic meaning. When data quality validation reports mismatches, trace each one back to a specific code change. If the mismatch is explained by a required migration fix (as here), verify the new behavior is correct and document it. If a mismatch cannot be explained, investigate before promoting to production.

4. Upgrade summary

After the agent completes the entire upgrade workflow, it produces a comprehensive upgrade summary following a structured template. This summary lets you review the job configuration updates, code modifications with diffs and file references, relevant migration rules applied, and data quality validation status.

Here is the summary the agent produced for this upgrade:

Upgrade plan

Compile and build project with current Spark 3.5.0: validated that Python files compile successfully.
Run baseline validation on source EMR Serverless (00g4vhvt1lhtrs09) with Spark 3.5.0: established data quality baseline.
Run target validation on target EMR Serverless (00g4vhvt3np1bj09) with Spark 4.0.1: fixed 4 issues iteratively across 4 validation attempts.
Compare data quality between source and target runs: detected expected mismatch in shipper_address.
Generate and persist upgrade summary.

Upgrade result

Upgrade completed with data validation enabled. Data validation detected an expected mismatch in the shipper_address column because of the charset encoding migration from unsupported charsets (Shift_JIS, GB2312, EUC-KR) to UTF-8.

Dependency changes

No external dependencies were changed in this project (no requirements.txt).

Job configuration changes

Parquet datetime rebase configuration key renamed.
- Change: spark.sql.legacy.parquet.datetimeRebaseModeInWrite → spark.sql.parquet.datetimeRebaseModeInWrite.
- Migration rule: In Spark 4.0, the legacy datetime rebasing SQL configurations with the prefix spark.sql.legacy are removed. The SQL configuration spark.sql.legacy.parquet.datetimeRebaseModeInWrite was removed in the version 4.0.0. Use spark.sql.parquet.datetimeRebaseModeInWrite instead.
Legacy coding error action enabled.
- Change: Added spark.sql.legacy.codingErrorAction set to true.
- Migration rule: In Spark 4.0, the encode() and decode() functions raise MALFORMED_CHARACTER_CODING error when handling unmappable characters. In Spark 3.5 and earlier versions, these characters are replaced with garbled text. To restore the previous behavior, set spark.sql.legacy.codingErrorAction to true.

Code changes

Validation attempt 1: Legacy Parquet configuration key.
- Validation run: EMR-Serverless job_run_id 00g4vm14v118vg0b.
- Error: The SQL config 'spark.sql.legacy.parquet.datetimeRebaseModeInWrite' was removed in the version 4.0.0.
- Applied changes: src/utils/spark_config.py: Changed .config("spark.sql.legacy.parquet.datetimeRebaseModeInWrite", "LEGACY") to .config("spark.sql.parquet.datetimeRebaseModeInWrite", "LEGACY").
Validation attempt 2: Parquet compression codec.
- Validation run: EMR-Serverless job_run_id 00g4vm5pm1hig00b.
- Error: [CODEC_NOT_AVAILABLE.WITH_AVAILABLE_CODECS_SUGGESTION] The codec lz4raw is not available.
- Applied changes: src/domain/fleet_management/telemetry_processor.py: Changed .option("compression", "lz4raw") to .option("compression", "lz4_raw").
Validation attempt 3: Stricter charset encoding.
- Validation run: EMR-Serverless job_run_id 00g4vm8sh4sp0g0b.
- Error: [MALFORMED_CHARACTER_CODING] Invalid value found when performing encode with ISO-8859-1.
- Applied changes: src/utils/spark_config.py: Added .config("spark.sql.legacy.codingErrorAction", "true") to the SparkSession builder.
Validation attempt 4: Unsupported charsets.
- Validation run: EMR-Serverless job_run_id 00g4vmc668ng6o0b.
- Error: [INVALID_PARAMETER_VALUE.CHARSET] charset in encode is invalid: expects one of iso-8859-1, us-ascii, utf-16, utf-16be, utf-16le, utf-32, utf-8, but got Shift_JIS.
- Applied changes: src/domain/international_shipping/shipment_processor.py: Replaced Shift_JIS, GB2312, EUC-KR with UTF-8 for shipper and consignee address encoding.

Data validation result

#	Validation	Status
1	Schema validation (column names, types, ordering)	Passed (no difference)
2	Row count validation (no data loss)	Passed (no difference)
3	Nullability validation (null handling changes)	Passed (no difference)
4	Statistical summary validation (numeric/string distributions)	Failed (with difference)

Data mismatch: 1. The shipper_address column max summary value changed in customs_declarations output. This is expected because of the charset encoding migration from Shift_JIS/GB2312/EUC-KR to UTF-8. 2. The shipper_address column min summary value changed in customs_declarations output for the same expected cause.

5. Conclusion

The AWS Spark Upgrade Agent turns a traditionally time-consuming PySpark migration into an automated, iterative workflow. For the Global Logistics Platform sample, the agent identified and resolved four distinct Spark 4.0 breaking changes: legacy Parquet configuration key removal, compression codec renames, stricter charset encoding validation, and character encoding restrictions. Each fix was applied across three domain processors, through natural language interaction in the IDE.

Every incompatibility was discovered through runtime validation on Amazon EMR Serverless. The agent submitted the unmodified application to the target application, and each failure revealed the next breaking change:

The spark.sql.legacy.parquet.datetimeRebaseModeInWrite configuration removal, which crashes SparkSession initialization.
The lz4raw → lz4_raw codec rename, which fails when Parquet writes run.
ISO-8859-1 encoding of CJK characters: ISO-8859-1 is a valid Spark 4.0 charset, so the failure surfaces only when the code runs against real multi-language data, because Spark 4.0 tightened charset encoding validation to reject unmappable characters.
Shift_JIS/GB2312/EUC-KR charsets removed from Spark 4.0’s supported charset list entirely.

The agent diagnosed each error from Amazon CloudWatch logs, applied the fix, rebuilt, and resubmitted without manual intervention beyond approving each step. The data quality validation then confirmed that the upgraded application produces equivalent output on Spark 4.0.1: schema, row counts, and nullability matched exactly. The one difference, in the shipper_address column, resulted from the charset migration from regional encodings to UTF-8, which actually improved data quality by eliminating garbled text from incorrect encoding roundtrips. With each mismatch traced back to a specific, understood code change, the upgraded application is production-ready.

#	Category	Spark 3.x behavior	Spark 4.0 change	Agent fix
1	Parquet datetime configuration	`spark.sql.legacy.parquet.datetimeRebaseModeInWrite`	`legacy.` prefix removed from key name	Update configuration key
2	Parquet compression	`lz4raw` codec name	Renamed to `lz4_raw` (with underscore)	Update codec name
3	Charset + CJK data	`ISO-8859-1` silently replaced unmappable CJK chars with `?`	Stricter charset validation throws `MALFORMED_CHARACTER_CODING` for unmappable characters	Add `spark.sql.legacy.codingErrorAction=true`
4	Character encoding	`encode()`/`decode()` supported Java charsets	Restricted to 7 standard charsets	Replace unsupported charsets with `UTF-8`

Next steps after your first upgrade:

Apply the agent to your production PySpark code base.
Integrate the upgrade workflow into your CI/CD pipeline.
Explore Scala application upgrades (see Part 3 of this series).

To get started with your own PySpark migration:

Deploy the AWS CloudFormation templates from Section 1.2 for one-time AWS IAM, Amazon S3, and Amazon EMR Serverless setup.
Configure the spark-upgrade MCP server in your MCP-compatible IDE.
Point the agent at your PySpark project and let it handle the rest.

For more information, see the Amazon EMR Serverless documentation, the Apache Spark 4.0 migration guide, and the AWS Spark Upgrade Agent setup guide.

6. Clean up resources

To avoid ongoing costs, delete the resources you created:

Delete the Amazon EMR Serverless stack:

aws cloudformation delete-stack --stack-name spark-emr-serverless-upgrade --region ${SMUS_MCP_REGION}

Delete the AWS IAM and Amazon S3 staging stack:

aws cloudformation delete-stack --stack-name spark-upgrade-mcp-setup --region ${SMUS_MCP_REGION}

If the Amazon S3 staging bucket contains objects, empty it before deleting the stack:
```
aws s3 rm s3://${STAGING_BUCKET_PATH} --recursive
```

About the authors

Announcing Spark Connect on Amazon EMR Serverless: Interactive PySpark development, anywhere

Al MS — Tue, 09 Jun 2026 16:15:54 +0000

Today, AWS is announcing support for Spark Connect on Amazon EMR Serverless with EMR release 7.13 (Apache Spark 3.5.6) and later versions. You can now build and debug Spark applications from your preferred local environment while running full-scale Spark operations on EMR Serverless.

Previously, code that worked on a local machine might break in production because of environment mismatches, dependency conflicts, or unexpected data patterns. The only way to catch it was a deploy-and-check cycle. With the Spark Connect feature, you can develop Spark code from a supported local environment, such as an IDE (for example, VS Code or PyCharm), Jupyter notebooks, Amazon SageMaker Unified Studio (SMUS) Data Notebooks, Amazon Q Developer, or Kiro. There are no clusters to provision, no code to repackage, and no deploy-and-check loop. Your local Python session can stay local as usual while Spark operations are automatically routed to a remote Spark server for execution.

Each Spark Connect session has its own AWS resource with a unique ARN, enabling per‑session AWS Identity and Access Management (AWS IAM) permissions, tag‑based cost allocation, audit through AWS CloudTrail, and session-specific configuration overrides. This gives teams finer control over who runs what, at what cost. You also get real-time visibility through the Spark UI, persistent session history, and a dedicated interface to monitor and manage active and completed sessions.

For more details, visit the EMR Serverless release notes or the EMR Serverless Developer Guide. For a quick look at the experience, here’s a demonstration of using Spark Connect in Amazon SageMaker Unified Studio Data Notebooks:

For a runnable end-to-end example, try the EMR Serverless Spark Connect sample notebook from your local IDE. See the following demonstration:

How Spark Connect works

Spark Connect uses a client-server architecture that separates application code from the Spark engine. The client, a lightweight PySpark library running on a local environment, sends Spark operations over a secure gRPC/TLS connection to a Spark Connect server running on EMR Serverless. Then the server runs that Spark code on EMR Serverless as compute. Finally, it returns results to your local session.

Your local machine doesn’t need Spark installed, doesn’t need direct access to the data, and doesn’t need to be sized for the workload. Because the client is a compact library, you can embed Spark operations in your Python applications that support PySpark. This includes web services, dashboards, and automation scripts. For example, a development team can add Spark-powered analytics directly into a FastAPI backend or a Streamlit dashboard, treating Spark like a database driver rather than a separate batch system. These capabilities extend Spark Connect use cases beyond traditional notebook and IDE development, since the compute-intensive processing happens on the server – EMR Serverless side. This allows you to use pandas, matplotlib, and your team’s internal Python libraries on your laptop or in your embedded clients, without installing those libraries on EMR Serverless.

With Spark Connect server sessions running on EMR Serverless, you pay for compute only while your session is active. When inactive, you’re not paying. EMR Serverless automatically scales compute up and down based on workload demands through dynamic resource allocation (DRA), eliminating the need to predict capacity ahead of time. For teams that run Spark Connect sessions regularly, you can configure pre-initialized capacity on your EMR Serverless application for faster session startup times. Additionally, your Spark Connect sessions have access to the full suite of EMR Serverless features, including AWS Graviton processors for cost optimization and secure VPC connectivity to your data sources. You also get access to custom images with flexibility and integrated observability through Amazon CloudWatch and the Spark UI.

Getting started

Getting started with Spark Connect on EMR Serverless takes three steps: create an application, start a session, and connect from your IDE.

Note: The resources created in this quick start incur charges while active. Make sure to follow the cleanup steps at the end of this tutorial to avoid ongoing charges.

Prerequisites

In addition to the required job runtime IAM role, these additional permissions are needed: emr-serverless:StartSession, GetSession, GetSessionEndpoint, TerminateSession, GetResourceDashboard, and iam:PassRole on the runtime role.
An existing EMR Serverless application running emr-7.13.0 or later, with interactiveConfiguration.sessionEnabled = true.
boto3 version 1.43.0 or later to access the latest EMR Serverless session APIs.

Step 1: Create an EMR Serverless application with Spark Connect enabled

Open the Amazon EMR console and navigate to EMR Serverless.
Choose Get started. A pop-up appears. Choose Create and launch EMR Studio.
This takes you to the Create application page.
Enter a Name for your application (for example, spark-connect-app).
For Type, select Spark.
For Release version, select emr-7.13.0 or later.
For Architecture, choose x86_64 (default). This is compatible with most third-party tools and libraries.
Under Application setup options, select Use default settings for interactive workloads. This automatically sets interactiveConfiguration.sessionEnabled = true.
Choose Create and start application.

Alternatively, using the CLI command:

# Create an application with Spark Connect enabled
APP_ID=$(aws emr-serverless create-application \
  --type "SPARK" \
  --name "spark-connect-app" \
  --release-label emr-7.13.0 \
  --interactive-configuration '{"sessionEnabled": true}' \
  --query 'applicationId' \
  --output text)
echo "Created application: $APP_ID"
# Start the application
aws emr-serverless start-application --application-id "$APP_ID"

Step 2: Start a session

Next, start a session and obtain the Spark Connect endpoint.

Provide an IAM execution role that grants the session access to your data, such as reading data from an Amazon S3 bucket or querying the AWS Glue Data Catalog. This is the same type of role used for EMR Serverless batch jobs.

# Start a session with your execution role
$ROLE_ARN="YOUR_ROLE" # example: arn:aws:iam::123456789012:role/EMRServerlessSessionRole
SESSION_ID=$(aws emr-serverless start-session \
  --application-id $APP_ID \
  --execution-role-arn $ROLE_ARN \
  --query sessionId \
  --output text)

# Get the session endpoint
aws emr-serverless get-session-endpoint \
  --application-id $APP_ID \
  --session-id $SESSION_ID

The get-session-endpoint response includes a secure endpoint URL and an authentication token. All communication between your local environment and EMR Serverless is encrypted using TLS. Treat the token as a sensitive credential. Consider using AWS Secrets Manager to store and retrieve tokens programmatically. The authentication token is time-limited to 1 hour, so for long-running sessions we recommend that you refresh it periodically.

Step 3: Connect from your local IDE

Use the returned endpoint URL and authentication token to connect to the Spark Connect server.

The connection URL uses the sc:// protocol, which is the Spark Connect standard. The use_ssl=true parameter supports encrypted communications over TLS, so your data and credentials are protected in transit.

from pyspark.sql import SparkSession

# Use the endpoint and auth token from get-session-endpoint
session_endpoint="<endpoint-from-get-session-endpoint>"
auth_token="<authToken-from-get-session-endpoint>"

spark_connect_url = (
    f"sc://{session_endpoint}:443/;use_ssl=true;x-aws-proxy-auth={auth_token}"
)

spark = SparkSession.builder \
    .remote(spark_connect_url) \
    .getOrCreate()

# Query data in your S3 data lake
df = spark.sql("SELECT * FROM my_catalog.my_database.my_table")
df.show()

# Run transformations at scale
df.groupBy("category").count().orderBy("count", ascending=False).show()
spark.stop()

Once connected, Spark operations you write in your IDE can be run on EMR Serverless. For debugging, you can pause the execution at breakpoints, inspect variables, and step through your transformations locally while EMR Serverless processes your data on remote, scalable infrastructure.

Sessions remain active for a configurable idle timeout (1 hour by default). If your connection drops, the session continues running, allowing you to reconnect without losing your work. You can also access the live Spark UI through the GetResourceDashboard API to monitor queries, stages, and executors in real time. After the session ends, the Spark History Server remains available for post-run analysis.

Clean up resources

If the 1-hour session idle timeout does not meet your needs, you can manually remove sessions to avoid ongoing costs. Note that terminating an active session will immediately stop you running Spark operations. Before doing that, verify all your critical data processing is completed, and results are saved.

# 1. Stop the active session
aws emr-serverless terminate-session \
  --application-id $APP_ID \
  --session-id $SESSION_ID

# 2. Stop the application
aws emr-serverless stop-application --application-id $APP_ID

Use cases

Spark Connect on EMR Serverless supports a wide range of development workflows. The following are some of the most popular use cases, including but not limited to:

Interactive ETL development — Build and test data pipelines interactively, validating transformations against full-scale datasets before promoting them to production as batch jobs.
SageMaker Unified Studio (SMUS) Data Notebooks — Run interactive PySpark sessions directly from SMUS Data Notebooks connected to EMR Serverless through Spark Connect.
Direct S3 and JDBC access without a catalog — Connect directly to S3 files and JDBC data sources without needing a metastore or catalog configuration.
Apache Iceberg Data Lakehouse analytics — Query and manage Iceberg tables through the AWS Glue Data Catalog, with full support for time travel, schema evolution, and partition management.
Amazon S3 Tables with federated catalog — Access S3 Tables as a federated Glue Data Catalog source, combining Iceberg features with serverless Spark execution.
dbt-spark — Run dbt-spark adapter against EMR Serverless via Spark Connect, allowing analytics engineers to develop and test transformations locally with dbt framework while using EMR Serverless as the remote Spark engine.
Exploratory data analysis and feature engineering — Analyze production-scale data from your preferred notebook environment instead of using sampled subsets, helping you catch data quality issues earlier.
Compute standardization — Standardize EMR Serverless as the Spark backend while giving you the flexibility to use preferred local tools, version control, and CI/CD workflows.

These use cases work across multiple client surfaces: IDEs, Jupyter notebooks, dbt-spark, and AI coding agents. Because Spark Connect is an open Apache Spark standard, the same PySpark code typically works across different Spark backends by changing the connection endpoint.

Availability and pricing

Spark Connect on EMR Serverless is now available with Apache Spark 3.5.6 on Amazon EMR release 7.13 and higher in all AWS Regions where EMR Serverless is available. There is no additional charge for using Spark Connect. You pay for the EMR Serverless compute resources (vCPU, memory, and storage) consumed during your session, the same pricing model as EMR Serverless batch jobs.

Conclusion

Spark Connect on EMR Serverless bridges the gap between local development and production-scale execution. Build and debug PySpark applications from your preferred environment (IDE, notebook, dbt, or AI coding agent) while EMR Serverless handles automatic scaling, per-session cost visibility, and infrastructure management behind the scenes. With ARN-addressable sessions, fine-grained IAM permissions, tag-based cost allocation, and per-session configuration overrides, your team gets the controls they need without sacrificing flexibility.

Get started today with EMR release 7.13.0 (Spark 3.5.6). Follow the step-by-step tutorial in the EMR Serverless Developer Guide to create your first Spark Connect session and experience interactive, serverless PySpark development firsthand.

About the authors

Build stateful streaming applications with Apache Spark 4.0 on Amazon EMR Serverless

Raj Ramasubbu — Tue, 09 Jun 2026 16:14:54 +0000

Apache Spark 4.0 represents a major milestone in stream processing, introducing new capabilities that fundamentally change how developers build stateful streaming applications. At the heart of these improvements is the transformWithState API – a new capability that enables first-class support for timers, automatic state management, and schema evolution to Spark Structured Streaming.

With Spark 4.0 now available on Amazon EMR Serverless, developers can build stateful streaming applications using the transformWithState API in a fully managed, serverless environment that automatically scales based on workload demands. This combination delivers the power of sophisticated stream processing without the operational overhead of cluster management.

In this post, we demonstrate how to build a production-ready IoT device monitoring system using Spark 4.0’s transformWithState API on Amazon EMR Serverless. This example showcases the key capabilities of stateful streaming and provides a template you can adapt for your own use cases.

Apache Spark 4.0: introducing transformWithState

Apache Spark 4.0’s latest streaming features solve common production challenges in stateful applications by introducing native timer support and advance state management capabilities for complex event processing workflows. The new transformWithState API provides:

Key features of transformWithState

Native timer support: Register timers that fire callbacks at specific times for use cases like heartbeat monitoring, session timeout detection, and SLA violation alerts.
Automatic state TTL (Time-To-Live): Configure automatic expiration policies to prevent state from growing indefinitely. This is useful for use cases like session state size control, clearing stale device telemetry, maintaining a recency cache, or tracking invalid logins within the last hour for fraud detection.
Schema evolution: Evolve state schema without restarting from a new checkpoint. Add optional fields, remove fields, or widen numeric types. This is particularly valuable for use cases where data structures are dynamic, and application downtime for schema migration is not acceptable, enabling more resilient and flexible real-time streaming applications.
Multiple state variables: Support for multiple independent state variables (ValueState, ListState, MapState) per key, well-suited for building complex, real-time applications that require sophisticated state management, such as storing a history of recent error codes, tracking counts of various alert types, or maintaining multiple dimensions of user activity within a single stateful operator.
State observability: Query application state mid-stream using the State Data Source Reader for debugging and monitoring. This is especially valuable in applications that require maintaining and evolving state through several steps, such as detection of sophisticated event patterns across multiple streams and over time, where visibility into state transitions is critical for troubleshooting and validation.
Operator chaining: Chain multiple stateful operators together for complex multi-stage processing pipelines.

These capabilities make Spark 4.0 ideal for applications that were previously difficult or impossible to implement efficiently, such as complex event processing, session analytics, anomaly detection, and real-time monitoring systems.

Use case: IoT heartbeat monitoring

Consider a fleet of 100,000 IoT sensors deployed across manufacturing facilities. Each sensor sends a heartbeat signal every 20 seconds to indicate it’s operational. Your operations team needs to be alerted within 30 seconds if any sensor goes offline, with repeat alerts every 60 seconds until the sensor comes back online.

This seemingly simple requirement presents several technical challenges. The application must maintain the last heartbeat timestamp for each of the 100,000 devices while independently managing timers to detect missed signals per device. It also needs to handle out-of-order heartbeats caused by network delays and clean up state for decommissioned devices to prevent unbounded memory growth. All of this must happen at scale, processing millions of events per minute with low latency, while recovering gracefully from failures without losing state.

To address the specific challenges of IoT heartbeat monitoring described above, we present a solution built on the transformWithState API in Spark 4.0. With its native timer support, automatic state management, and built-in fault tolerance, making it the ideal solution for IoT heartbeat monitoring at scale.

Solution overview

Our solution architecture follows a serverless, event-driven design:

IoT devices send heartbeat events to Amazon Kinesis Data Streams containing device ID, timestamp, and metadata (battery level, signal strength, firmware version).
Amazon EMR Serverless reads from Kinesis using the Spark aws-kinesis connector using VPC Endpoint for Kinesis, then parses JSON events into structured DataFrames and grouping by device_id.
transformWithState processes each device’s stream. On heartbeat arrival, it updates state and registers a 30-second timer; when the timer expires without a new heartbeat, it emits an offline alert.
State is automatically persisted to RocksDB locally and checkpointed to Amazon Simple Storage Service (Amazon S3), enabling fault-tolerant recovery and exactly-once processing semantics.
Alerts are delivered via Amazon Simple Notification Service (Amazon SNS) to configured subscribers (email, SMS, AWS Lambda, webhooks).

Prerequisites

Before implementing this solution, verify that you have:

AWS account: With permissions for EMR Serverless, Kinesis, SNS, S3, VPC, and IAM.
AWS Command Line Interface (AWS CLI): Configured with appropriate credentials.
VPC setup: VPC with private subnets and security groups configured.
Kinesis VPC interface endpoint: VPC endpoint for private connectivity to Kinesis.
Kinesis Data Stream: Created for ingesting heartbeat events (for example, iot-heartbeats). For testing your streaming data solution, refer to Test your streaming data solution with the new Amazon Kinesis Data Generator.
SNS topic: Created for sending alerts (for example, iot-alerts).
S3 bucket: For storing application code, dependencies, and checkpoints.

Step-by-step implementation

The following steps walk you through setting up an EMR Serverless application with Spark 4.0, configuring the stateful streaming processor, and deploying the IoT heartbeat monitoring solution.

Step 1: Create the EMR serverless application

Run the following command in your terminal using the AWS CLI. Replace the subnet and security group IDs with the values from your VPC setup.

# Create EMR Serverless application with Spark 4.0 and VPC
aws emr-serverless create-application \
  --name "iot-heartbeat-monitor" \
  --release-label "emr-spark-8.0.0" \
  --type "SPARK" \
  --network-configuration '{
    "subnetIds": ["subnet-xxxxx", "subnet-yyyyy"],
    "securityGroupIds": ["sg-zzzzz"]
  }' \
  --region us-east-1

The command returns a JSON response containing the application details. Note the applicationId value from the output, as you will need it in subsequent steps.

Step 2: Implement the heartbeat monitor

The core of our solution is the HeartbeatMonitor class that extends StatefulProcessor. This class demonstrates the key features of Spark 4.0’s transformWithState API. Download the full implementation script and upload it to your local S3 bucket for execution. Let’s walk through each component to understand how it works.

2.1 Initialize state variables

The init() method is called once when the processor is initialized. This is where we define and register our state variables.

from pyspark.sql.streaming.stateful_processor import (
    StatefulProcessor, StatefulProcessorHandle
)

class HeartbeatMonitor(StatefulProcessor):

    def init(self, handle: StatefulProcessorHandle) -> None:
        self.handle = handle

        # Define state schemas
        last_seen_schema = StructType([
            StructField("timestamp", TimestampType(), True)
        ])

        device_info_schema = StructType([
            StructField("battery_level", StringType(), True),
            StructField("firmware_version", StringType(), True)
        ])

        # Initialize multiple independent state variables
        self.last_seen = handle.getValueState("last_seen", last_seen_schema)
        self.device_info = handle.getValueState(
            "device_info", device_info_schema
        )

In the init() method, we use StatefulProcessorHandle to define and initialize two per-key state variables, last_seen and device_info, using Spark’s StructType schemas and the getValueState() API. These state variables are automatically stored in RocksDB and checkpointed to S3, allowing for fault-tolerant state management across streaming micro-batches.

2.2 Handle incoming heartbeat events and register timers

The handleInputRows() method is called whenever new events arrive for a device. This is where we update state and register timers.

def handleInputRows(
    self, key: tuple, rows: Iterator[pd.DataFrame], timerValues
) -> Iterator[pd.DataFrame]:
    device_id = key[0]

    # Process incoming heartbeats - iterate through all rows to find latest
    latest_timestamp = None
    for pdf in rows:
        for _, row in pdf.iterrows():
            ts = row['timestamp']
            if pd.isna(ts):
                continue
            if latest_timestamp is None or ts > latest_timestamp:
                latest_timestamp = ts

    if latest_timestamp is None:
        yield pd.DataFrame()
        return

    # Check if we have existing state
    existing_timestamp = None
    if self.last_seen.exists():
        existing_state = self.last_seen.get()
        existing_timestamp = existing_state[0]

    # Update state only if new heartbeat is more recent
    if existing_timestamp is None or latest_timestamp > existing_timestamp:
        # Cancel existing timers (device is back online)
        for timer in self.handle.listTimers():
            self.handle.deleteTimer(timer)

        # Update state with new timestamp
        self.last_seen.update((latest_timestamp,))

        # Register timer for heartbeat deadline detection
        current_time_ms = timerValues.getCurrentProcessingTimeInMs()
        deadline_ms = current_time_ms + HEARTBEAT_INTERVAL_MS
        # 30 seconds from now
        self.handle.registerTimer(deadline_ms)

    yield pd.DataFrame()  # No output from input handling

The handleInputRows() method processes incoming heartbeat events for each device by extracting the latest timestamp, updating the last_seen state, and managing timers. It cancels existing ones and registering a new 30-second expiry timer to detect future inactivity. Because alerts are only emitted upon timer expiration, the method yields an empty dataframe during normal heartbeat processing.

2.3 Handle timer expiration and emit alerts

The handleExpiredTimer() method is called when a registered timer fires. This is where we detect offline devices and emit alerts.

def handleExpiredTimer(
    self, key: tuple, timerValues, expiredTimerInfo
) -> Iterator[pd.DataFrame]:
    device_id = key[0]
    current_time_ms = timerValues.getCurrentProcessingTimeInMs()

    # Verify state exists
    if not self.last_seen.exists():
        yield pd.DataFrame()
        return

    # Get last seen timestamp from state
    last_seen_state = self.last_seen.get()
    last_seen_timestamp = last_seen_state[0]

    if last_seen_timestamp is None or pd.isna(last_seen_timestamp):
        yield pd.DataFrame()
        return

    # Calculate how long device has been offline
    last_seen_ms = int(last_seen_timestamp.timestamp() * 1000)
    offline_duration_ms = current_time_ms - last_seen_ms
    offline_duration_seconds = offline_duration_ms / 1000.0

    # Create alert as a Pandas DataFrame
    alert_df = pd.DataFrame({
        "device_id": [device_id],
        "alert_type": ["DEVICE_OFFLINE"],
        "last_seen": [last_seen_timestamp],
        "offline_duration_seconds": [offline_duration_seconds],
        "alert_timestamp": [datetime.fromtimestamp(current_time_ms / 1000.0)]
    })

    # Register another timer for repeat alerts (every 60 seconds)
    next_alert_time = current_time_ms + ALERT_REPEAT_INTERVAL_MS
    self.handle.registerTimer(next_alert_time)

    yield alert_df  # Emit the alert

The handleExpiredTimer() method is triggered automatically when a device’s inactivity timer expires, retrieving the last_seen state to calculate the offline duration and yielding an alert dataframe to the output stream. It also registers a follow-up timer for repeat alerts every 60 seconds, which continues until a new heartbeat arrives and cancels the timer via handleInputRows().

There are several ways you could extend this solution for production use. You could implement exponential backoff for repeat alerts to reduce noise, for example, alerting after 60 seconds, then 2 minutes, then 5 minutes, and so on. Other improvements could include adding severity escalation based on offline duration, integrating with notification services like Amazon SNS for downstream alerting, or setting a maximum retry limit to stop alerts for permanently decommissioned devices.

2.4 Apply transformWithState to the streaming DataFrame

Now we connect everything together by applying our HeartbeatMonitor processor to the streaming data.

# Read and parse heartbeat events from Kinesis
parsed_df = kinesis_df \
    .selectExpr("CAST(data AS STRING) as json_data") \
    .select(from_json(col("json_data"), heartbeat_schema).alias("heartbeat")) \
    .select(
        col("heartbeat.device_id"),
        to_timestamp(col("heartbeat.timestamp")).alias("timestamp"),
        col("heartbeat.battery_level"),
        col("heartbeat.signal_strength"),
        col("heartbeat.firmware_version")
    )

# Apply transformWithState for stateful processing
alerts_df = parsed_df \
    .groupBy("device_id") \
    .transformWithStateInPandas(
        statefulProcessor=HeartbeatMonitor(),
        outputStructType=alert_output_schema,
        outputMode="append",
        timeMode="processingTime"
    )

# Write alerts to SNS
query = alerts_df.writeStream \
    .outputMode("append") \
    .foreachBatch(send_to_sns) \
    .option("checkpointLocation", CHECKPOINT_LOCATION) \
    .trigger(processingTime="10 seconds") \
    .start()

# Send to SNS for alerts
def send_to_sns(batch_df, batch_id):
    if batch_df.count() > 0:
        sns_client = boto3.client('sns', region_name=KINESIS_REGION)
        for row in batch_df.collect():
            message = {
                "device_id": row["device_id"],
                "alert_type": row["alert_type"],
                "last_seen": str(row["last_seen"]),
                "offline_duration_seconds": row["offline_duration_seconds"],
                "alert_timestamp": str(row["alert_timestamp"])
            }
            sns_client.publish(
                TopicArn=SNS_TOPIC_ARN,
                Message=json.dumps(message),
                Subject=f"Device Offline Alert: {row['device_id']}"
            )

The streaming pipeline parses JSON heartbeat events from Kinesis, partitions them by device_id, and applies the HeartbeatMonitor stateful processor using transformWithStateInPandas() with processing-time timers and append output mode. The resulting alert stream is written to SNS via foreachBatch() with checkpointing enabled for fault tolerance and micro-batches triggered every 10 seconds.

To summarize, implementing the heartbeat monitor requires just three methods. The init() method sets up your state variables, handleInputRows() processes incoming heartbeats and manages timers, and handleExpiredTimer() generates offline alerts. The transformWithState API handles the underlying complexity of state management, checkpointing, and timer scheduling automatically.

Step 3: Create IAM role for job execution

Create an IAM role that allows EMR Serverless to assume it for running your Spark job. For detailed instructions on creating an IAM role, see Creating an IAM role. Use the following trust policy for the role.

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {
      "Service": "emr-serverless.amazonaws.com"
    },
    "Action": "sts:AssumeRole"
  }]
}

Attach a permissions policy that grants the role access to read from the Kinesis stream, write to the S3 bucket for checkpoints and application artifacts, and publish alerts to the SNS topic:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "KinesisAccess",
      "Effect": "Allow",
      "Action": [
        "kinesis:GetRecords",
        "kinesis:GetShardIterator",
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
        "kinesis:SubscribeToShard"
      ],
      "Resource": "arn:aws:kinesis:us-east-1:*:stream/iot-heartbeats"
    },
    {
      "Sid": "SNSPublish",
      "Effect": "Allow",
      "Action": "sns:Publish",
      "Resource": "arn:aws:sns:us-east-1:*:iot-alerts"
    },
    {
      "Sid": "S3Access",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::your-bucket",
        "arn:aws:s3:::your-bucket/*"
      ]
    }
  ]
}

Step 4: Upload external dependencies required for executing the streaming job

In this step, you will download the required external dependencies and upload them to your S3 bucket to make them available for your EMR Serverless streaming job.

Spark-kinesis-connector.jar (download link) and copy to local S3 bucket s3://your-bucket/jars/spark-kinesis-connector.jar.
Protobuf Dependency (download link) and copy to local S3 bucket s3://your-bucket/pyfiles/protobuf_pkg.tar.gz.

Step 5: Submit the streaming job

Now that the application, IAM role, and dependencies are in place, you can submit the streaming job. This step configures the Spark job parameters and submits it to your EMR Serverless application in streaming mode. For more details on submitting jobs, see Starting a job run.

First, create a file named job-driver.json with the following content. Replace the S3 paths with the locations where you uploaded your script and dependencies in the previous steps.

{
  "sparkSubmit": {
    "entryPoint": "s3://your-bucket/scripts/heartbeat_monitor.py",
    "sparkSubmitParameters": "--jars s3://your-bucket/jars/spark-kinesis-connector.jar --archives s3://your-bucket/pyfiles/protobuf_pkg.tar.gz#protobuf_pkg --conf spark.executor.cores=4 --conf spark.executor.memory=16g --conf spark.driver.cores=4 --conf spark.driver.memory=16g --conf spark.executor.instances=3 --conf spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider --conf spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled=true --conf spark.emr-serverless.driverEnv.PYTHONPATH=./protobuf_pkg --conf spark.executorEnv.PYTHONPATH=./protobuf_pkg"
  }
}

Then, run the following command to submit the job. Replace the application ID and account ID with your own values.

aws emr-serverless start-job-run \
  --application-id <YOUR_APPLICATION_ID> \
  --execution-role-arn arn:aws:iam::<ACCOUNT_ID>:role/EMRServerlessJobRole \
  --job-driver file://job-driver.json \
  --mode STREAMING \
  --retry-policy maxFailedAttemptsPerHour=1 \
  --region us-east-1

Running transformWithState on Amazon EMR Serverless provides several operational advantages over self-managed Spark clusters. In streaming mode, the Spark driver remains alive between micro-batches, eliminating the overhead of repeatedly starting and stopping the application. You don’t need to provision or manage executors because EMR Serverless automatically scales compute resources up and down based on workload demands, so you only pay for what you use. Your IoT heartbeat monitor can handle traffic spikes, such as thousands of devices reconnecting simultaneously after a network outage, without manual intervention. EMR Serverless also provides built-in job resiliency, real-time monitoring, and enhanced log management, reducing the operational burden of running streaming applications in production.

Testing the solution

Now that our streaming application is deployed, let’s test it by sending heartbeat events and observing the offline detection behavior.

Step 1: Open AWS CloudShell

Open AWS CloudShell in your AWS account from the AWS Management Console.

Step 2: Send heartbeat events using CLI

Execute the following bash script to send heartbeat events every 10s.

#!/bin/bash

while true; do
  aws kinesis put-record \
    --stream-name iot-heartbeats \
    --partition-key device-001 \
    --data $(echo "{\"device_id\":\"device-001\",\"timestamp\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"battery_level\":87.5,\"signal_strength\":-42.3,\"firmware_version\":\"v2.1.0\"}" | base64) \
    --region us-east-1

  aws kinesis put-record \
    --stream-name iot-heartbeats \
    --partition-key device-002 \
    --data $(echo "{\"device_id\":\"device-002\",\"timestamp\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"battery_level\":87.5,\"signal_strength\":-42.3,\"firmware_version\":\"v2.1.0\"}" | base64) \
    --region us-east-1

  aws kinesis put-record \
    --stream-name iot-heartbeats \
    --partition-key device-003 \
    --data $(echo "{\"device_id\":\"device-003\",\"timestamp\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"battery_level\":87.5,\"signal_strength\":-42.3,\"firmware_version\":\"v2.1.0\"}" | base64) \
    --region us-east-1

  sleep 10
done

Update the timestamp field to use the current time for each event or use a script to automate sending events at regular intervals.

Step 3: Observe normal operation

As you send heartbeat events every 10 seconds, the Spark application receives each event and updates the device’s state. A timer is then registered for 30 seconds in the future. Each new heartbeat cancels the existing timer and registers a new one, effectively resetting the countdown. As long as heartbeats continue to arrive within the 30-second window, no alerts are sent.

The above timeline diagram shows a 60-second window of normal device operation. Heartbeat events arrive every 10 seconds (at 0s, 10s, 20s, 30s, 40s, 50s, and 60s), each resetting the 30-second timer window. Because every heartbeat arrives well within the 30-second threshold, the timer never expires, the device state remains online, and no alerts are triggered.

Step 4: Test offline detection

Stop sending heartbeat events for the device and wait 30 seconds. You should receive an SNS alert indicating the device is offline.

Timeline diagram showing offline detection over 110 seconds. Device sends heartbeats at 0s, 10s, and 20s before going offline. The 30-second timer expires at 50s triggering Alert #1 via SNS, followed by a repeat Alert #2 at 110s after a 60-second repeat timer.

If you continue to not send heartbeats, additional alerts will be sent every 60 seconds.

Step 5: Test device recovery

Resume sending heartbeat events using the same CLI command. The application will cancel all existing timers for the device and will stop sending SNS alerts.

Timeline diagram showing the complete device recovery lifecycle over 140 seconds across three phases: normal operation with heartbeats, offline detection with SNS alerts, and recovery where timers are canceled and the device returns to online state

Clean up

To avoid incurring ongoing charges, follow these steps to clean up the resources.

Step 1: Stop the EMR serverless application

Stop your running streaming job:

aws emr-serverless stop-job-run \
  --application-id <your-application-id> \
  --job-run-id <your-job-run-id>

Step 2: Delete the EMR serverless application

aws emr-serverless delete-application \
  --application-id <your-application-id>

Step 3: Delete kinesis data stream

aws kinesis delete-stream --stream-name iot-heartbeat

Step 4: Remove S3 objects

Delete the checkpoint data, scripts, and dependencies from your S3 bucket:

aws s3 rm s3://your-bucket/checkpoints/ --recursive
aws s3 rm s3://your-bucket/scripts/ --recursive
aws s3 rm s3://your-bucket/jars/ --recursive
aws s3 rm s3://your-bucket/pyfiles/ --recursive

Real-world use cases for stateful streaming

The transformWithState API enables developers to build sophisticated streaming applications that were previously difficult to implement. Here are a few examples of how it can be applied across industries.

Telecommunications and network monitoring: Telecom providers need to detect network anomalies and SLA violations as they happen across millions of concurrent sessions. With transformWithState, developers can maintain per-session state to track call detail records, compare real-time network metrics against established baselines, and trigger alerts the moment thresholds are breached. Automatic state TTL ensures that completed session records are cleaned up without manual intervention.

Financial services and fraud detection: Detecting fraud requires correlating multiple signals across a sequence of transactions in real time. With transformWithState, developers can maintain per-account state that tracks transaction histories, flags suspicious patterns like rapid purchases across geographies, and calculates rolling risk scores. Multiple state variables per key allow tracking different dimensions of activity, such as transaction velocity, location changes, and spending deviations, within a single stateful operator.

E-commerce and customer engagement: Understanding customer behavior in real time is critical for driving conversions. Using transformWithState, developers can build session-aware applications that track browsing and cart activity with timer-based state expiration, detecting cart abandonment after a configurable timeout and triggering personalized re-engagement notifications. The State Data Source Reader enables teams to inspect session state mid-stream, making it easier to debug and validate real-time customer journey logic.

Conclusion

Apache Spark 4.0’s transformWithState API represents a significant advancement in stateful stream processing, making it simpler to build complex real-time applications like IoT device monitoring. Combined with Amazon EMR Serverless, you get a fully managed platform that scales automatically and eliminates infrastructure management overhead.

This post demonstrates how to use the native timer support capability of transformWithState to build a real-time IoT device monitoring application. We encourage you to explore other capabilities such as Automatic State TTL, Schema Evolution, and Multiple State Variables on Amazon EMR Serverless to build more sophisticated streaming applications tailored to your needs.

About the authors

Announcing general availability of Apache Spark 4.0 on Amazon EMR

Suthan Phillips — Tue, 09 Jun 2026 16:13:21 +0000

As data volumes grow and pipelines become more complex, you need an engine that handles semi-structured data natively, supports streaming state without operational overhead, and allows you to develop interactively against production-scale compute. Spark 4.0 addresses these three challenges that slow modern data teams: wrangling semi-structured data, managing streaming state, and bridging the gap between interactive development and production-scale execution. With VARIANT data type, state-management improvements, and Spark Connect availability in Spark 4.0, you can now handle these workloads with less code, fewer operational trade-offs, and faster iteration cycles, all on Amazon EMR optimized runtime, which runs Spark workloads up to 4.5× faster than open-source Apache Spark.

With this general availability announcement, Spark 4.0 is now supported across Amazon EMR Serverless, Amazon EMR on EC2, and Amazon EMR on EKS deployment options. In this post, you’ll learn about key Spark 4.0 capabilities now available on Amazon EMR including Spark Connect, the Variant data type, SQL scripting, Python API improvements, and streaming enhancements, along with infrastructure changes in the new emr-spark-8.0 release.

New features in GA

Apache Spark 4.0 introduces several capabilities that are now generally available on Amazon EMR.

Spark Connect

Most Spark development is iterative and disconnected from production. You write code locally, test it against a sample, then package and deploy it to a cluster. It often fails due to data issues at scale, environment mismatches, or dependency conflicts. The feedback loop is slow, and the gap between development and production is where most time is lost.

Spark Connect closes that gap by introducing a decoupled client-server architecture that changes how your application communicates with Spark. In previous versions, your application code and the Spark driver ran inside the same JVM process, meaning issues in your application code could destabilize the Spark driver and disrupt the entire session. Your application runs as a lightweight client that submits logical plans to a Spark server over gRPC. The server handles execution independently. Your client doesn’t require a local Spark installation, a JVM, and doesn’t need to run on a cluster node. It only needs connectivity to the server endpoint.

With Amazon EMR, this means you can write PySpark from your preferred IDE (VS Code, PyCharm), Jupyter notebooks, Amazon SageMaker Unified Studio Data Notebooks, Amazon Q Developer, or Kiro, and Spark Connect routes your DataFrame transformations and SQL queries to Amazon EMR for execution over a secure connection.You can set breakpoints, inspect variables, and step through transformations while your data is processed on serverless compute, catching issues during development instead of after deployment. There are no clusters to provision, no code to repackage, and no infrastructure to manage.

This architecture also improves session resilience. A client-side failure doesn’t affect the Spark server, so other workloads continue to run without disruption. Spark Connect is an open Apache Spark standard. The same PySpark code works across different Spark backends by changing the connection endpoint.

For example, connecting to Amazon EMR Serverless from your local IDE takes minimal lines of spark code:

from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .remote("sc://<endpoint>:443/;use_ssl=true;x-aws-proxy-auth=") \
    .getOrCreate()
df = spark.sql("SELECT * FROM my_catalog.my_database.my_table")
df.groupBy("category").count().show()

On Amazon EMR Serverless, start a session to retrieve your endpoint and auth token, then connect remotely using the standard sc:// protocol. Every Spark operation executes on Amazon EMR Serverless while your code stays local.

The following video showcases Spark Connect and Variant features together.

For a step-by-step getting-started walkthrough, visit Announcing Spark Connect on Amazon EMR Serverless: Interactive PySpark development, anywhere.

Data type and table format enhancements

This section covers the VARIANT data type and Apache Iceberg V3 support. These two additions improve how you store and query semi-structured data.

Apache Iceberg V3 support

Amazon EMR has supported Apache Iceberg V3 since Amazon EMR release 7.x, introducing capabilities such as deletion vectors and row lineage. With Spark 4.0 on Amazon EMR, that support deepens unlocking capabilities that had a hard dependency on Spark 4.0 itself, including VARIANT column storage and unknown type handling. For teams running data lakehouse workloads, the table format underneath your data determines how efficiently it is stored, how reliably it evolves, and how safely multiple tools can read and write it simultaneously.

What this means for your workloads:

VARIANT and Iceberg working together: VARIANT columns can now be stored natively in Iceberg V3 tables, combining efficient semi-structured data storage with Iceberg’s schema evolution and time travel capabilities. This eliminates the pipeline complexity of upfront schema definitions.
More efficient partitioning: Multi-argument transforms accept multiple input columns in a single partition expression, such as range (order_date, product_category), giving you finer control over data layout. They produce a single composite key instead of separate columns whose cartesian product can explode partition count. The result is less data scanned, faster queries, and lower compute costs for high volume workloads.
Safer schema evolution: Unknown type handling ensures that older readers do not break when newer writers introduce new column types, reducing coordination overhead across teams and tools during upgrades.
Fine-grained access control (FGAC): Column-level and row-level permissions are now available through AWS Lake Formation, giving you governed access control at a granular level across your Iceberg tables, no custom access logic required.

Variant data type

The new VARIANT data type, supported through Apache Iceberg v3, brings native support for semi-structured JSON data directly into Spark SQL. This matters most when you don’t control the data being written because platform teams and shared services often receive data from partners and upstream teams with unpredictable or evolving structures.

Without VARIANT, handling semi-structured data meant accepting real tradeoffs: defining schemas upfront that broke when data evolved, storing everything as strings with heavy parsing costs on every read, or building wide tables with nullable columns that wasted storage on empty fields. The most realistic option was breaking nested structures apart into separate columns before running queries. This ETL step added latency, increased storage costs, and broke every time an upstream team added or removed fields from their data feed.

VARIANT eliminates the process entirely. Data stays nested and is queryable with variant_get(), without a separate ETL pipeline. You ingest without defining a schema first and apply structure at query time.

For example, querying nested fields is now a single expression:

SELECT
    variant_get(payload, '$.user.name') AS user_name,
    variant_get(payload, '$.event.type') AS event_type,
    variant_get(payload, '$.event.timestamp') AS event_timestamp
FROM VALUES
    (PARSE_JSON('{"user":{"name":"Alice"},"event":{"type":"click","timestamp":"2025-03-01"}}'))
AS t(payload)
WHERE variant_get(payload, '$.event.timestamp') > '2025-01-01';

For a deep dive into how VARIANT is stored in Parquet, shredding mechanics, and a full end-to-end walkthrough on Amazon EMR Serverless, see Beyond JSON blobs: Implementing the VARIANT data type in Apache Iceberg V3.

Key benefits for your workloads:

Reduced pipeline fragility: Schema changes no longer break ingestion. Data lands as-is, and you apply structure at query time based on what each analysis needs, without upstream coordination.
Improved query performance: Optimized storage format enables efficient access to nested fields without parsing overhead, so queries run faster even on deeply nested payloads.
Better storage efficiency: Compact encoding eliminates the waste of NULL-heavy wide tables, reducing storage costs for semi-structured data at scale.

VARIANT is especially well-suited where schemas are unpredictable or evolving: IoT and sensor data with device-specific payloads, logging and telemetry with variable event structures, and API responses and webhooks from third-party services where the schema changes without notice.

SQL enhancements

You can now write and maintain Spark pipelines using the same standard SQL you already know, no Spark-specific functions or syntax required. Apache Spark 4.0 expands ANSI SQL compliance so that functions behave consistently, opening Spark to anyone who can write SQL rather than requiring Spark specialists.

Standard SQL syntax such as OFFSET, LIMIT ... OFFSET, and lateral column aliases now work as expected. For example:

-- Standard OFFSET syntax now supported
SELECT id, name
FROM VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Carol'), (4, 'Dave') AS t(id, name)
ORDER BY id
LIMIT 2 OFFSET 1;

-- Lateral column aliases work inline
SELECT amount * 1.1 AS adjusted, adjusted * 0.08 AS tax
FROM VALUES (100.0), (200.0), (350.0) AS t(amount);

Beyond syntax, SQL scripting brings procedural logic directly into Spark SQL. You can now use variables, IF/ELSE conditionals, loops, and multi-statement blocks without switching to Python or JVM-based languages. Before SQL scripting, multi-step workflows (such as ETL logic with conditional branching or iterative data quality checks) required wrapping SQL statements in Python or Scala to handle control flow. SQL scripting removes that dependency. SQL-native teams can author and maintain these workflows entirely in SQL.

Key benefits:

Simplified ETL workflows: Multi-step transformation logic that previously required an external language can now live entirely in SQL, reducing code complexity and making pipelines easier to build and maintain.
Lower barrier for SQL-native teams: Teams that primarily work in SQL no longer need to context-switch into Python or Scala to implement conditional logic or iterative processing. The entire pipeline stays in SQL.

Python advances

Earlier versions of Spark required Python users to step outside Python at two key points: building custom data connectors required Java or Scala, and diagnosing UDF performance lacked built-in visibility. Spark 4.0 addresses both directly, removing the two biggest blockers for organizations where Python is the primary language.

Python Data Source API

With the Python Data Source API, you can build custom, reusable data connectors in Python without any JVM or Scala code. Custom connectors participate in Spark’s query optimization, including predicate pushdown and schema inference. This matters when your data system only has a Python client, or when your team does not have Java or Scala expertise: you can now wrap any custom format or external source as a Spark DataFrame source or sink without leaving Python.

Spark 4.0 also introduces polymorphic Python UDTFs (User-Defined Table Functions) that can return different schema shapes depending on input, with an analyze() method that produces a schema dynamically based on parameters. This is particularly useful for processing varying JSON schemas or splitting inputs into a variable set of outputs.

If you’re ingesting data from a REST API with a Python client, you can implement a custom Spark data source entirely in Python, register it, and use it directly in Spark SQL or the DataFrame API. What previously required a Scala developer and a custom JVM connector can now be built and maintained by your Python team running the pipeline.

Python UDF enhancement

Python UDF profiling provides built-in visibility into execution time, serialization overhead, and memory usage at the individual UDF level without external tooling. Additionally, it enables performance or memory profiling depending on what you need to diagnose.

Arrow-based vectorized UDF support reduces serialization overhead between Python and the JVM using a columnar format, replacing row-at-a-time processing with batch-oriented columnar exchange.

Together, these give you a complete optimization loop: build custom connectors in Python, profile your UDF performance, and optimize with confidence.

Practical benefits for Python teams:

Lower barrier for Python teams: Custom data connectors no longer require Java or Scala knowledge. If your data system has a Python client, you can build a production-grade Spark connector entirely in Python.
Flexible data transformation: Polymorphic UDTFs let your functions adapt to varying input schemas dynamically, reducing the need to write and maintain multiple transformation functions for different data shapes.
Faster UDF optimization: Built-in profiling surfaces exactly where execution time and memory are being spent at the UDF level, replacing guesswork with direct visibility and making performance tuning significantly faster.

Streaming enhancements

This section covers improvements to state management and observability in structured streaming workloads.

Queryable state for structured streaming

Structured streaming jobs maintain state continuously (running totals, session windows, aggregated counts). However, in earlier versions of Spark that state was locked inside the running job. Inspecting it meant stopping the stream or manually parsing checkpoint files. For production workloads, this created real operational risk: teams had no way to verify whether state was correct, corrupted, or drifting without taking the job down.

Time-sensitive applications faced an additional problem: timers in Spark streaming only fired when new data arrived, so a five-minute heartbeat timeout could silently miss its window if no data came in, making applications like heartbeat monitoring and session tracking unreliable by design.

Spark 4.0 changes this. The new transformWithState API provides deterministic timer execution because timers fire on schedule regardless of data arrival patterns. It also delivers automatic state TTL to prevent unbounded growth, schema evolution without restarting from a new checkpoint, and state observability for mid-stream debugging. External systems can now read live aggregated state from a running streaming job without interrupting it. State is accessible as a DataFrame, queryable during development, verifiable in unit tests, and inspectable during production incidents without touching the running stream.

This is backed by three improvements working together. First, the transformWithState operator replaces mapGroupsWithState from earlier Spark versions (which had limited timer support and no TTL-based cleanup). Second, the state data source reader exposes streaming state as a queryable DataFrame. Lastly, RocksDB changelog checkpointing improvements address throughput bottlenecks in high-volume stateful workloads.

Consider a fleet of 100,000 IoT sensors across manufacturing facilities, each requiring an alert within 30 seconds of going offline. The sensors track heartbeat state per device, managing independent timers, handling late data, and cleaning up decommissioned devices at scale had no clean solution in earlier Spark versions. The transformWithState operator handles all of this natively, and queryable state lets your operations team inspect live device state in real time without stopping the stream:

# Timers fire on schedule regardless of data arrival, making heartbeat monitoring reliable
alerts = events_df.groupBy("device_id").transformWithState(
    HeartbeatMonitor(),
    outputStructType=StructType([
        StructField("device_id", StringType()),
        StructField("alert", StringType())
    ]),
    outputMode="Append"
)

Combined with Amazon EMR Serverless, which scales compute automatically based on workload demands, you can deploy stateful streaming pipelines without managing clusters or predicting capacity.

Benefits:

Real-time operational visibility: Live streaming state is now accessible externally without interrupting the job, powering dashboards and monitoring systems that reflect current aggregations.
Faster debugging: State values can be queried directly as a DataFrame, making it significantly easier to diagnose production incidents and verify correctness during development.
Better performance at scale: RocksDB checkpointing improvements reduce bottlenecks in high-throughput stateful workloads, improving reliability for long-running streaming jobs.

What’s new in the emr-spark-8.0 release

Beyond the Spark 4.0 capabilities covered in the preceding sections, the emr-spark-8.0 release introduces infrastructure and runtime changes that simplify how you deploy, patch, and manage Amazon EMR workloads. The release focuses exclusively on Spark, reducing the surface area you need to patch and test.

Fewer components to patch and test

The emr-spark-8.0 release includes Apache Spark 4.0, Apache Iceberg 1.10, Apache Hudi 1.0.2, Delta Lake 4.0, and connectors for Amazon DynamoDB, Amazon Kinesis, Amazon Redshift, and Amazon Simple Storage Service (Amazon S3) (via the S3A connector). Apache Livy and JupyterEnterpriseGateway are available as opt-in components on Amazon EMR on EC2. If your workloads require Apache Flink, Trino, Presto, or other execution engines, you can continue to use Amazon EMR 7.x releases.

Simplified patch management

You can specify emr-spark-8.0.x when creating a cluster or application, and Amazon EMR will automatically select the latest patch version. For example, emr-spark-8.0.1, emr-spark-8.0.2, and so on as patches are released. This “.x” wildcard is supported through AWS APIs and AWS Command Line Interface (AWS CLI). On Amazon EMR on EKS and Amazon EMR Serverless, new jobs automatically run on the latest Amazon Linux patches, so you no longer need to track date-based version tags.

Latest Python, Java, and Scala runtimes

The release ships with modernized runtimes: Python 3.11 as the default, with support for Python 3.12 and 3.13. Java 17 is the default, with Java 21 also available. Both are provided through Amazon Corretto. Scala 2.13 is the supported Scala runtime.

A few infrastructure changes to note: AWS SDK for Java v2 replaces v1, bringing improved performance and alignment with the latest AWS APIs. The EMR S3A connector replaces EMR File Systems (EMRFS) for Amazon S3 access, delivering better performance and compatibility with open-source Spark. For shuffle-intensive workloads on Amazon EMR Serverless, enabling Serverless Storage can reduce data processing costs by up to 20%. For more information, see Optimize Amazon EMR Runtime for Apache Spark with EMR S3A for benchmarks, Amazon EMR Serverless eliminates local storage provisioning, and Reducing costs for shuffle-heavy Apache Spark workloads with serverless storage for Amazon EMR Serverless.

Migration and compatibility notes

If you are migrating from Spark 3.5 to Spark 4.0, the Apache Spark upgrade agent for Amazon EMR can accelerate your migration by analyzing existing applications and identifying changes needed for Spark 4.0 compatibility. For more information, see the upgrade guidance.

If your workflows use Apache Pig, Apache Oozie, JupyterHub, Apache Zeppelin, or Hue, you can continue to use Amazon EMR 7.x releases. These components are not included in emr-spark-8.0. For interactive Spark development, use Amazon EMR Studio, with Apache Livy and JupyterEnterpriseGateway available on Amazon EMR on EC2.

For the complete list of supported components and configurations, see the Amazon EMR release guide.

Get started

Spark 4.0 is now available across Amazon EMR on EC2, Amazon EMR on EKS, and Amazon EMR Serverless. To begin, choose your deployment model and follow the relevant getting started guide:

Amazon EMR on EC2 — Getting started with Amazon EMR on EC2.
Amazon EMR Serverless — Getting started with Amazon EMR Serverless.
Amazon EMR on EKS — Getting started with Amazon EMR on EKS.
Amazon EMR release guide — Apache Spark.

Conclusion

Spark 4.0 on Amazon EMR delivers improvements across query validation, semi-structured data handling, Python development, and streaming observability. ANSI SQL mode catches invalid operations at query time rather than silently propagating nulls downstream, and SQL scripting removes the need to context-switch between SQL and Python for complex ETL logic. The VARIANT data type eliminates parsing overhead for semi-structured JSON workloads and can now be stored natively in Iceberg V3 tables with fine-grained access control at the column and row level. Queryable streaming state gives you live visibility into running jobs without interruption, and Spark Connect lets you develop against Amazon EMR from Jupyter notebooks, Amazon SageMaker Unified Studio Data Notebooks, Amazon Q Developer, Kiro, or your preferred IDE without managing cluster connectivity.

Ready to build or migrate? Choose your deployment model from the preceding section and get started today. For detailed guidance, see the Amazon EMR Release Guide and the Amazon EMR Serverless User Guide.

About the authors

Unlock cost savings with incremental snapshot billing for Amazon Redshift Serverless and Amazon Redshift RG

Nidhi Nayak — Mon, 08 Jun 2026 17:24:53 +0000

Amazon Redshift customers rely heavily on snapshots, which are point-in-time backups of their data, for disaster recovery, compliance retention, and data portability across AWS Regions. Amazon Redshift supports two types of snapshots: automated and manual. For provisioned clusters, automated snapshots are enabled by default and retained for up to 35 days; manual snapshots persist until you delete them. For serverless workgroups, Amazon Redshift automatically creates recovery points that are retained for 24 hours, and you can also create manual snapshots with a configurable retention period. For details on snapshot creation and backup storage pricing, you can refer to Amazon Redshift pricing for more details.

Starting June 8, 2026, Amazon Redshift is introducing an incremental snapshot billing model for Amazon Redshift Serverless and Amazon Redshift RG (provisioned instances powered by AWS Graviton). With this enhancement, you pay only for the unique data blocks across your active manual snapshots within your account. This delivers significant cost savings for customers who have multiple snapshots that contain largely identical data blocks.

In this post, you will learn how the new incremental snapshot billing model works, the customer use cases it addresses, and how it helps you optimize costs while improving your Recovery Point Objective (RPO).

Incremental snapshot billing

With this new billing model, Amazon Redshift bills manual snapshots based on unique data blocks. When you take multiple manual snapshots of the same workgroup or cluster, much of the data remains unchanged between snapshots. The billing model recognizes this overlap and charges only for the unique data blocks across your active snapshots. Data that has not changed between snapshots is counted once.

Consider a 10 TB data warehouse with three manual snapshots:

Snapshot 1 (Day 1): Full backup, 10 TB of unique data blocks
Snapshot 2 (seconds later): Nothing changed, shares data blocks with Snapshot 1, no additional charge
Snapshot 3 (two days later): 1 TB of new unique data blocks created from changes
Total billed: 11 TB of unique data blocks

Using this example, customers pay for the 10 TB of unique data blocks in Snapshot 1 plus the 1 TB of new unique data block in Snapshot 3. Snapshot 2 shares its blocks with Snapshot 1, so it adds zero cost. Hence, total 11 TB of unique data blocks are billed.

Key billing model details

With the new incremental snapshot billing model, you are charged only for the unique data blocks at the existing snapshot rates. Following are the key details of the new feature:

Scope: Amazon Redshift Serverless and Amazon Redshift RG instances. Amazon Redshift RA3 instances retains the current tiered S3 billing.
Rate: Based on the existing snapshot pricing for your Region.
Deduplication level: Account-level for Amazon Redshift Serverless and RG.
Automated snapshots: Unchanged, still available at no additional cost (35 days for Provisioned, 1 day for Serverless).
Existing snapshots: Automatically transition to the incremental snapshot billing model. No action required.

This model is especially valuable for customers needing backup retention beyond the automated snapshot windows available at no additional cost. Serverless customers needing backup beyond 24 hours can now take manual snapshots knowing they pay for a unique data block, making extended retention more practical and affordable.

Benefits

With the incremental snapshot billing model, customer can adopt stronger data protection strategies at optimized costs:

Compliance-driven long-term retention

Regulated industries (financial services, healthcare, government, and life sciences) must often retain backups for 90 days to 5+ years. Since this billing model charges only for unique data blocks, retention policies become significantly more affordable as snapshots accumulate.

How this feature helps: You can now maintain backup retention (90-day, 1-year, 7-year) on Amazon Redshift Serverless and RG at optimized cost. A 10 TB warehouse with 5% daily change rate retaining 90 days of daily snapshots pays for ~14.5 TB of unique data blocks total across all snapshots.

Disaster recovery with better Recovery Point and Time Objectives (RPO/ RTO)

Many customers want more frequent snapshots (hourly instead of daily) for tighter recovery objectives. Because each additional snapshot is billed only for its new unique data blocks, frequent backups are practical and affordable.

How this feature helps: You can take hourly snapshots where each one adds only ~0.2% in new unique data (assuming 5% daily change rate). More snapshots mean more recovery points and less data loss in a failure scenario, all at optimized cost.

Cross-Region disaster recovery at lower cost

Snapshots copied to another region for disaster recovery are also billed based on unique data blocks. Organizations maintaining multi-Region disaster recovery (DR) strategies pay proportionally to actual data changes, making geographic redundancy affordable.

How this feature helps: If you are running active-passive or active-active multi-Region architectures, you can copy snapshots across Regions more frequently, improving cross-Region RPO while keeping DR costs proportional to actual data changes rather than full dataset size.

Affordable extended backups

With the incremental snapshot billing model, extended manual backups are more affordable for customers, regardless of their workload size. Even retention policies (7-day, 14-day) cost proportionally to actual data changes, for enhanced data protection posture across the board.

How this feature helps: Customers no longer need to choose between data protection and budget. This billing model helps make extended retention cost effective for workloads of varying sizes.

Pricing example

For example, you have an Amazon Redshift Serverless workgroup with 10 TB of active data in US East (Ohio). You take daily manual snapshots with 7-day retention. Your data changes at 5% per day (0.5 TB/day).

Component	Calculation	Monthly Cost
Active data	10 TB × 1,024 GB/TB × $0.023	$235.52
Unique snapshot blocks (after deduplication)	13 TB × 1,024 GB/TB × $0.023	$306.18
Total		$541.70

Because shared blocks across snapshots are counted only once. You pay for 13 TB of unique snapshot data rather than the full cumulative size of all seven daily snapshots.

Compounding savings on Amazon Redshift RG

If you are evaluating migrating from RA3 to RG, the savings stack significantly. Some of the compounding savings on RG include:

RG instances are priced at 30% discount as compared to RA3 instances.
Reserved Instances (RI) pricing is available for RG which provide further compute savings.
Incremental billing alleviates duplicate snapshot charges for backup storage.
Data lake queries are included in RG compute pricing, thereby avoiding the per-terabyte scanning charges of Amazon Redshift Spectrum.

The combined effect of these options for RG can deliver an aggregate greater than 30% cost reduction over RA3. You can lock in RI pricing on RG clusters for predictable, long-term savings on top of the incremental snapshot benefit.

Getting started

No action is required on your end. Your existing manual snapshots automatically transition to the incremental snapshot billing model on June 8, 2026.

To maximize the benefit:

Review your current snapshot usage in the AWS Billing and Cost Management console.
Increase snapshot frequency. More frequent snapshots now cost proportionally less since each additional snapshot only adds its unique data blocks to your bill.
Extend retention policies. Compliance driven retention (90-day, 1-year, 7-year) is now significantly more affordable.
Evaluate RA3 to RG migration. Consider the 30% compute savings, combined with RI eligibility during RG evaluation for migrating from RA3.
Explore Serverless. The enhanced billing model makes Serverless a cost-effective option for customers who need backup retention beyond the 24-hour automated recovery point window.

Conclusion

The incremental snapshot billing model for Amazon Redshift Serverless and Amazon Redshift RG charges only for unique data blocks across your snapshots. This supports more frequent snapshots for better disaster recovery, affordable long-term compliance retention, and a compelling path to Amazon Redshift Serverless adoption. Combined with Amazon Redshift RG’s 30% compute discount and Reserved Instances, this delivers meaningful total cost savings across your entire Amazon Redshift spend.

Review your snapshot strategy today and share your feedback on AWS re:Post. For full pricing details, visit the Amazon Redshift pricing page.

About the authors

Nidhi Nayak
Nidhi is a Senior Technical Account Manager with AWS, she helps enterprise customers build scalable, high-performance cloud applications and optimize cloud operations. With over a decade of experience in Data Analytics, Nidhi currently focuses on Redshift & Generative AI integration with Redshift.

Raza Hafeez
Raza is a Senior Product Manager, Technical at Amazon Redshift. He has 15+ years of experience building and optimizing enterprise data warehouses and is passionate about making cloud analytics accessible and cost-effective for customers of all sizes.

Sushmita Barthakur
Sushmita is a Senior Data Solutions Architect at AWS, supporting Strategic customers architect their data workloads on AWS. With a background in data analytics, she has extensive experience helping customers architect and build enterprise data lakes, ETL workloads, data warehouses and data analytics solutions, both on-premises and the cloud. Sushmita is based in Florida and enjoys traveling, reading and playing tennis.

Amy Huang
Amy is a Senior Financial Analyst at AWS and a CPA with over 7 years of progressive experience across Strategic Finance, Banking, and Auditing. She specializes in pricing, financial modeling and valuation, and data-driven analysis. Outside of work, she enjoys yoga and hiking.

Migrate JMS applications to Amazon MQ for RabbitMQ with minimal changes

Vinodh Kannan Sadayamuthu — Mon, 08 Jun 2026 15:52:03 +0000

Running JMS applications on on-premises brokers or Apache ActiveMQ requires manual patching cycles, capacity planning for peak loads, and maintaining high availability across multiple data centers. With Amazon MQ version 4 and above, you can migrate your existing JMS applications without rewriting your messaging layer, removing weeks of rewrite work.

This post shows you how to migrate your JMS applications and walks through a complete setup, from creating the broker to sending and receiving messages. You will also see a real-world scenario: migrating an existing Apache ActiveMQ workload to an Amazon MQ broker running RabbitMQ. The post covers configuration changes, monitoring with Amazon CloudWatch, and validation steps to make sure that your migration succeeds.

Amazon MQ version 4 and above includes built-in support for the RabbitMQ JMS Client and the JMS Topic Exchange plugin. The RabbitMQ JMS Client and JMS Topic Exchange plugin work together, allowing your existing JMS applications to connect using familiar JMS APIs. You update the connection factory configuration and broker endpoint. Your business logic, message producers, consumers, and listeners stay exactly as written.

Understanding JMS and AMQP

How the RabbitMQ JMS Client works

Use the RabbitMQ JMS Client to connect your Java application to Amazon MQ. The client translates your JMS API calls (javax.jms or jakarta.jms) into AMQP 0-9-1 messages that the broker understands.

Advanced Message Queuing Protocol (AMQP) defines how messages are formatted and transmitted across the network at the wire level. This means that non-Java services can consume the same messages using native AMQP clients, making the protocol language-agnostic

JMS version support

Migrate at the JMS version that your application already uses. The client supports JMS 1.1, 2.0, and 3.1 (Jakarta Messaging), so you don’t need to upgrade your application code before migrating brokers. The client integrates with Spring Framework and Spring Boot applications without requiring custom bean factories or application context configuration.

Because the JMS abstraction layer sits between your application and the broker, most migrations require only a connection factory change, not a logic rewrite.

RabbitMQ JMS Topic Exchange plugin

Your existing publish/subscribe patterns work without client-side routing logic. The JMS Topic Exchange plugin adds server-side support for JMS topic semantics, handling topic routing and SQL-based message selection directly in the broker.

The plugin handles SQL-based message selection (JMS selectors like OrderType = `Electronics` AND Priority > 5) and topic hierarchies with wildcard pattern matching (* for single level, # for multiple levels). Your application uses standard JMS topic APIs (createTopic(), setMessageSelector()) without additional filtering logic.

Getting started

This walkthrough shows you how to set up Amazon MQ and connect your existing JMS application. You will create a broker, configure the connection factory, and send and receive messages.

Prerequisites

You need an existing JMS application built on Apache ActiveMQ or another JMS provider to migrate. If you don’t have one, you can still follow Steps 1–5 to create a broker and test the connection pattern. Before you begin, confirm that you have the following in place:

An active AWS account
AWS Command Line Interface (AWS CLI) installed. For instructions, see Installing the AWS CLI.
Java 11 or later installed on your local development environment.
An AWS Identity and Access Management (IAM) principal (user or role) with the AmazonMQFullAccess managed policy attached.
Maven or Gradle for dependency management.

Amazon MQ broker charges apply based on instance type and usage. Review the Amazon MQ pricing page before you start.

Step 1: Create an Amazon MQ for RabbitMQ broker

The following command creates a single-instance broker running RabbitMQ 4.2 on an mq.m7g.medium instance.

aws mq create-broker \ 
--broker-name my-rabbitmq-broker \ 
--engine-type rabbitmq \ 
--engine-version 4.2 \ 
--deployment-mode SINGLE_INSTANCE \ 
--host-instance-type mq.m7g.medium \ 
--auto-minor-version-upgrade \ 
--publicly-accessible \ 
--users "Username=admin,Password=[PASSWORD]" \ 
--region us-west-2

Replace <broker-name> with the name that you want to give to the broker. Replace <username> and <password> as described in the create-broker CLI documentation. After the command runs successfully, the command line displays the BrokerArn and BrokerId.

Note: This command creates a publicly accessible broker for demonstration purposes only. For production workloads, create brokers in private subnets within your VPC and restrict access using security groups. Don’t use the –publicly-accessible flag. For more information, see Security best practices for Amazon MQ.

The command returns output similar to:

{
    "BrokerArn": "arn:aws:mq:us-west-2:111122223333:broker:my-rabbitmq-broker:b-c8352341-ec91-4a78-ad9c-a57f23f235bb",
    "BrokerId": "b-c8352341-ec91-4a78-ad9c-a57f23f235bb"
}

Save the BrokerId value for the next step.

The broker takes approximately 15–20 minutes to reach the Running state. Run the following command every 2 minutes to check the status:

aws mq describe-broker --broker-id <BrokerId> --region us-west-2 --query 'BrokerState'

Proceed to the next step after the broker state is RUNNING.

To get the broker endpoints, run:

aws mq describe-broker --broker-id <BrokerId> --region us-west-2 --query 'BrokerInstances'

Note the ConsoleURL and Endpoints from the output. The command returns output similar to:

[{
    "ConsoleURL": "https:// b-c8352341-ec91-4a78-ad9c-a57f23f235bb.mq.us-west-2.on.aws",
    "Endpoints": ["amqps://b-c8352341-ec91-4a78-ad9c-a57f23f235bb.mq.us-west-2.on.aws:5671"]
}]

Step 2: Add the RabbitMQ JMS Client dependency

Choose the dependency that matches your application’s current JMS version. If your imports reference javax.jms packages, use version 2.12.0. If your imports reference jakarta.jms packages (JMS 3.1 / Jakarta EE 9+), use version 3.4.0.

For JMS 1.1 and 2.0 (javax.jms):

<dependency>
<groupId>com.rabbitmq.jms</groupId>
<artifactId>rabbitmq-jms</artifactId>
<version>2.12.0</version>
</dependency>

For JMS 3.1 / Jakarta JMS 3.1 / Jakarta Messaging (jakarta.jms):

<dependency>
<groupId>com.rabbitmq.jms</groupId>
<artifactId>rabbitmq-jms</artifactId>
<version>3.4.0</version>
</dependency>

Step 3: Configure the connection factory

Store your broker credentials in AWS Secrets Manager before configuring the connection factory. This keeps credentials out of your source code and configuration files.

Create the secret:

aws secretsmanager create-secret \
--name dev-rabbitmq \
--description "Amazon MQ broker credentials" \
--secret-string '{"username":"admin","password":"[PASSWORD]"}' \ 
--region us-west-2

Add the AWS SDK for Secrets Manager to your pom.xml:

<dependency> 
<groupId>software.amazon.awssdk</groupId> 
<artifactId>secretsmanager</artifactId> 
<version>2.20.0</version> </dependency>

<dependency> 
<groupId>com.fasterxml.jackson.core</groupId> 
<artifactId>jackson-databind</artifactId> 
<version>2.15.0</version> 
</dependency>

Replace your existing broker URL with the Amazon MQ endpoint. In most cases, this is the only change required in your application configuration:

import com.rabbitmq.jms.admin.RMQConnectionFactory;
import software.amazon.awssdk.services.secretsmanager.SecretsManagerClient;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueRequest;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueResponse;
import software.amazon.awssdk.regions.Region;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;
import javax.jms.*; 

// Retrieve credentials from AWS Secrets Manager
Map<String, String> creds;
try (SecretsManagerClient secretsClient = SecretsManagerClient.builder().region(Region.US_WEST_2).build()) {    
    GetSecretValueResponse response = secretsClient.getSecretValue(GetSecretValueRequest.builder().secretId("dev-rabbitmq").build());        
    ObjectMapper objectMapper = new ObjectMapper();    
    creds = objectMapper.readValue(response.secretString(),new TypeReference<Map<String, String>>() {});
    } 

// Create and configure the connection factory
RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
connectionFactory.setHost("b-c8352341-ec91-4a78-ad9c-a57f23f235bb.mq.us-west-2.on.aws");
connectionFactory.setPort(5671);connectionFactory.setVirtualHost("/");
connectionFactory.useSslProtocol();
connectionFactory.setUsername(creds.get("username"));
connectionFactory.setPassword(creds.get("password"));

Replace the host value with your broker endpoint from Step 1.

The connection factory requires four parameters:

Host: Your broker endpoint from the describe-broker output
Port: 5671 for AMQP over TLS (Amazon MQ requires encryption in transit)
VirtualHost: “/” (the default RabbitMQ virtual host)
UseSslProtocol: true (required by Amazon MQ)

Step 4: Send messages

The following examples show how to send messages to a queue and a topic using the JMS 2.0 simplified API.

Point-to-point (queue) for one-to-one delivery:

try (JMSContext context = connectionFactory.createContext()) {
Queue queue = context.createQueue("orders");
context.createProducer().setProperty("OrderType", "Electronics").send(queue, "Order #12345");
System.out.println("Sent message to queue: orders");
}

Publish/subscribe (topic) for one-to-many broadcast:

try (JMSContext context = connectionFactory.createContext()) {
	Topic topic = context.createTopic("orders.electronics");
	context.createProducer().setProperty("MessageType", "Broadcast").send(topic, "New electronics order received!");
	System.out.println("Published message to topic: orders.electronics");
}

Message properties(OrderType, MessageType) are JMS headers that consumers can use for filtering. These properties become AMQP message headers when transmitted to the broker.

Step 5: Receive messages asynchronously

To receive messages asynchronously, attach a MessageListener to a consumer. The listener fires each time a message arrives.

Queue consumer:

Asynchronous consumers process messages in a background thread without blocking your main application logic. The MessageListener callback fires each time a message arrives, allowing your application to handle messages as they’re delivered rather than polling with receive().

try (JMSContext context = connectionFactory.createContext()) {
	Queue queue = context.createQueue("orders");
	JMSConsumer consumer = context.createConsumer(queue);
	consumer.setMessageListener(message -> {
		if (message instanceof TextMessage) {
			try {
				System.out.println("Received: " + ((TextMessage) message).getText());
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	});

	System.out.println("Listening for messages on queue: orders");

	// Keep the consumer active for 30 seconds
	Thread.sleep(30000);
}

Topic subscriber:

try (JMSContext context = connectionFactory.createContext()) {
	Topic topic = context.createTopic("orders.electronics");
	JMSConsumer consumer = context.createConsumer(topic);
	consumer.setMessageListener(message -> {
		if (message instanceof TextMessage) {
			try {
				System.out.println("Subscriber received: " + ((TextMessage) message).getText());
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	});

	System.out.println("Subscribed to topic: orders.electronics");

	// Keep the consumer active for 30 seconds
	Thread.sleep(30000);}

The Thread.sleep(30000) call keeps the consumer active for 30 seconds.

Use case: Migrating an ActiveMQ Workload to Amazon MQ for RabbitMQ

Migrate your Apache ActiveMQ applications to Amazon MQ by updating four configuration points. Your business logic, message producers, consumers, and listeners stay exactly as written. This walkthrough uses a real JMS 1.1 application with a centralized broker configuration class to show precisely which lines change and which remain identical.

Apache ActiveMQ powers messaging infrastructure for thousands of Java applications worldwide. If you run JMS applications on ActiveMQ, you can migrate to Amazon MQ for RabbitMQ with minimal code changes. The following steps demonstrate a complete migration using an application that includes a centralized broker configuration class, a message producer, and a message consumer.

Step 1: Update the Maven dependency

Replace the ActiveMQ client dependencies with the RabbitMQ JMS client in your pom.xml. The rabbitmq-jms artifact includes the RabbitMQ AMQP client and JMS API as transitive dependencies, so a single entry replaces both ActiveMQ artifacts.

Before (ActiveMQ):

<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>5.18.6</version>
</dependency>

<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.18.6</version>n>5.18.6</version>
</dependency>

After (Amazon MQ):

<dependency>
<groupId>com.rabbitmq.jms</groupId>
<artifactId>rabbitmq-jms</artifactId>
<version>2.12.0</version>
</dependency>

The rabbitmq-jms artifact pulls in the RabbitMQ AMQP client and the JMS API as transitive dependencies, so a single entry replaces both ActiveMQ artifacts.

Step 2: Update the broker configuration

If your application centralizes connection details in a shared configuration class, that class is the only file that needs to change. The queue name and everything else your application references remain the same.

Before (ActiveMQ):

// BrokerConfig.java - ActiveMQ version
public final class BrokerConfig {
	// OpenWire endpoint
	public static final String BROKER_URL = "tcp://localhost:61616";
	public static final String USERNAME = "[PASSWORD]";
	public static final String PASSWORD = "[PASSWORD]";
	public static final String QUEUE_NAME = "demo.queue";
	private BrokerConfig() {}}

After (Amazon MQ):

// BrokerConfig.java - Amazon MQ version

import com.fasterxml.jackson.core.type.TypeReference;import com.fasterxml.jackson.databind.ObjectMapper;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.secretsmanager.SecretsManagerClient;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueRequest;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueResponse;
import java.util.Map;

public final class BrokerConfig {
	// AMQPS endpoint (TLS required by Amazon MQ)
	public static final String BROKER_URL = "amqps://b-c8352341-ec91-4a78-ad9c-a57f23f235bb.mq.us-west-2.on.aws:5671";

	// Queue name carries over unchanged
	public static final String QUEUE_NAME = "demo.queue";

	// Secret name in AWS Secrets Manager
	private static final String SECRET_ID = "dev-rabbitmq";
	private static final Map<String, String> CREDENTIALS = loadCredentials();
	public static String getUsername() {return CREDENTIALS.get("username");}
	public static String getPassword() {return CREDENTIALS.get("password");}
	private static Map<String, String> loadCredentials() { 
		try (SecretsManagerClient client = SecretsManagerClient.builder().region(Region.US_WEST_2).build()) {
			GetSecretValueResponse response = client.getSecretValue(GetSecretValueRequest.builder().secretId(SECRET_ID).build());
			ObjectMapper mapper = new ObjectMapper();
			return mapper.readValue(response.secretString(), new TypeReference<Map<String, String>>() {});
		} catch (Exception e) {
			throw new RuntimeException("Failed to load broker credentials from Secrets Manager", e);
		}
	}
	private BrokerConfig() {}}

Two things changed in this file compared to the ActiveMQ version: the protocol prefix (tcp:// to amqps://) and the host and port (OpenWire on 61616 to AMQP over TLS on 5671). The queue name is identical. Credentials are no longer stored as static string constants. Instead, loadCredentials() retrieves them from AWS Secrets Manager at startup, and getUsername() and getPassword() expose them to the rest of the application. This follows AWS security best practices and streamlines credential rotation.

Step 3: Update the message producer

The producer requires two changes: the import statement and the factory instantiation. Every JMS API call after the factory (createConnection, createSession, createProducer, send) is identical to the ActiveMQ version.

Before (ActiveMQ):

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class MessageProducer {
	public static void main(String[] args) {
		Connection connection = null;
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BrokerConfig.USERNAME,BrokerConfig.PASSWORD,BrokerConfig.BROKER_URL);
			connection = factory.createConnection();
			connection.start();
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(BrokerConfig.QUEUE_NAME);
			javax.jms.MessageProducer producer = session.createProducer(destination);
			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
			for (int i = 1; i <= 5; i++) {
				TextMessage message = session.createTextMessage("Hello from ActiveMQ - message #" + i);
				producer.send(message);
				System.out.println("Sent: " + message.getText());
			}
			producer.close();
			session.close();
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try { 
					connection.close();
				} catch (JMSException ignored) {}}
			}
		}
	}

After (Amazon MQ):

import com.rabbitmq.jms.admin.RMQConnectionFactory;
import javax.jms.*;

public class MessageProducer {

	public static void main(String[] args) {
		Connection connection = null;
		try {
			RMQConnectionFactory factory = new RMQConnectionFactory();
			factory.setUri(BrokerConfig.BROKER_URL);
			factory.setUsername(BrokerConfig.getUsername());
			factory.setPassword(BrokerConfig.getPassword());

			// Everything below this line is identical to the ActiveMQ version
			connection = factory.createConnection();
			connection.start();
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(BrokerConfig.QUEUE_NAME);
			javax.jms.MessageProducer producer = session.createProducer(destination);
			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
			for (int i = 1; i <= 5; i++) {
				TextMessage message = session.createTextMessage("Hello from Amazon MQ - message #" + i);
				producer.send(message);
				System.out.println("Sent: " + message.getText());
			}
			producer.close();
			session.close();
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException ignored) {}
			}
		}
	}
}

The import changes from org.apache.activemq.ActiveMQConnectionFactory to com.rabbitmq.jms.admin.RMQConnectionFactory. The factory construction switches from a constructor that accepts credentials and URL to a no-arg constructor with explicit setter calls. Credentials are now retrieved from AWS Secrets Manager through BrokerConfig.getUsername() and BrokerConfig.getPassword(). That is the complete change set for the producer.

Step 4: Update the message consumer

The consumer follows the same pattern as the producer. Swap the factory class and import, update the credential calls, and keep everything else.

Before (ActiveMQ):

import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class MessageConsumer {
	public static void main(String[] args) {
		Connection connection = null;
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BrokerConfig.USERNAME, BrokerConfig.PASSWORD, BrokerConfig.BROKER_URL);
			connection = factory.createConnection();
			connection.start();
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(BrokerConfig.QUEUE_NAME);
			javax.jms.MessageConsumer consumer = session.createConsumer(destination);
			System.out.println("Waiting for messages on queue: " + BrokerConfig.QUEUE_NAME);
			Message message;
			while ((message = consumer.receive(10000)) != null) {
				if (message instanceof TextMessage) {
					TextMessage textMessage = (TextMessage) message;
					System.out.println("Received: " + textMessage.getText());
				}
			}
			consumer.close();
			session.close();
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try { 
					connection.close(); 
				} catch (JMSException ignored) {}
			}
		}
	}
}

After (Amazon MQ):

import com.rabbitmq.jms.admin.RMQConnectionFactory;
import javax.jms.*;

public class MessageConsumer {
	public static void main(String[] args) {
		Connection connection = null;
		try {
			RMQConnectionFactory factory = new RMQConnectionFactory();
			factory.setUri(BrokerConfig.BROKER_URL);
			factory.setUsername(BrokerConfig.getUsername());
			factory.setPassword(BrokerConfig.getPassword());

			// Everything below this line is identical to the ActiveMQ version
			connection = factory.createConnection();
			connection.start();
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(BrokerConfig.QUEUE_NAME);
			javax.jms.MessageConsumer consumer = session.createConsumer(destination);
			System.out.println("Waiting for messages on queue: " + BrokerConfig.QUEUE_NAME);
			Message message;
			while ((message = consumer.receive(10000)) != null) {
				if (message instanceof TextMessage) {
					TextMessage textMessage = (TextMessage) message;
					System.out.println("Received: " + textMessage.getText());
				}
			}
			consumer.close();
			session.close();
		} catch (JMSException e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try { 
					connection.close(); 
				} catch (JMSException ignored) {}
			}
		}
	}
}

The import changes from org.apache.activemq.ActiveMQConnectionFactory to com.rabbitmq.jms.admin.RMQConnectionFactory. The factory construction switches to a no-arg constructor with explicit setter calls, and BrokerConfig.USERNAME / BrokerConfig.PASSWORD are replaced with BrokerConfig.getUsername() / BrokerConfig.getPassword(). The session creation, queue lookup, consumer setup, and message processing loop are identical to the ActiveMQ version.

Configuration

The following table summarizes the changes required when migrating from Apache ActiveMQ.

	ActiveMQ	Amazon MQ for RabbitMQ
Maven dependency	activemq-client 5.18.6	rabbitmq-jms 2.12.0
Connection factory class	ActiveMQConnectionFactory	RMQConnectionFactory
Import package	org.apache.activemq	com.rabbitmq.jms.admin
Broker URL format	tcp://host:61616	amqps://broker-id.mq.region.on.aws:5671
Protocol	OpenWire	AMQP 0-9-1
Port	61616 (OpenWire)	5671 (AMQP over TLS)
TLS	Optional	Required
Credentials	Plain text / JNDI	AWS Secrets Manager (recommended)
Virtual host	N/A	/ (default)
JMS version support	JMS 1.1	JMS 1.1, 2.0, 3.1 (Jakarta)
Queue/Topic names	demo.queue	demo.queue (no change)
JMS API calls	Standard JMS 1.1	Standard JMS 1.1 (no change)

Validating the migration

Run your application against Amazon MQ for RabbitMQ in a staging environment before directing production traffic to the new broker. Verify that messages flow correctly, consumers process as expected, and no data loss occurs during cutover.

The RabbitMQ management console provides real-time visibility into broker operations. Access it through the ConsoleURL from your broker details. The console shows queue depths, consumer counts, and message rates. Use it during testing to identify routing or throughput issues before production deployment

The console displays jms.durable.queues and jms.durable.topic exchanges. The JMS client creates these automatically when your application creates queues and topics, so no manual exchange configuration is required.

Monitoring with Amazon CloudWatch

Amazon MQ publishes broker metrics to Amazon CloudWatch with no additional configuration needed. This gives you persistent monitoring and alerting that works alongside the rest of your AWS observability setup, beyond what the RabbitMQ management console provides in real time.

After your JMS messages reach the Amazon MQ for RabbitMQ broker, they’re transported as AMQP messages, which means standard RabbitMQ operational best practices apply. Keep queue depth low to avoid memory pressure and consumer lag. Follow message durability and reliability guidelines to prevent message loss during broker restarts. For connection management, review broker setup and connection best practices to avoid connection churn.

Set Amazon CloudWatch alarms on MessageCount and ConnectionCount first. A rising queue depth with a stable or dropping consumer count is an early signal of a processing bottleneck. A sudden drop in connections can indicate a client configuration issue that’s more straightforward to catch before it affects production traffic.

Clean up

To avoid ongoing charges after testing, delete the Amazon MQ broker and Secrets Manager secret using the AWS CLI.

Delete the broker:

aws mq delete-broker --broker-id <your-broker-id> --region us-west-2

Delete the Secrets Manager secret:

aaws secretsmanager delete-secret \
--secret-id dev-rabbitmq \
--force-delete-without-recovery \
--region us-west-2

Broker deletion is permanent and can’t be undone. Amazon MQ removes all messages, configurations, and user credentials. Leaving the broker running incurs hourly charges based on the instance type, plus storage costs for message data retained on the broker.

Conclusion

In this post, we walked you through how to migrate your JMS applications. We also walked through a complete setup, from creating the broker to sending and receiving messages. Migrating the broker is the straightforward part. The more significant question is what you do next. After your JMS application is running on Amazon MQ for RabbitMQ, you have access to native AMQP clients, which means non-Java services can start consuming the same messages without a JMS layer. A Java-centric messaging system becomes a shared event backbone that service can participate in. The migration is a starting point, not just a lift-and-shift.

Next Steps

To get started with your migration, create your first Amazon MQ for RabbitMQ broker. For detailed technical guidance, see the Amazon MQ Developer Guide and explore the RabbitMQ JMS Client documentation.

About the authors

Query Amazon Redshift using natural language with Kiro

Hitesh Dodiya — Thu, 04 Jun 2026 22:06:00 +0000

It’s Monday morning and your VP pings you: “Revenue dropped 15 percent over the weekend. What happened?” The clock starts. You open the AWS Management Console, find the right Amazon Redshift cluster, open the query editor, and start hunting. Which database has the revenue data, analytics_db or reporting_db? Is the table called orders, transactions, or sales_events? You find it, but now you need the schema. Is the amount column total_amount, revenue, or order_value? 20 minutes in, you haven’t answered the question. You’ve been navigating infrastructure.

This scenario plays out daily across data teams, and it’s why the landscape is shifting. With AI agents entering the analytics workflow, a growing number of business users can now perform complex data analysis. They no longer need to file a ticket with the data engineering team and wait days for a response. The bottleneck is no longer SQL expertise. It’s the friction between having a question and getting an answer.

The Amazon Redshift MCP server paired with Kiro removes that friction. Instead of memorizing cluster endpoints, reverse-engineering schemas, and hand-writing SQL, you describe what you need in plain text and get results. That Monday morning question becomes a single sentence: “Show me daily revenue for the past two weeks, broken down by region.” Kiro finds the cluster, discovers the schema, writes the query, and returns the answer in seconds, not minutes.

In this post, you learn how to:

Install and configure Kiro with the Amazon Redshift MCP server.
Discover clusters, databases, and schemas using natural language.
Run analytical queries and cross-cluster comparisons conversationally.
Implement security best practices for production Amazon Redshift environments.

You can use Kiro in two forms: Kiro integrated development environment (IDE), a full desktop development environment, and Kiro command line interface (CLI), which brings the same AI capabilities directly to your terminal. The Redshift MCP server works with both. The CLI experience is particularly well suited for the conversational analytics workflow this post describes, because you can start querying your data warehouse from a terminal session without opening an IDE.

Important: Before using this integration with production Amazon Redshift environments, read the Security tips section. This section covers critical considerations around AWS Identity and Access Management (IAM) permissions and Kiro autonomy modes.

What is the Amazon Redshift MCP server?

The Model Context Protocol (MCP) is an open standard that provides AI agents with secure connections to external data sources and tools. The Amazon Redshift MCP server is an open source implementation that bridges the Kiro AI agent with your Amazon Redshift infrastructure.

With the Redshift MCP server, you can:

Automatically find both provisioned clusters and serverless workgroups with cluster discovery.
Browse databases, schemas, tables, and columns with metadata exploration.
Run SQL in READ ONLY mode with built-in safety protections with safe query execution.
Work with multiple clusters and workgroups simultaneously with multi-cluster support.

The server translates your natural language requests into the appropriate Amazon Redshift Data API calls and SQL queries. No manual endpoint configuration or SQL writing is required.

How the Redshift MCP server relates to the AWS MCP server

You might have noticed that AWS also offers the AWS MCP server (part of the Agent Toolkit for AWS), which provides broad access to AWS services, including the Redshift Data API. A common question is: if the AWS MCP server can already reach Redshift, why add a dedicated Redshift MCP server?

The two are complementary, not competing. The AWS MCP server gives Kiro general AWS capabilities (service decision guides, SDK usage guidance, troubleshooting skills, and access to AWS APIs). The Redshift MCP server adds a purpose-built analytics layer on top. It provides single-call query execution (compared to a minimum of three API calls for submit, poll, and fetch), read-only safety by default, transparent provisioned and serverless cluster handling, and dedicated metadata navigation tools. Upcoming features like query plan explanation, native identity propagation, cluster analysis, and UDF discovery will further extend this specialized layer.

You can use both together, or use the Amazon Redshift MCP server on its own. There’s no either-or requirement.

Setting it up

The following sections walk you through the installation and configuration process.

Prerequisites

Before you begin, make sure that you have:

On your machine:

Kiro IDE or Kiro CLI installed.
Python 3.10 or newer.
The uv package manager from Astral.

On AWS:

AWS credentials configured through the AWS Command Line Interface (AWS CLI), environment variables, or IAM roles.
At least one Amazon Redshift provisioned cluster or serverless workgroup.
IAM permissions for Amazon Redshift access (see the following section).

Step 1: Install the uv package manager

If you don’t have uv installed, run one of the following commands.

For macOS or Linux:

curl -LsSf https://astral.sh/uv/install.sh | sh

For Windows PowerShell:

powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

Then install Python 3.10 or newer if needed:

uv python install 3.10

Step 2: Configure IAM permissions

Your AWS identity needs the following permissions. Attach this policy to your IAM user or role:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "redshift:DescribeClusters",
        "redshift:GetClusterCredentialsWithIAM",
        "redshift:GetClusterCredentials",
        "redshift-serverless:ListWorkgroups",
        "redshift-serverless:GetWorkgroup",
        "redshift-serverless:GetCredentials",
        "redshift-data:ExecuteStatement",
        "redshift-data:DescribeStatement",
        "redshift-data:GetStatementResult"
      ],
      "Resource": "*"
    }
  ]
}

You also need database-level permissions: SELECT on tables you want to query, USAGE on schemas you want to explore, and connection access to the target databases.

Step 3: Configure the MCP server in Kiro

Open (or create) your Kiro MCP configuration file and add the Amazon Redshift server.

For Kiro IDE:

User-level configuration (applies globally): ~/.kiro/settings/mcp.json.
Workspace-level configuration (applies to a specific project): .kiro/settings/mcp.json.

For Kiro CLI:

User-level configuration: ~/.kiro/settings/mcp.json.
Workspace-level configuration: .kiro/settings/mcp.json in your project directory.

The configuration format is the same for both. Add the following:

{
  "mcpServers": {
    "awslabs.redshift-mcp-server": {
      "command": "uvx",
      "args": ["awslabs.redshift-mcp-server@latest"],
      "env": {
        "AWS_PROFILE": "default",
        "AWS_DEFAULT_REGION": "us-east-1",
        "FASTMCP_LOG_LEVEL": "ERROR"
      },
      "disabled": false,
      "autoApprove": []
    }
  }
}

Tip: Replace AWS_PROFILE and AWS_DEFAULT_REGION with values matching your environment. If you use AWS_REGION instead of AWS_DEFAULT_REGION, AWS_REGION takes higher precedence.

Note: Windows users: use uv instead of uvx with additional args. See the Amazon Redshift MCP server documentation for the Windows-specific configuration.

Step 4: Verify the connection

In Kiro IDE, restart Kiro (or reconnect MCP servers from the MCP Server view in the Kiro feature panel). Then open the Kiro chat and enter the following prompt:

List my available Redshift clusters

If everything is configured correctly, Kiro discovers your clusters and lists them with details like cluster identifiers, status, node types, and endpoints.

Practical examples

The following scenarios show how Kiro and the Amazon Redshift MCP server work together in real-world workflows.

Tip: If you want Kiro to show the SQL that it executes alongside every result, tell it at the beginning of your session:

For every query you run against Amazon Redshift, show me the exact SQL you executed along with the results.

Kiro follows this instruction for the rest of the conversation. For a permanent setup, add this rule to a steering file (see the Tips section that follows) so you don’t need to repeat it. The following examples reflect this behavior.

Example 1: Discover and explore your infrastructure

Without Kiro: Log in to the AWS Management Console, navigate to Amazon Redshift, choose clusters, open the query editor, sign in to each database, and manually browse schemas.

With Kiro:

What Amazon Redshift clusters and serverless workgroups do I have?

Kiro responds with a structured overview:

Found two provisioned clusters and one serverless workgroup:

analytics-cluster (provisioned, 2x ra3.xlplus nodes, available)
reporting-cluster (provisioned, 4x ra3.4xlarge nodes, available)
etl-workgroup (serverless, available)

Then drill deeper:

What databases and schemas exist in analytics-cluster?

Kiro uses the MCP metadata tools to explore the cluster. No SQL is needed for discovery:

Database: analytics_db (local)

information_schema — system schema
pg_catalog — system schema
public — default schema
sales — user-created schema

Database: dev (local)

information_schema — system schema
pg_catalog — system schema
public — default schema

The only user-created schema is sales in the analytics_db database.

Show me the tables in the sales schema of the analytics database

There are five base tables in analytics_db.sales:

customers
order_items
orders
products
returns

Describe the columns in the orders table

Column	Type	Nullable
`order_id`	INTEGER	NO
`customer_id`	INTEGER	NO
`order_date`	TIMESTAMP	NO
`total_amount`	numeric(12,2)	NO
`status`	VARCHAR(20)	YES

Each request builds on the previous context. Kiro remembers which cluster you’re exploring and navigates the hierarchy for you.

Example 2: On-demand data analysis in plain English

Without Kiro: Write SQL from scratch, guess at column names, run the query, fix syntax errors, iterate.

With Kiro:

Show me the top 10 customers by total purchase amount from the analytics-cluster,
including how many orders each customer placed

Kiro does the following:

Identifies the relevant cluster and database.
Explores the schema to find customer and order tables.
Understands the table relationships.
Generates and runs the appropriate SQL, showing you the query first.
Presents the results in a readable format.

The response includes the query and results:

Query executed:

SELECT
    c.customer_id,
    c.customer_name,
    COUNT(o.order_id) AS total_orders,
    SUM(o.total_amount) AS total_spent
FROM sales.customers c
JOIN sales.orders o ON c.customer_id = o.customer_id
GROUP BY c.customer_id, c.customer_name
ORDER BY total_spent DESC
LIMIT 10;

customer_id	customer_name	total_orders	total_spent
4821	Acme Corp	347	$1,284,500.00
1190	Globex Inc	289	$1,102,340.00
7734	Initech LLC	256	$987,210.00
…	…	…	…

You didn’t need to remember table names, column names, or join conditions. Kiro figured it out by exploring the metadata first. You can see exactly what it ran, copy the SQL, and reuse it.

Example 3: Cross-cluster data comparison

Without Kiro: Open two SQL clients, sign in to different clusters, run similar queries, manually compare results in a spreadsheet.

With Kiro:

Compare the total sales revenue and transaction counts between
analytics-cluster and etl-workgroup for the last quarter.
Are there any discrepancies?

Kiro queries both environments, aligns the results, and provides a comparison, showing every query it runs along the way.

Query executed on analytics-cluster:

SELECT
    COUNT(*) AS total_records,
    SUM(total_amount) AS total_revenue,
    COUNT(CASE WHEN order_date >= DATE_TRUNC('quarter', CURRENT_DATE - INTERVAL '3 months')
        THEN 1 END) AS q4_transactions
FROM sales.orders;

Query executed on etl-workgroup:

SELECT
    COUNT(*) AS total_records,
    SUM(total_amount) AS total_revenue,
    COUNT(CASE WHEN order_date >= DATE_TRUNC('quarter', CURRENT_DATE - INTERVAL '3 months')
        THEN 1 END) AS q4_transactions
FROM sales.orders;

Sales data comparison: Analytics vs ETL.

Metric	Analytics Cluster	ETL Workgroup	Status
Total Records	172,456	172,456	Match
Total Revenue	$110,765,431	$110,765,431	Match
Q4 Transactions	38,061	38,061	Match

Because both queries are visible, you can verify that Kiro compared the same metrics on both sides, with no hidden assumptions.

Example 4: Schema documentation on demand

Without Kiro: Manually document table structures in a wiki or spreadsheet. Documentation frequently falls out of date.

With Kiro:

Generate documentation for the tables in the sales schema of analytics-cluster.
Include column names, data types, and notable relationships.

Kiro explores every table using the discovery tools and produces structured documentation.

Table: sales.customers

Column	Type	Nullable	Description
`customer_id`	INTEGER	NO	Primary key
`customer_name`	VARCHAR(255)	NO
`email`	VARCHAR(255)	YES
`created_at`	TIMESTAMP	NO

Table: sales.orders

Column	Type	Nullable	Description
`order_id`	INTEGER	NO	Primary key
`customer_id`	INTEGER	NO	FK to customers
`order_date`	TIMESTAMP	NO
`total_amount`	numeric(12,2)	NO
`status`	VARCHAR(20)	YES

(continues for all tables in the schema…)

You can paste this directly into your team wiki or README, and anyone can rerun the same discovery flow to refresh the docs later.

Example 5: Troubleshooting data quality

Check the orders table in analytics-cluster for any NULL values
in the customer_id or total_amount columns.
How many records are affected?

Query executed:

SELECT
    COUNT(*) FILTER (WHERE customer_id IS NULL) AS null_customer_ids,
    COUNT(*) FILTER (WHERE total_amount IS NULL) AS null_total_amounts,
    COUNT(*) AS total_rows
FROM sales.orders;

Out of 172,456 rows: 0 NULL customer_ids, 3 NULL total_amounts.

Find duplicate order_ids in the sales.orders table and show me
a few examples with their timestamps

Query executed:

SELECT order_id, COUNT(*) AS occurrences, MIN(order_date) AS first_seen, MAX(order_date) AS last_seen
FROM sales.orders
GROUP BY order_id
HAVING COUNT(*) > 1
ORDER BY occurrences DESC
LIMIT 5;

Found two duplicate order_ids: order 88412 (two occurrences), order 91003 (two occurrences).

What's the date range of data in the events table?
Are there any gaps in the daily record counts?

Query executed:

WITH daily_counts AS (
    SELECT DATE(event_date) AS day, COUNT(*) AS record_count
    FROM sales.events
    GROUP BY DATE(event_date)
),
date_range AS (
    SELECT MIN(day) AS start_date, MAX(day) AS end_date FROM daily_counts
),
all_dates AS (
    SELECT (start_date + (n || ' days')::INTERVAL)::DATE AS day
    FROM date_range, generate_series(0, (end_date - start_date)) AS n
)
SELECT a.day AS missing_date
FROM all_dates a
LEFT JOIN daily_counts d ON a.day = d.day
WHERE d.day IS NULL
ORDER BY a.day;

Date range: 2024-01-01 to 2025-04-20. Found three missing dates: 2024-03-15, 2024-07-04, 2024-12-25 (likely holidays).

Every query is right there in the response. You can copy them into your own SQL client, modify them, or save them as reusable scripts.

Tips for getting the most out of Kiro and Redshift

Start with discovery. Begin each session by asking Kiro to list your clusters and explore the database structure. This gives the agent context for subsequent queries.
Be specific about which cluster. If you have multiple clusters, mention the cluster name in your request to avoid ambiguity.
Iterate gradually. Start with simple questions and build complexity. Ask for a count before asking for a full breakdown.
Use steering files for team conventions. Create a .kiro/steering/redshift.md file in your project with details about your cluster naming conventions, important schemas, and common query patterns. This gives Kiro persistent context about your environment.

Example steering file:

---
inclusion: auto
---

# Redshift Environment Context

## Clusters

- **analytics-cluster**: Primary analytics warehouse. Use database `analytics_db`.
- **etl-workgroup**: Serverless workgroup for ETL pipelines. Use database `staging_db`.

## Key Schemas

- `sales`: Customer transactions, orders, and revenue data
- `marketing`: Campaign performance and attribution data

## Conventions

- Always filter by `is_deleted = false` on soft-delete tables
- Date columns use `TIMESTAMP WITHOUT TIME ZONE` in UTC

## Query Transparency

- Always show the exact SQL query text being executed before or alongside the results.
  Users should be able to see, verify, and reuse every query that Kiro runs against Redshift.

That last Query Transparency rule is a small addition with a big impact. By default, Kiro might summarize results without showing the underlying SQL. Adding this steering instruction makes every query visible, which helps maintain consistent behavior across your data team and supports auditing, learning, and trust.

Use hooks for automation. Set up agent hooks to run common validation queries automatically. For example, trigger a data quality check whenever you edit a specific SQL file.
Verify important results. Always cross-check critical business findings with stakeholders before acting on them.

Security tips

When configured with least privilege IAM policies and Supervised mode, the Redshift MCP server provides multiple layers of protection. Under the AWS shared responsibility model, you are responsible for configuring access controls appropriately.

Safety layers at a glance

Data stays in your account. The MCP server runs locally, queries run inside your Amazon Redshift cluster, and no data is sent to third-party services.
Content not used for training. Enterprise users are excluded from service improvement usage. Free or individual tier users can opt out in Kiro settings.
Read-only by default. Every query is wrapped in a read-only transaction, preventing accidental writes.
Standard IAM controls. Kiro only gets permissions you’ve explicitly granted.
Supervised mode. Review each agent action before it’s applied in production environments.

Important caveat: The read-only enforcement only applies to queries routed through the MCP server’s execute_query tool. Kiro also has shell access. If your IAM credentials have write permissions, direct CLI calls (aws redshift-data execute-statement) bypass this guardrail entirely.

Takeaway: Use least privilege IAM policies (scoped to read and describe operations only) as your primary defense. Avoid broad policies like AdministratorAccess or AmazonRedshiftFullAccess. When IAM is properly scoped, even a direct CLI call cannot perform writes.

Verify Kiro’s output

Kiro is a powerful assistant, but it’s not infallible. Like any AI tool, it can misinterpret your intent, generate incorrect SQL, or present results that look plausible but are wrong. Examples include a misplaced join, a wrong filter, or an aggregation that silently excludes rows. This is especially important when working with production data where decisions have real business impact.

Treat Kiro’s output as a strong starting point, not a final answer. Review the SQL it generates before acting on the results. Use the Query Transparency steering rule (described in the Tips section) so you can see the exact query behind every result. When findings inform business decisions, validate them independently by running the query yourself, cross-checking with a colleague, or comparing against a known baseline.

Summary

Layer	What it protects against
MCP server read-only mode	Accidental writes through the MCP execute_query tool
Least privilege IAM policy	Write operations via any path, including direct CLI calls
Kiro Supervised mode	Unreviewed autonomous actions by the agent

Defense in depth: use these layers together for production environments.

What you can achieve with Kiro and Amazon Redshift

Before	Now you can
Switch between the console, SQL clients, and documentation	Use one interface for discovery, querying, and analysis
Memorize cluster endpoints, database names, and schemas	Ask in plain text and let Kiro discover the structure
Write SQL from scratch for every query	Describe what you want and get results
Manually compare data across clusters	Run single-sentence cross-cluster analysis
Schema documentation is frequently stale	Generate fresh docs on demand
Onboarding new analysts takes days	New team members can explore immediately

Every minute you spend hunting for a table name or debugging a SQL syntax error is a minute that you’re not spending on actual analysis. You can reduce that overhead by letting Kiro handle the mechanical parts (discovery, navigation, and query construction) so you can focus on the questions that matter to your business.

Cleaning up

If you created Amazon Redshift resources specifically for this walkthrough, or if you no longer need the MCP server integration, follow these steps. They help you avoid ongoing charges and remove the configuration.

Remove the MCP server configuration.
Detach the IAM policy.
Delete test Amazon Redshift resources (if applicable).
Uninstall uv (optional).

Conclusion

In this post, you learned how to set up Kiro with the Amazon Redshift MCP server to query your data warehouse using natural language. You explored cluster discovery, schema browsing, analytical queries, cross-cluster comparisons, and data quality checks, all without writing SQL from scratch or switching between tools.

To go further:

New to Amazon Redshift? Get started with Amazon Redshift to create your first cluster or serverless workgroup.
Read the MCP protocol specification to understand how AI agents work with external tools.
Visit kiro.dev for Kiro’s full capabilities, including specs, hooks, and steering files.

As you get comfortable with the basics, try combining steering files with agent hooks to automate recurring workflows like daily data quality checks or weekly schema documentation refreshes.

About the author

Build governance dashboards for Amazon SageMaker Catalog with Amazon Quick

Steve Phillips — Thu, 04 Jun 2026 15:41:17 +0000

Maintaining visibility into your data catalog’s health requires more than ad-hoc queries. Data stewards and compliance teams need automated dashboards that surface governance metrics and alert them when issues arise. These issues include undocumented assets, missing ownership, and stale metadata.

In a previous post, we showed you how to query Amazon SageMaker Catalog metadata using SQL by using the metadata export feature. This post builds on that foundation by demonstrating how to create governance dashboards with Amazon Quick.

Amazon Quick is an agentic AI-powered digital workspace that provides integrated analytics, automation, and research capabilities. With Amazon Quick Sight, a component of Amazon Quick, you can create interactive dashboards and visualizations with automatic chart suggestions and machine learning (ML) insights.

We walk through how to connect Amazon Quick Sight to your Amazon SageMaker Catalog metadata and build governance dashboards using natural language prompts.

Solution overview

This solution extends the metadata export architecture by adding a visualization layer:

Amazon SageMaker Catalog exports asset metadata daily to Amazon Simple Storage Service (Amazon S3) Tables
Amazon Athena queries the metadata using standard SQL
Amazon Quick Sight connects to Athena for interactive dashboards
Amazon Quick uses natural language to build visualizations

Figure 1 – Amazon SageMaker Catalog governance dashboard architecture

Prerequisites

Before you begin, complete the following steps from Analyzing your data catalog: Query SageMaker Catalog metadata with SQL. You must also have the following:

Amazon SageMaker Catalog metadata export enabled
Amazon Athena configured with query results S3 bucket
AWS Lake Formation permissions configured for AWS Identity and Access Management (IAM)-based access
Verified that the asset_metadata.asset table contains data

Additionally, you need:

Amazon Quick Sight subscription (Standard or Enterprise edition)
AWS Identity and Access Management permissions to create Amazon Quick Sight datasets and dashboards

Building a governance dashboard with Amazon Quick Sight

To visualize catalog health metrics, connect Amazon Quick Sight to your Athena metadata tables.

Configure Amazon Quick Sight permissions

Grant permissions to the Amazon Quick Sight service role.

The Amazon Quick Sight service role (default name: aws-quicksight-service-role-v0) needs permissions to access Amazon S3 Tables and AWS Glue catalog:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3tables:GetTableBucket",
        "s3tables:GetTable",
        "s3tables:GetTableMetadataLocation"
      ],
      "Resource": "arn:aws:s3tables:REGION:ACCOUNT_ID:bucket/aws-sagemaker-catalog/*"
    },
    {
      "Effect": "Allow",
      "Action": "glue:GetCatalog",
      "Resource": "arn:aws:glue:REGION:ACCOUNT_ID:catalog"
    }
  ]
}

Add this as an inline policy to the Amazon Quick Sight service role in the IAM console.

Grant AWS Lake Formation permissions:

Both the Amazon Quick Sight service role and your Amazon Quick Sight admin user need AWS Lake Formation permissions on the S3 Tables catalog. First, find your Amazon Quick Sight admin user ARN by running this AWS Command Line Interface (AWS CLI) command:

aws quicksight list-users \
  --aws-account-id ACCOUNT_ID \
  --namespace default \
  --region us-east-1

Amazon Quick Sight users are managed in the Amazon Quick Sight home AWS Region (us-east-1).To grant permissions, use the Lake Formation console.

Navigate to AWS Lake Formation in the AWS Management Console.
Select Data permissions and Grant.
For Principals, choose SAML users and groups.
Enter your Amazon Quick Sight admin user ARN (from the preceding command).
Under LF-Tags or catalog resources, choose Named Data Catalog resources.
For Catalogs, choose the S3 Tables catalog: ACCOUNT_ID:s3tablescatalog/aws-sagemaker-catalog.
For Databases, choose asset_metadata.
Under Tables, choose asset.
For Table permissions, choose Select and Describe.
Select Grant.

Figure 2 – Grant access to Amazon SageMaker Catalog resources

Repeat steps 1–9 for the Amazon Quick Sight service role, but in step 2 choose IAM users and roles instead.

When choosing the catalog in the Lake Formation console, you must choose the full S3 Tables catalog identifier (ACCOUNT_ID:s3tablescatalog/aws-sagemaker-catalog) to see the asset_metadata database.

Create an Amazon Quick Sight dataset.

Access S3 Tables data by creating a Quick Sight dataset using an Amazon Athena data source and the custom SQL option. An S3 Tables data source is also available but requires additional permissions. See Introducing new data source with S3 Tables in Amazon Quick for using S3 Tables as an Amazon Quick data source.

Open Amazon Quick Sight in the AWS Management Console.
Select Analyses and Create analysis.

Figure 3 – Create Amazon Quick Sight analysis

Choose Create dataset and Create data source.

Figure 4 – Create dataset

Select Amazon Athena as the data source and select Next.
Enter a Data source name (for example, “SageMaker Catalog Metadata”) and choose Create data source.

Figure 5 – Create data source

Select Use custom SQL and enter a custom SQL query that references the S3 Tables catalog using the full three-part name.

Figure 6 – Use custom SQL

Figure 7 – Enter custom SQL

SELECT * FROM "s3tablescatalog/aws-sagemaker-catalog".asset_metadata.asset

Select Confirm query.
Choose Directly query your data (SPICE import may fail with S3 Tables catalogs)

Figure 8 – Directly query your data

Choose Visualize and Create to start building your dashboard.

Create visualizations with Amazon Quick.

With Amazon Quick, you can build governance dashboards using natural language prompts. This removes the need for manual field configuration. This approach is faster and more intuitive than traditional dashboard building.The Amazon Quick Sight user must have AdminPro or AuthorPro subscription (the Build feature isn’t available for Reader users).Start building your dashboard with the following steps:

Select Build in the top toolbar to open the natural language builder.

Figure 9 – Amazon Quick build dashboard

You will see a text box where you can describe the visualization that you want to create.

Create each visualization using natural language. For each of the six recommended visualizations, enter the corresponding natural language prompt, select Build, then choose ADD TO ANALYSIS.

Figure 11 – Add to analysis

Visualization 1: Asset inventory by type

Show count of asset_id by resource_type_enum as a pie chart

After the pie chart is created, choose ADD TO ANALYSIS.

Visualization 2: Documentation completeness

Show count of asset_id where business_description is not null asa KPI

After the KPI is created, choose ADD TO ANALYSIS.

Visualization 3: Monthly registration trends

Show count of asset_id by asset_created_time month as a line chart

After the line chart is created, choose ADD TO ANALYSIS.

Visualization 4: Asset count by account

Show count of asset_id by account_id as a bar chart

After the bar chart is created, choose ADD TO ANALYSIS.

Visualization 5: Namespace distribution

Show count of asset_id by namespace as a treemap

After the treemap is created, choose ADD TO ANALYSIS.

Visualization 6: Resource type by namespace

Show count of asset_id by resource_type_enum and namespace as a heat map

Choose ADD TO ANALYSIS

Arrange and publish your governance dashboard with the following steps:
Delete any empty or unwanted visualizations by choosing the three dots menu and choosing Delete.
Arrange visualizations by dragging them into your preferred layout.
Resize visualizations to emphasize key metrics.
Add titles to each visualization for clarity.
Choose PUBLISH in the top right corner.
Enter a dashboard name: “SageMaker Catalog Governance Dashboard”.
Verify these options are selected:
1. Allow executive summary.
2. Allow sharing stories.
3. Allow sharing scenarios.
Choose Publish dashboard.

Figure 12 – Amazon SageMaker Catalog governance dashboard

1. Analyze your dashboard with natural language.

After you publish, you can ask questions about your governance data:

1. On the dashboard, choose Analyze this dashboard in a Scenario in the top center.
2. In the Data to Insights panel, enter natural language questions such as:
  1. “Which resource types have the lowest documentation rates?”
  2. “How many assets were registered last month compared to this month?”
  3. “What percentage of assets lack ownership information?”
3. Choose Submit to generate AI-powered insights.

Amazon Quick analyzes your data and provides insights with supporting visualizations.

1. Generate executive summaries

Create automated governance reports for data stewards and compliance teams:

1. Choose the Amazon Quick logo in the top left to return to the home page
2. Select Dashboards from the left panel
3. Choose your “SageMaker Catalog Governance Dashboard”
4. Choose the Create dropdown menu in the top right
5. Select Executive Summary

Amazon Quick will automatically generate a summary with key governance insights, including Total asset counts and growth trends, Documentation completeness metrics, Ownership coverage statistics, and Classification distribution analysis.

1. Create governance stories.

Build governance reports that combine multiple dashboards:

1. From the Create dropdown, select Story.
2. Enter a prompt: “Write a summary of catalog governance metrics and data quality trends”.
3. Choose Add to select dashboards to include in the report.
4. Choose Build (this might take a few minutes to complete).

Amazon Quick will generate a narrative report combining your visualizations with AI-generated insights. Share the reports with leadership or compliance teams.

Governance dashboards contain metadata such as ownership and classification details. Restrict access to users who need it. In the Amazon Quick Sight console, open the dashboard, choose Share, and grant access to named users or a dedicated Quick Sight group (for example, data-stewards) instead of selecting Everyone in this account. Review the dashboard’s permissions periodically and remove entries that are no longer needed.

Cleaning up

To avoid ongoing charges, clean up the resources created in this walkthrough. Delete Amazon Quick Sight resources including the dashboard, analyses, and dataset.

Conclusion

In this post, you connected Amazon Quick Sight to your Amazon SageMaker Catalog metadata export, built governance dashboards using the Amazon Quick natural language prompts. This approach gives data stewards and compliance teams visibility into catalog health through six key visualizations covering asset inventory, documentation completeness, registration trends, account distribution, classification coverage, and stale asset detection.

Together with the metadata export and SQL query capabilities covered in the Analyzing your data catalog: Query SageMaker Catalog metadata with SQL post, this solution provides a complete, low-overhead governance monitoring pipeline from raw catalog metadata to executive-ready.

To learn more about Amazon SageMaker Catalogs, see Amazon SageMaker Catalog documentation. To expand the work done with Amazon Quick, review Amazon Quick Sight documentation.

About the authors

Accelerate SQL development with SageMaker Data Agent in Query Editor

Jason Ramos — Thu, 04 Jun 2026 15:40:11 +0000

When you develop SQL against Amazon Redshift and Amazon Athena, you spend time finding the right tables across hundreds of databases, writing complex joins and aggregations, debugging failed queries without context from previous attempts, and re-specifying filters for every new question. Amazon SageMaker Data Agent in Query Editor takes a different approach. You describe what you need in natural language, and the Data Agent generates the SQL. It references your actual tables through AWS Glue Data Catalog, proposes step-by-step plans for complex questions, retains context across your session, and offers one-click error recovery with Fix with AI. In this post, you learn how to use Data Agent in Query Editor to explore data, build multi-step analyses, recover from errors, and summarize results using a public education dataset.

Solution overview

You can go from a natural language question to executable SQL in seconds. Data Agent in Query Editor provides a conversational interface with direct access to your AWS data environment, so you spend less time on query mechanics and more time on analysis. Data Agent in Query Editor focuses specifically on SQL development against Amazon Redshift and Amazon Athena. (For Python, SQL, and PySpark across broader analytical and machine learning (ML) workloads, use Data Agent in notebooks.)

Data Agent provides four key capabilities:

Catalog-aware SQL generation. You don’t need to browse catalog structures or memorize schema details. Data Agent reads your table metadata directly.
Querybook and session context. You build on previous work. Data Agent uses context from your earlier queries and results.
Step-by-step planning. You review and approve a structured plan before Data Agent generates SQL.
Fix with AI. You recover from failed queries with one click.

Data Agent integrates with AWS Glue Data Catalog and reads your actual table names, column types, descriptions, and relationships, so generated SQL references your real tables. Each follow-up question builds on your current Query Editor session—the SQL cells in your querybook, the active connection, your selected cell, and execution results from previously run cells. For complex requests, Data Agent produces a structured plan that specifies which data to retrieve, how to aggregate it, and what filters to apply. You review and approve each step before Data Agent proceeds. When a query fails, choose Fix with AI to get a corrected query based on the error and the failed cell’s context.

[Figure 1: The Query Editor Fix with AI panel, showing a corrected SQL query ready for your review.]

Walkthrough: Education data analysis

In this section, you use Data Agent in Query Editor to analyze California schools data and identify where SAT improvement investment has the most impact. The walkthrough covers four tasks:

Explore available data.
Build a multi-step analysis plan.
Summarize insights from your queries.
Recover from a failed query.

The same workflow applies to your own data, whether you are analyzing sales figures, operational metrics, or financial records.

The California schools dataset contains SAT score results, school demographic information, and county-level data for public schools across California. The dataset includes tables that organize SAT scores by subject (reading, writing, math), school details (name, address, county, district), and enrollment figures. After you upload the data into your project database, you directly access the tables from Query Editor through your Amazon Athena or Amazon Redshift Lakehouse connection.

Prerequisites

To complete this walkthrough, you need intermediate SQL knowledge and basic familiarity with the AWS Management Console. You don’t need prior AWS Glue experience, but familiarity with data catalogs (centralized metadata repositories) helps.

You can choose one of two setup paths:

Quick start (5 minutes). SageMaker Unified Studio provides a sample database (sagemaker_sample_db) with pre-loaded data. To explore it, choose Data in the navigation pane, expand AwsDataCatalog, and select sagemaker_sample_db.
Full setup (30–45 minutes). Upload the California schools dataset into your project’s Lakehouse database. This dataset is publicly available from the California Department of Education. Download the SAT scores, school information, and county-level data files, then upload them through the SageMaker Unified Studio UI. In your project, go to Build, choose Query editor, right-click your project database in the Data explorer, and choose Create table. Drag and drop each CSV file to create the tables. SageMaker Unified Studio stores the data in the project-managed Amazon Simple Storage Service (Amazon S3) location, registers it in AWS Glue Data Catalog, and applies AWS Lake Formation governance automatically.

Running queries against Amazon Athena or Amazon Redshift might incur costs. For pricing details, refer to Amazon Athena pricing and Amazon Redshift pricing. For detailed setup instructions, refer to AWS Identity and Access Management (IAM)-based domains and projects. Before starting the walkthrough, you must have a SageMaker Unified Studio IAM-based domain with a project using the SQL analytics or All Capabilities project profile. The project automatically provisions an AWS Glue database, the required IAM role, and Athena or Redshift Lakehouse connections.

[Figure 2: The Data Explorer panel in Query Editor, showing the california_schools_db and sagemaker_sample_db tables.]

Explore available data. To start, enter the following prompt in the Data Agent panel:

Query my SAT scores from my california_schools_db

Data Agent searches AWS Glue Data Catalog, locates the relevant tables, and generates an initial exploratory query that retrieves SAT score records. It adds a SQL cell directly to your querybook.

Review the generated SQL in the comparison view, which highlights the proposed code.
Choose Accept, Reject, or Accept and run.
After you run the cell, the results appear inline, giving you a view of the data (column names, score ranges, and the number of records) before you write SQL.

[Figure 3: Data Agent returns an exploratory query for the california_schools_db tables, ready for your review.]

[Figure 4: The SQL query results appear beneath the cell after you choose Accept and run.]

Build a multi-step analysis plan. With the data explored, enter a more complex analytical question:

Identify which subjects need investment to improve SAT scores in the lowest-performing counties. Include school-level details with addresses.

Data Agent proposes a step-by-step plan before generating SQL. For this request, Data Agent breaks the question into three steps:

Aggregate SAT scores by county and subject to find performance patterns.
Filter to counties with a sufficient number of schools and rank the lowest performers.
Join school address data to produce a final detailed list.

Review the plan in the Data Agent panel and choose Run step-by-step to proceed.

[Figure 5: Data Agent proposes a multi-step plan with options to Cancel plan or Run step-by-step.]

Data Agent generates SQL for each step and adds it as a separate querybook cell. Review each cell’s SQL in the comparison view, then choose Accept and run to execute it. The results from each step are visible inline, so you can verify the intermediate output (county-level aggregations, the filtered ranking, and the final school list) before moving to the next step. When the steps are complete, your querybook contains the full analytical progression from raw scores to a detailed investment list.

[Figure 6: Each plan step produces a querybook cell that you can review and run independently.]

Summarize insights from your queries. After running the analysis, enter the following prompt:

Summarize the insights from my queries

Data Agent has context on your querybook, including the SQL and the query results from each cell. It generates a natural language summary: which counties are underperforming, which subjects (reading, writing, or math) need the most attention in each county, and how many schools appear on the investment list. This summary provides a starting point for a report or presentation.

[Figure 7: Data Agent summarizes insights from the accumulated query results in the querybook.]

Recover from a failed query. During the analysis, a generated query might produce an error, for example, referencing a column name that doesn’t match the schema or a join condition that returns unexpected results. When a cell fails, Query Editor displays the error message and a Fix with AI option.

Choose Fix with AI, and Data Agent reads the error in the context of the failed cell, then generates corrected SQL and updates the querybook cell. Run the corrected cell to verify the fix.

[Figure 8: After you choose Fix with AI, Data Agent is prompted to generate a corrected query for the failed cell.]

[Figure 9: Data Agent returns corrected SQL for you to review.]

Security and governance

Data Agent operates within your AWS environment and only accesses data that your IAM policies explicitly permit. Your existing IAM access controls and AWS Lake Formation permissions determine what data Data Agent can reach. To use Data Agent, your project role must have permissions to invoke specific Amazon DataZone APIs. For more information, refer to Actions, resources, and condition keys for Amazon DataZone.

Data Agent includes content filtering that prevents it from responding to off-topic requests, requests to reveal its system prompt, and requests for internal technical implementation details. Data Agent is restricted to AWS-related topics and English-language output.

Amazon SageMaker stores your natural language prompts and generated SQL in the AWS Region where you created your SageMaker Unified Studio domain. Data Agent doesn’t store your data, querybook context, or catalog metadata.

To opt out of data usage for service improvement, configure an AI services opt-out policy for Amazon DataZone in AWS Organizations. For more information, refer to Data storage in the SageMaker Data Agent, Service improvement, and AI services opt-out policies.

Clean up

The walkthrough creates querybook cells in your Query Editor session but doesn’t provision standalone infrastructure. To remove the generated SQL cells, delete them from your querybook or delete the querybook itself.

If you uploaded the California schools dataset specifically for this walkthrough, remove the following resources to avoid ongoing charges:

SageMaker Unified Studio domain. If you created a domain solely for this walkthrough, delete it to stop incurring charges. Refer to the SageMaker Unified Studio administration guide for deletion steps.
Uploaded tables. In the Data explorer, right-click each table you created and choose Delete table to remove the data from your project database and the underlying S3 storage.
Amazon Athena query results. Amazon Athena stores query results in an S3 output location. Delete the query result files from that bucket, or delete the bucket if you created it solely for this walkthrough.
Amazon CloudWatch logs. If Amazon Athena queries generated CloudWatch log groups, delete those log groups to avoid storage charges.

Conclusion

Data Agent in Query Editor brings conversational, catalog-aware SQL development to your Amazon Redshift and Amazon Athena workloads. In this post, you explored unfamiliar data, built a multi-step investment analysis, recovered from query errors, and summarized findings through natural language prompts.

Data Agent works within your existing IAM and AWS Lake Formation security controls, keeps your data within your AWS environment, and retains context across your analytical workflow so each question builds on the last.

Get started with these next steps:

Run your first prompt. Open Query Editor in your SageMaker Unified Studio domain and enter Show me the top 10 tables in my catalog with the most columns. For setup, refer to the SageMaker Unified Studio getting started guide.
Add descriptions to your AWS Glue Data Catalog. Table descriptions and column-level business metadata improve the quality of generated SQL. For best practices, refer to Populating the AWS Glue Data Catalog.
Try a multi-step analysis. Enter Which product categories had declining revenue quarter-over-quarter, and which regions drove the decline? and review Data Agent’s plan step by step.

For more information, refer to the Amazon SageMaker Data Agent documentation, the What’s New blog post, Amazon Redshift documentation, and Amazon Athena documentation. To learn how Data Agent works in notebooks, refer to Accelerate context-aware data analysis and ML workflows with Amazon SageMaker Data Agent.

About the authors

Schedule notebook runs in Amazon SageMaker Unified Studio

Shivani Mehendarge — Wed, 03 Jun 2026 20:19:11 +0000

If you build notebooks for recurring tasks such as daily customer analysis, weekly report generation, or data quality checks in Amazon SageMaker Unified Studio, you’ve likely wanted to run them automatically on a schedule. Until now, there wasn’t a native way to do this. Teams had to manage orchestration separately, even though the interactive notebook experience was already in place. Now, notebook scheduling is available, so you can configure your production workloads to run automatically with minimal manual intervention.

In this post, we walk you through the new scheduling and orchestrating capabilities for notebooks in Amazon SageMaker Unified Studio. You will learn how to:

Trigger on-demand background runs, such as a model re-training job, without waiting at your desk.
Create recurring schedules for tasks such as nightly data freshness checks or weekly business reviews.
Parameterize notebooks so a single template can generate reports across different AWS Regions or customer segments.
Orchestrate multi-notebook workflows where one notebook’s output feeds into the next. For example, an extract, transform, and load (ETL) pipeline followed by a summary dashboard refresh.
Debug failed runs with AI-assisted troubleshooting.

Sample use case overview

In this walkthrough, you will take on the role of a logistics analyst who monitors shipping performance across carriers. The notebook loads shipping data from the ShippingLogs.csv dataset, identifies late deliveries, and generates a performance summary. You want to run this notebook every morning without manual intervention, reuse it across different carriers, and know when something goes wrong.

You will start by running a notebook in the background and viewing the results. Next, you will create a recurring schedule for daily runs, then parameterize the notebook to generate reports for different carriers. You will also orchestrate the notebook in a multi-step workflow and debug a failed run using AI-assisted troubleshooting.

Prerequisites

Before you begin, you need:

An Amazon SageMaker Unified Studio project with Notebooks enabled. See Set up IAM-based domains for permission requirements.
A sample dataset. We use the ShippingLogs.csv dataset, which contains shipping data including estimated and actual delivery times, carriers, and origins. You can download it from the Workshop Studio (the file is named ShippingLogs.csv on the linked page).

Setting up the notebook

Start by creating a new notebook in your SageMaker Unified Studio project. If you haven’t already, upload the ShippingLogs.csv file under the Shared tab in the Files panel.

In the first cell, we load and explore the dataset. To reference the file in code, select the file in the Shared tab and copy the Amazon Simple Storage Service (Amazon S3) URI shown in the file details. Alternatively, you can reference it with this code:

import pandas as pd
from sagemaker_studio import Project

# Initialize the project
proj = Project()

# Get the S3 root path
s3_root = proj.s3.root

df = pd.read_csv(s3_root + '/ShippingLogs.csv')
df.head()

The dataset contains columns including Carrier, ActualShippingDays, ExpectedShippingDays, ShippingOrigin, ShippingPriority, and OnTimeDelivery. Add a second cell to analyze shipping performance for a single carrier:

import matplotlib.pyplot as plt

carrier_data = df[df['Carrier'] == 'GlobalFreight']
# Flag late deliveries
carrier_data['is_late'] = carrier_data['ActualShippingDays'] > carrier_data['ExpectedShippingDays']
late_pct = carrier_data['is_late'].mean() * 100
# Visualize actual vs expected shipping days
plt.figure(figsize=(12, 4))
plt.hist(carrier_data['ActualShippingDays'] - carrier_data['ExpectedShippingDays'], bins=20, edgecolor='black')
plt.axvline(x=0, color='red', linestyle='--', label='On time')
plt.title(f'Shipping Delay Distribution - GlobalFreight ({late_pct:.1f}% late)')
plt.xlabel('Days Over Expected')
plt.ylabel('Number of Shipments')
plt.legend()
plt.show()

With the notebook working interactively, you’re ready to automate it.

Running a notebook asynchronously

To trigger an asynchronous run, open your notebook. In the notebook header, choose the menu on the Run all button, and then choose Run in background.

This captures a snapshot of the notebook in its current state and starts a run on a separate dedicated compute. You can continue working on other tasks or close the browser entirely. Your interactive session isn’t affected.

You will see a notification at the bottom of your screen confirming that the run started. To check the status of your run, choose View Run in the notification. This opens a view showing every background and scheduled run with its status, duration, and a link to view the full output.

You can choose to view the run details at any point to view results as cells run. The run details include three tabs:

Output: The notebook in read-only mode with cell results rendered, including dataframe outputs, visualizations, and print statements.
Parameters: The parameter values used for this run.
Logs: Run logs for debugging.

You can also access past runs by selecting the View Runs option in the notebook header.

Stopping an in-progress run

If you need to cancel a run, open the run, and choose Stop. The run terminates, and its status updates to reflect the cancellation.

What to know about background runs

Compute: Each background run uses its own dedicated compute, separate from your interactive session. Your interactive work isn’t interrupted.

Packages: The packages that you install through the notebook’s package manager will be available in your background runs. When you use !pip install in code cells, the asynchronous run installs those packages as well.

Local files: Background runs can’t access files stored locally in your notebook environment. Reference data from your project’s shared storage (Amazon S3) or connected data sources instead.

Startup time: Expect a few minutes of startup time while compute is provisioned and your environment is prepared.

Creating a recurring schedule

Now that you’ve confirmed asynchronous runs work correctly, you can automate the notebook on a schedule. Choose the schedule icon in the notebook header to open the schedule creation form.

Configure the following settings:

Schedule name: Enter a descriptive name, such as Daily Shipping Report.
Schedule type: Choose Recurring for repeated runs or One-time for a single future run.
Frequency: Define how often the notebook runs using a rate (for example, every one day) or a cron expression. Set the time zone and the start and end dates for the schedule. For example, set the schedule to run every day at 7:00 AM UTC starting tomorrow.
Flexible time window (optional): The number of minutes after the scheduled start time within which the run can be invoked. For example, with a 5-minute window, the notebook runs within 5 minutes of the start time.
Advanced settings:
- Compute Instance: Keep the current settings or override with a different instance type for the asynchronous run to use.
- Timeout: Set a maximum run duration to help prevent notebooks from running indefinitely. If left blank, it defaults to 60 minutes.

Choose Create.

The schedule appears in the Schedules tab of the activity panel. SageMaker Unified Studio creates an Amazon EventBridge Scheduler schedule for each schedule you configure.

Viewing schedule run history

To view past runs for a schedule, choose the schedule name in the Schedules activity panel. This opens the schedule details view, where you can see the list of runs triggered by that schedule, the duration of each run, and a link to open the notebook output for an individual run.

Editing and deleting schedules

To modify a schedule, choose Edit next to it in the Schedules panel. You can change the frequency, instance type, timeout, and other configuration fields. To pause or resume a schedule, choose Pause or Resume from the same menu. To remove a schedule, choose Delete from that menu. Deleting a schedule stops future runs but preserves historical run outputs in Amazon S3 for auditing purposes.

Parameterizing notebooks

With parameters, you can reuse a single notebook across different inputs without duplicating code. For example, you can run the same shipping performance report for each carrier by passing a different carrier name to each run.

Defining parameters

Open the Parameters activity panel and choose Add. Set the parameter name to carrier and the default value to GlobalFreight.

Using parameters in code

In your notebook, replace the second cell with the following code. This retrieves the carrier parameter value using the SageMaker Unified Studio Python SDK instead of the hardcoded value:

import sagemaker_studio
import matplotlib.pyplot as plt

carrier = sagemaker_studio.nbutils.parameters.get("carrier")

carrier_data = df[df['Carrier'] == carrier].copy()
carrier_data['is_late'] = carrier_data['ActualShippingDays'] > carrier_data['ExpectedShippingDays']
late_pct = carrier_data['is_late'].mean() * 100

plt.figure(figsize=(12, 4))
plt.hist(carrier_data['ActualShippingDays'] - carrier_data['ExpectedShippingDays'], bins=20, edgecolor='black')
plt.axvline(x=0, color='red', linestyle='--', label='On time')
plt.title(f'Shipping Delay Distribution - {carrier} ({late_pct:.1f}% late)')
plt.xlabel('Days Over Expected')
plt.ylabel('Number of Shipments')
plt.legend()
plt.show()

Creating schedules with different parameter values

Now create three schedules for the same notebook, each targeting a different carrier:

“daily-shipping-gf” with carrier = GlobalFreight.
“daily-shipping-mc” with carrier = MicroCarrier.
“daily-shipping-shipper” with carrier = Shipper.

When you view a historical run, a separate Parameters tab in the run output displays the parameter values that were active for that run.

You can also override parameter values when triggering an on-demand background run. Choose the menu on the Run all button, then choose Run with settings. You can keep the defaults or provide custom values for that run.

Orchestrating with Workflows

To combine notebooks into a multi-step pipeline, such as running a data calculation notebook before the shipping log notebook, you can use the Notebook Operator in the Workflows tool to orchestrate them.

To do this, choose the Add to workflows button under the options menu of the notebook header.

This takes you to the Workflows tool, adding a new Notebook Operator task with prefilled properties from your notebook. When configuring the Operator task:

Select the target notebook from the notebook menu.
Use the Parameters widget to pass notebook parameters into the run of the notebook.
Specify optional arguments such as the compute instance and timeout configuration for the run.

Workflows also supports polling for the status of a notebook run for a particular notebook using Notebook Sensor. In Workflows, you can add a new Sensor task by hovering on the edge of the existing Operator task, where a plus (+) button is displayed.

You can then search for and add the Notebook Sensor to the canvas.

When configuring the Sensor task, specify the notebook run ID within the text field. The Operator’s form field contains Jinja templating to retrieve the notebook run. If the Sensor is used within the same workflow as the Operator, this template can be copied to use within a Sensor to poll the notebook run. Select the target notebook from the notebook menu.

Within Workflows, you can configure notebook runs to emit outputs and use those outputs as inputs for subsequent notebook runs.

Building off of the previous shipping log notebook example, we will pass the carrier parameter from an upstream notebook’s output. Your shipping-logs-analysis notebook should be already set up.

Because the notebook depends on the carrier parameter, you can specify it in the Parameters panel.

Now, define a second notebook, calculate-best-carrier, which performs a calculation to determine our best carrier to use for shipping:

import pandas as pd
from sagemaker_studio import Project

# Initialize the project
proj = Project()

# Get the S3 root path
s3_root = proj.s3.root

df = pd.read_csv(s3_root + '/ShippingLogs.csv')
df.head()

carrier_stats = df.groupby('Carrier').agg(
    total=('OrderID', 'count'),
    late=('OnTimeDelivery', lambda x: (x == 'Late').sum())
).reset_index()
carrier_stats['late_pct'] = carrier_stats['late'] / carrier_stats['total'] * 100

best = carrier_stats.sort_values('late_pct', ascending=True).iloc[0]
best_carrier = best['Carrier']

print("Late % by carrier:")
print(carrier_stats.to_string(index=False))
print(f"\nBest carrier: {best_carrier} ({best['late_pct']:.1f}% late)")

To configure the calculate-best-carrier notebook’s outputs, you can choose the Variables panel. A new selector is available at the bottom of this panel which allows you to select variables to mark as outputs.

We want this notebook to emit the best_carrier variable.

Now, use the Add to workflows button as previously demonstrated to quickly add this notebook within a workflow. Chain a second Notebook Operator that points to our shipping-logs-analysis notebook. Because we specified a parameter dependency on carrier for this notebook, it’s available as an option in the Parameters widget menu.

When they’re chained, the notebook tasks detect the outputs set in upstream notebook runs. These outputs can be selected as keys within the Parameters widget of the Operator to pass into the run. This can be done recursively for an arbitrary number of Operator tasks. We can select the emitted best_carrier output from the calculate-best-carrier notebook.

You can now choose the Save button on the top left of the visual canvas and the Run button to start the workflow. When the workflow is completed, the specified notebook outputs are available in the Task Output panel and the notebook run result can be viewed in the Notebooks tool.

In a similar manner, the Notebook Sensor will also emit the notebook outputs from a particular notebook’s run which can be used within other tasks. This is useful when you want to retrieve outputs from a notebook run in another workflow.

Debugging a failed run with AI assistance

When viewing your past runs, you notice that a run from earlier today has a Failed status. Choose the failed run to open the notebook output in read-only mode.

In this example, suppose you incorrectly referred to column name ActualShippingDays as DeliveryDays. The run would fail with a KeyError: 'DeliveryDays' in the cell that computes late deliveries.

At the top of the failed run output, choose Troubleshoot with AI. Choosing the Troubleshoot with AI button lands you in the notebook with the Agent chat panel open.

The data agent analyzes the cell outputs, identifies the cell that errored, explains the root cause, and suggests a fix. In this case, it identifies that the column DeliveryDays doesn’t exist in the dataframe and suggests updating the code reference. You can review the change, then verify the fix by choosing Run in background from the Run all menu to trigger a test run before the next scheduled run.

Note: You can also use the Data Agent to create schedules and start notebook runs using natural language, without having to navigate.

Cleaning up

To avoid incurring future charges, delete the resources that you created in this walkthrough:

Delete any schedules that you created from the Schedules panel in your notebook.
Delete test notebooks if you don’t need them.
Navigate to the Workflows page and delete any workflows that you created during this walkthrough.
Your project’s Amazon S3 storage retains historical run outputs until you manually remove them.

Conclusion

In this post, we showed how to run notebooks in the background in Amazon SageMaker Unified Studio using background runs, schedules, parameterization, workflow orchestration, and AI-assisted debugging. Using a shipping logistics dataset, we demonstrated how a single notebook can be parameterized to generate performance reports for different carriers on independent schedules, all without duplicating code or managing extensive infrastructure.

To get started, open a notebook in your SageMaker Unified Studio project, choose the menu on the Run all button in the notebook header, and choose Run in background. For more advanced use cases, explore workflows in Amazon SageMaker Unified Studio to build multi-step data pipelines, or review the Amazon SageMaker Unified Studio User Guide for additional configuration options.

Learn more:

If you have feedback or questions, reach out on AWS re:Post for Amazon SageMaker Unified Studio.

About the authors

Amazon OpenSearch Service: Mechanisms to secure your domain

Imtiaz Sayed — Wed, 03 Jun 2026 15:21:53 +0000

Imagine you’re building a product search feature for your website or storing customer records in Amazon OpenSearch Service to power full-text search. The moment that real user data enters your domain, security becomes essential.

Whether your workload is a public-facing website search, an internal application querying sensitive data, or a pipeline handling personally identifiable information (PII), the questions you face are the same:

Who should be allowed to connect to my domain?
How do I authenticate users and services?
How do I make sure that even authenticated users only see data they are entitled to see?
How do I satisfy regulatory requirements such as HIPAA, PCI DSS, or SOC 2?

This post offers an overview of the security mechanisms available for Amazon OpenSearch Service, spanning authentication and authorization, encryption, and network access controls. You learn how to implement fine-grained access control, manage AWS Identity and Access Management (IAM) roles, and secure data both in transit and at rest for both public and virtual private cloud (VPC) access domains.

Scope: This post covers security for Amazon OpenSearch Service managed clusters only. It doesn’t cover Amazon OpenSearch Serverless, which uses a different security model. For serverless security, see Amazon OpenSearch Serverless security in the AWS documentation.

To begin, let’s look at the security layers in Amazon OpenSearch Service.

Amazon OpenSearch Service security layers

Amazon OpenSearch Service has multi-layer security. The following diagram illustrates the multi-layer security in Amazon OpenSearch Service.

Figure 1: Multi-layer security.

The three main layers of security are network, domain access policy, and fine-grained access control.

Network – The first security layer is the network, which determines whether requests reach an OpenSearch Service domain. If you choose Public access when you create a domain, requests from any internet-connected client can reach the domain endpoint. If you choose VPC access, clients must connect to the Amazon Virtual Private Cloud (Amazon VPC) (and the associated security groups must permit it) for a request to reach the endpoint.

Domain access policy – The second security layer is the domain access policy. After a request reaches a domain endpoint, the resource-based access policy allows or denies the request access to a given URI. The access policy accepts or rejects requests at the edge of the domain, before they reach data or indexes in OpenSearch itself.

Fine-grained access control – The third and final security layer is fine-grained access control. After a resource-based access policy allows a request to reach a domain endpoint, fine-grained access control evaluates the user credentials and either authenticates the user or denies the request. If fine-grained access control authenticates the user, it fetches all OpenSearch internal roles mapped to that user and uses the full set of permissions to determine how to handle the request.

With fine-grained access control, you can control access to your data in Amazon OpenSearch Service. For example, depending on who makes the request, you might want to hide certain fields in your documents or exclude certain documents altogether. With fine-grained access control, you can:

Define role-based access control to determine who can perform which actions on which indexes, documents, and fields.
Define security at the index, document, and field level to allow access to only required data.

Fine-grained access control requires OpenSearch or Elasticsearch 6.7 or later. It also requires HTTPS for all traffic to the domain, encryption of data at rest, and node-to-node encryption. Depending on how you configure the advanced features of fine-grained access control, more processing of your requests might require compute and memory resources on individual data nodes. After you turn on fine-grained access control, you can’t turn it off. For more details, see Fine-grained access control in Amazon OpenSearch Service in the AWS documentation.

To learn more about security features in an OpenSearch Service domain, let’s start by configuring a new public access domain. We discuss a VPC access domain later in the post.

Public access domain

With a public access domain, you can configure an OpenSearch Service domain so that the domain endpoint is accessible from the internet.

The AWS console for Amazon OpenSearch Service provides a guided wizard that you can use to configure and reconfigure your provisioned Amazon OpenSearch Service domains. Follow the Tutorial: Configure a domain with the internal user database and HTTP basic authentication in the AWS documentation to configure a domain with basic authentication and validate fine-grained access control.

Let’s review some important configuration attributes for a public access domain.

Network:

Public access. To simplify the network access configurations, you can use Public access, but for production workloads, we recommend VPC access.

With the domain in public access, you have several options to secure access. While you can use a resource-based access policy to restrict access to specific IAM principals or IP addresses, the recommended approach is to turn on fine-grained access control (FGAC) and use it as the primary mechanism for securing your domain. With FGAC turned on, you can set an open access policy (allowing all traffic to reach the domain) and let FGAC handle authentication and authorization at the index, document, and field level.

When using IAM-based authentication with FGAC, you should map IAM roles to backend roles in OpenSearch. You can use backend roles to assign permissions to groups of users based on their IAM role, rather than managing individual user mappings. This is especially important because if your IAM federation or authentication mechanism changes, the backend role mappings make sure of consistent access control within OpenSearch.

Figure 2: Use public access domain.

Fine-grained access control: Fine-grained access control provides numerous features to help you keep your data secure, such as document-level security, field-level security, read-only users, and OpenSearch Dashboards/Kibana tenants. Fine-grained access control requires a primary user, which is the administrator identity we discuss through the rest of this post.

The primary user is the administrator identity for your OpenSearch domain. This user can set up additional users in Amazon OpenSearch Service, assign roles to them, and assign permissions for those roles. You can choose username and password authentication for the primary user or use an IAM identity. You use these credentials to log in to OpenSearch Dashboards. Following the best practices on choosing your primary user, you should move to an IAM primary user for production workloads.

Fine-grained access control can be applied regardless of how you log in. You can follow your organization’s suggested authentication mechanism and apply fine-grained access control on top of it.

FGAC provides security at multiple levels to meet your security needs:

Index-level security – Controls who can create, search, read, write, update, or delete within specific indexes.
Document-level security – Restricts which documents within an index a user can see, using OpenSearch query filters (for example, only show documents where department: “sales”).
Field-level security – Controls which fields within documents are visible (include or exclude specific fields).
Field masking – Anonymizes sensitive field data (for example, hash a release_date or SSN field) rather than hiding it entirely.

Fine-grained access control supports several authentication mechanisms, including HTTP basic authentication using an internal user database, Amazon Cognito for web-based Dashboards access, SAML for enterprise identity provider integration, JSON Web Tokens (JWT) for token-based authentication, and AWS Identity and Access Management with SigV4 signing for IAM users and roles.

Encryption:

Amazon OpenSearch Service encrypts data both in transit and at rest. When you turn on fine-grained access control, encryption is required—the corresponding settings are automatically turned on and can’t be changed. These include Transport Layer Security (TLS 1.2 or later) for requests to the domain and for traffic between nodes in the domain, and encryption of data at rest through AWS Key Management Service (AWS KMS).

For encryption at rest, OpenSearch Service supports three key types: AWS owned keys, AWS managed keys, and customer managed keys. While AWS owned keys provide a quick-start option with no additional configuration, customer managed keys are the recommended best practice. Customer managed keys give you full control over the encryption key lifecycle, including key rotation policies, granular access control through key policies, and the ability to audit key usage through AWS CloudTrail. To use a customer managed key, create a symmetric encryption key in AWS KMS and select it when configuring your domain’s encryption settings.

For a basic public access domain with FGAC, all traffic reaches the domain freely (no VPC restriction), and an open access policy is used so no SigV4 signing is needed. FGAC then takes over, authenticating users through the internal user database (username/password) and enforcing role-based permissions at the index, document, and field level.

The public access configuration we discussed is useful for development and testing, but for production workloads, a best practices deployment combines VPC access, IAM-based authentication, and fine-grained access control. This approach layers all three security mechanisms—network isolation, identity verification, and granular permissions—to protect your domain end to end.

VPC access domain

Placing your OpenSearch Service domain inside a VPC restricts network-level access to resources within the VPC or connected networks. Traffic between your applications and the OpenSearch endpoint doesn’t traverse the public internet, and you can use security groups to further limit which entities can communicate with the domain. OpenSearch Service places a VPC endpoint (VPCe) using AWS PrivateLink into one, two, or three subnets of your VPC depending on your Availability Zone configuration. For high availability (HA), turn on multiple Availability Zones with each subnet in a different zone within the same AWS Region. For more details, see Launching your Amazon OpenSearch Service domains within a VPC.

For this best practices deployment, we use an IAM primary user with Amazon Cognito authentication for OpenSearch Dashboards and for fine-grained access control. We configure a primary IAM role and a limited IAM role, associate them with users in Amazon Cognito through a user pool and identity pool, and then use fine-grained access control to manage permissions. The primary user can then sign in to OpenSearch Dashboards, create backend roles, map the limited user to a restricted role, and enforce granular access at the index, document, and field level. For more details, see Tutorial: Configure a domain with an IAM master user and Amazon Cognito authentication in the AWS documentation.

The following high-level steps detail what’s needed to configure a VPC access domain with Amazon Cognito users. These steps use the Amazon Cognito user pool for authentication. The same basic process works for any Cognito authentication provider that lets you assign different IAM roles to different users.

Create an Amazon Cognito user pool.
Add users in the user pool for the primary user and a limited-access user.
Create an Amazon Cognito identity pool.
Update the IAM role for the primary user to allow access to OpenSearch Dashboards.
Create an IAM role for the limited user.
Create the domain.

You can follow Creating and managing Amazon OpenSearch Service domains in the AWS documentation to provision a domain. The following sections describe some important attributes for the domain.

Network:

VPC access. Public access isn’t recommended for production workloads. We recommend that you use VPC access for all production workloads. Pick the VPC, subnets, and security group that you have created for the OpenSearch domain.

Figure 4: Use VPC access.

Fine-grained access control:

Turn on fine-grained access control with OS[MasterUserRole] as the primary user. You can follow steps in Tutorial: Configure a domain with an IAM master user and Amazon Cognito authentication to create OS[MasterUserRole].

Figure 5: Turn on fine-grained access control with an IAM role.

Fine-grained access control provides numerous features to help you keep your data secure, such as document-level security, field-level security, read-only users, and OpenSearch Dashboards/Kibana tenants. Fine-grained access control requires a primary user.

The primary user is the administrator identity for your OpenSearch domain. This user can set up additional users in Amazon OpenSearch Service, assign roles to them, and assign permissions for those roles. You can choose username and password authentication for the primary user or use an IAM identity. You use these credentials to log in to OpenSearch Dashboards. Following the best practices on choosing your primary user, you should choose an IAM primary user for production workloads.

Fine-grained access control can be applied regardless of how you log in. You can follow your organization’s suggested authentication mechanism and apply fine-grained access control on top of it.

Amazon Cognito authentication:

To turn on Amazon Cognito authentication, select Enable Amazon Cognito authentication and choose the Amazon Cognito user pool and Amazon Cognito identity pool for your OpenSearch Dashboards.

Figure 6: Turn on Amazon Cognito authentication.

Access policy:

The access policy controls whether a request is accepted or rejected when it reaches the Amazon OpenSearch Service domain. You can configure a domain-level access policy to allow access to your Amazon OpenSearch Service domain.

Figure 7: Configure domain-level access to the domain.

Encryption:

For encryption at rest, OpenSearch Service supports three key types: AWS owned keys, AWS managed keys, and customer managed keys. While AWS owned keys provide a quick-start option with no additional configuration, customer managed keys are the recommended best practice. Customer managed keys give you full control over the encryption key lifecycle, including key rotation policies, granular access control through key policies, and the ability to audit key usage through AWS CloudTrail. To use a customer managed key, create a symmetric encryption key in AWS KMS and select it when configuring your domain’s encryption settings.

With these configurations, you can configure your Amazon OpenSearch domain and OpenSearch Service Dashboards so that they’re accessible only within the chosen VPC. For your production scenario, you can follow your organization’s approved mechanism to access the resources in a VPC. You can access OpenSearch Service Dashboards with a primary user to create a limited-access role and map it to the IAM role with limited access to validate fine-grained access control.

Conclusion

In this post, we looked at the important security configurations for a public and a VPC-based Amazon OpenSearch domain. You can examine more settings for fine-grained access control in the OpenSearch Dashboards Security section.

If you have feedback about this post, submit comments in the Comments section. If you have questions about this post, start a new thread on the Amazon OpenSearch Service forum or contact AWS Support.

About the author

Capture data lineage of Amazon EMR spark jobs into Amazon SageMaker Unified Studio

Jose Romero — Mon, 01 Jun 2026 18:37:47 +0000

Data engineers running Apache Spark jobs on Amazon EMR face a persistent challenge: understanding how data moves through Spark pipelines as it’s transformed, joined, and written to downstream tables . Tracking these transformations manually requires examining job logs, reviewing code, and piecing together transformation logic across multiple sources. As pipelines scale, this process becomes complex. The visibility gap affects key business activities: troubleshooting data quality issues takes longer – impact analysis for schema changes requires more effort – and compliance audits need extensive documentation of data provenance.

Amazon SageMaker is the center for all your data and analytics where you can find and access all the data in your organization and act on it using tools across various use case. This unified platform addresses the data visibility challenge by bringing together data governance, collaboration, and discovery into a single interface. At the heart of this platform is Amazon SageMaker Catalog, a centralized hub that enables organizations to catalog, govern, and discover all their data assets with complete visibility into lineage. By capturing data lineage across your entire data ecosystem from raw sources through transformations to final outputs, SageMaker Catalog enables you to track data provenance across your entire platform, enable collaboration with clear visibility into data ownership and quality metrics, build trust through comprehensive data lineage that supports compliance and confident decision-making, and accelerate discovery of trustworthy, governance-ready data assets. You can access and visualize this lineage directly in Amazon SageMaker Unified Studio, which serves as the unified interface to explore data relationships and collaborate across your analytics workflows.

Amazon EMR, starting from version 7.11, now includes native OpenLineage support that automates lineage capture. OpenLineage is an open-source framework for data lineage that automatically emits lineage metadata from your data transformation jobs directly into Amazon SageMaker Catalog, or other data governance solutions, without requiring customizations.

This EMR native support of OpenLineage is part of a growing set of integrations across AWS analytics services including AWS Glue, Amazon EMR Serverless, and Amazon Redshift. The complete list of services with native OpenLineage integration can be found in the data lineage support matrix.

In this post, you’ll walk through a practical, step-by-step example that shows how to capture and track data lineage from Spark jobs running on Amazon EMR directly into Amazon SageMaker Catalog using OpenLineage. You’ll see how lineage metadata flows automatically and explore data relationships and dependencies across your workflows in Amazon SageMaker Unified Studio.

Solution overview

Imagine you’re part of a large enterprise that relies on HR analytics to optimize workforce planning, compensation strategies, and talent retention practices. Your data engineering team owns the delivery of these analytical products by processing raw HR datasets (including employee records, attendance logs, and compensation details), with Spark jobs running on your Amazon EMR infrastructure.

With time, Spark jobs have grown in complexity. Your team now struggles to maintain visibility into how data moves through pipelines, who modified it, and how to map dependencies between datasets and final analytical products.

The following solution demonstrates how you can address these challenges by automatically capturing data lineage end-to-end from Spark jobs running on your EMR infrastructure and visualizing it in Amazon SageMaker Unified Studio so that you and the business understand data provenance of the final analytical products.

The architecture includes a Data Layer with CSV files containing employee, attendance, salary, and bonus data stored in Amazon S3 (Simple Storage Service), representing typical HR and payroll source systems.

The Processing Layer uses Amazon EMR cluster running Apache Spark jobs that transform raw data into analytical tables. The first Spark job joins employee and attendance data while the second Spark job combines attendance with compensation data. Both jobs use Apache Iceberg table format to provide ACID (Atomic, Consistent, Isolated, and Durable) transactions and time travel capabilities.

The Metadata Layer uses AWS Glue Data Catalog to store Iceberg table metadata, making tables discoverable and accessible across AWS analytics services. A Lineage Layer uses the OpenLineage integration in EMR to automatically track input/output datasets (CSV files and Iceberg tables), transformation logic at column level (joins, filters, aggregations), and job execution metadata.

Finally, the Data Governance Layer uses Amazon SageMaker Catalog to capture and process OpenLineage events posted by the EMR Spark jobs and automatically build a comprehensive lineage graph that shows complete data provenance from CSV source files through Spark transformations to Iceberg analytical tables.

Before you deploy this solution, make sure you have the following resources in place.

Prerequisites

For this walkthrough, you should have the following prerequisites:

An AWS account.
Your assumed role should have full access to Amazon EMR serverless, Amazon S3, Amazon Identity and Access Management (IAM) and AWS Lambda. Note that for production workloads, minimum permissions are recommended.
A Amazon VPC (Virtual Private Cloud) with at least one subnet with internet access. You can provision this VPC as you create the Amazon SageMaker domain next.
An existing Amazon SageMaker Unified Studio domain and project. To get started, use the quick setup option as explained here. To create a project, follow the instructions here.
An S3 bucket with the sample data files and Spark scripts uploaded (see Prepare Your Source Data below)
Default EMR service roles — if this is your first time using EMR in this account, run `aws emr create-default-roles` from the AWS CLI or CloudShell to create them.

With these prerequisites in place, let’s examine what the AWS CloudFormation template will deploy to your AWS environment.

Architecture components

The deployment creates several interconnected components that work together to capture and visualize lineage:

An S3 bucket to store all data and artifacts for the solution.
An EMR cluster (v 7.12.0) with Apache Iceberg support enabled and OpenLineage integration pre-installed, ready to run Spark jobs with lineage tracking.
A set of IAM policies that grant the necessary permissions to the EMR cluster to post lineage events to your SageMaker Unified Studio domain.
A set of AWS Lake Formation permissions that grant the EMR cluster to create, alter, and drop Iceberg tables in your specified Glue database.

With an understanding of what will be deployed, you’re ready to launch the CloudFormation stack.

Deploy the solution

Note: While this walkthrough uses the AWS EMR console and AWS CLI to verify the cluster and run Spark jobs, you can also perform these steps directly from Amazon SageMaker Unified Studio. SMUS provides a unified interface to create and manage EMR clusters, submit Spark jobs, and monitor execution — all within the same environment where you’ll later explore the lineage captured in Amazon SageMaker Catalog.

Prepare your source data

Before deploying the CloudFormation stack, clone or download the Git repository.

git clone https://github.com/aws-samples/sample-capture-data-lineage-of-amazon-emr-ec2

Upload the CSV files downloaded from git to the input/ prefix and the spark scripts in scripts/ prefix. You can run the following command to upload the files:

aws s3 cp employees.csv s3://YOUR-BUCKET/input/
aws s3 cp attendance.csv s3://YOUR-BUCKET/input/
aws s3 cp salary_adjustments.csv s3://YOUR-BUCKET/input/
aws s3 cp bonus_payments.csv s3://YOUR-BUCKET/input/
aws s3 cp emr-lineage-spark-job.py s3://YOUR-BUCKET/scripts/
aws s3 cp emr-lineage-compensation-job.py s3://YOUR-BUCKET/scripts/

To deploy the solution, complete the following steps in CloudFormation console:

Create new stack by specifying the CloudFormation yaml file previously download from git repository PutHereThe YMLFileName
Enter a stack name (such as, emr-lineage-demo) and provide the following parameters:
- SourceS3BucketName: S3 bucket containing your CSV files and Spark scripts
- SourceCSVPrefix: S3 prefix where CSV files are located
- SourceScriptsPrefix: S3 prefix where Spark scripts are located
- GlueDatabaseName: The name of the Glue database associated to your Amazon SageMaker Unified Studio project.
- DataZoneDomainId: Your SageMaker Unified Studio domain ID.
- VpcId: The id of the VPC that was deployed as part of the prerequisites.
- For EMRReleaseLabel, MasterInstanceType, CoreInstanceType and CoreInstanceCount, keep the default values.
Acknowledge IAM resource creation, choose Next and then Submit. The CloudFormation stack takes approximately 10 to 15 minutes to complete.
In the EMR console, wait for the cluster status to show as WAITING before moving to the next step.

Now that the EMR cluster is running with OpenLineage enabled, let’s examine how the Spark jobs are configured to capture lineage metadata.

Explore data lineage configuration in EMR

When submitting Spark jobs to EMR, specific configurations enable OpenLineage to create and post lineage events to SageMaker Unified Studio as the job runs:

spark.hadoop.hive.metastore.client.factory.class – Configures Spark to use AWS Glue as the Hive metastore.
spark.jars – Path to the pre-installed OpenLineage library (available on EMR 7.11+).
spark.extraListeners – Registers an OpenLineage listener to capture metadata of input / output datasets and transformations.
spark.openlineage.transport.type – Uses the OpenLineage DataZone transport option to send lineage events directly into SageMaker Catalog.
spark.openlineage.transport.domainId – The ID of your SageMaker Unified Studio domain, that serves as the target for lineage events.
spark.glue.accountId – Your AWS account ID for Glue data catalog operations.

Now that you understand the configuration that enables automatic lineage capture, you’re ready to run the data pipeline.

When running this two-step pipeline, you will calculate the total employee compensation by combining salary adjustments, bonuses, and attendance data. The final analytical asset will serve payroll processing and budgeting.

Run employee attendance analysis job

The first job reads employee details (in employees.csv dataset) and attendance records (in attendance.csv dataset), joins the datasets on EmployeeID and creates a unified dataset (employee_attendance Iceberg table) in your Glue database.

Follow the steps below to run this first job:

In the CloudFormation console, navigate to the stack’s Outputs tab
Copy the value of the Job1SubmitCommand output key. Note that this is the command you’ll use to submit the first job in EMR with the right configuration.

Run the command in your terminal or AWS CloudShell.
Monitor the job in the Amazon EMR console under Steps.

Run employee compensation analysis job

Now, you will calculate the total employee compensation (Iceberg table) by combining salary adjustments (salary_adjustments.csv dataset), bonuses (bonus_payments.csv dataset), and attendance (calculated in the last step):

Repeat the steps 1 to 4 to run Job 2.
After completion, open the AWS Glue console.
Navigate to Data Catalog, then Tables and select your SageMaker project’s database.
Confirm that employee_attendance and employee_compensation tables are listed.

With both Spark jobs complete, you can now visualize the complete data lineage graph in Amazon SageMaker Unified Studio.

Visualizing lineage in SageMaker Unified Studio

SageMaker Unified Studio provides a graph-based data lineage visualization that helps data engineers, analysts, and data scientists clearly understand which source datasets (files or tables) feed into each dataset, what transformations and logic are applied at every step, which downstream analytics assets consume the data, and how changes to upstream data or transformations may impact the rest of the data pipeline.

Now that the data pipeline run successfully, let’s review the captured lineage for the HR data in SageMaker Unified Studio:

Navigate to the SageMaker Unified Studio console, sign in to your domain.
Open your project and go to Data Sources
Find your AWS Glue Data Catalog source

Click RUN. Two new assets will be created.

Navigate to Assets and Click on employee_compensation. Under the LINEAGE tab you’ll find the lineage graph view that SageMaker builds based on the OpenLineage metadata captured from the EMR Spark jobs as they run.

- You’ll first see three lineage nodes from left to right: one representing the EMR Spark job that created the final Iceberg table, a second one representing the actual Iceberg table in the Glue catalog, and a third one representing the data asset in the SageMaker Catalog inventory that maps to the Glue table.
- Click on any lineage node to view its underlying metadata in the details pane, including dataset names, S3 locations, schema, data types, job execution details and more.

Expand the lineage to the left by clicking on the double arrow next to the first lineage node. Keep expanding until you hit the originating datasets.

- Expanding the graph to the left reveals the complete data pipeline back to original CSV source files. You can see how compensation data depends on upstream attendance analytics.
- Note how each lineage node represents an element in the data pipeline you run, including both Spark jobs and even the intermediate employee_attendance Iceberg table that connects them.

You can expand column-level lineage by clicking on the column section of a lineage node of a dataset or data asset. This allows you to understand how data changes at a column level as it goes downstream your data pipeline.

Cleanup

To avoid ongoing charges, clean up the resources:

First, empty the destination bucket by running the following command in your terminal or with AWS CloudShell.

aws s3 rm s3://${DEST_BUCKET}/ --recursive

Delete the CloudFormation stack.
- On the AWS CloudFormation console, choose Stacks in the navigation pane.
- Choose the stack you created, then choose Delete and then Delete stack when prompted.

Conclusion

In this post, you explore how to capture data lineage from Spark jobs in Amazon EMR (v7.11+) directly into Amazon SageMaker Unified Studio. You learned how to set up an Amazon EMR cluster with native OpenLineage support to automatically track lineage metadata from Spark jobs processing your data. You also configured the integration between EMR and Amazon SageMaker Catalog to ensure lineage information flows seamlessly into your governance platform. Finally, you explored the resulting lineage graph in SageMaker Unified Studio and saw how it provides comprehensive visibility into data transformations, from source CSV files through Spark processing jobs to final analytical tables using Apache Iceberg format.

We encourage you to now test these capabilities with your own data pipelines running on EMR. By implementing automated lineage tracking, many customers have strengthened their governance frameworks while gaining valuable insights into data dependencies, impact analysis, and compliance requirements. This approach enables data teams to build trust in their analytics outputs while maintaining the agility needed to derive business value from their data assets.

About the authors

The next generation of Amazon OpenSearch Serverless: Built from the ground up for agents

Sohaib Katariwala — Thu, 28 May 2026 18:24:54 +0000

Audience note: This is the deep-dive technical launch post. For a shorter overview of what changed and why, see the related post on the AWS News Blog.

Today, we are announcing a ground-up re-architecture of Amazon OpenSearch Serverless that delivers up to 20 times faster autoscaling, scale to zero, and up to 60% lower cost than provisioning clusters for peak load. Amazon OpenSearch Service is a fully managed, open source retrieval engine that unifies vector, lexical, hybrid, and agentic search, delivering low-latency, accurate and relevant results. Amazon OpenSearch Serverless is an automatically scaled deployment option.

Modern workloads are increasingly dynamic and unpredictable. An ecommerce platform sees a 10x traffic spike during a flash sale. An artificial intelligence (AI) agent triggers hundreds of concurrent vector queries while reasoning through a multi-step task, then goes idle. A multi-tenant SaaS application serves dozens of tenants with wildly different activity patterns. These workloads need infrastructure that scales up to meet demand and releases resources when demand drops.

That is why we rebuilt the Amazon OpenSearch Serverless architecture from the ground up. The new architecture decouples compute from storage. The service provisions infrastructure in seconds instead of minutes, and scales compute all the way to zero when your application is idle. In this post, we walk through the new architecture, what it means for your applications, and how to get started with a hands-on tutorial.

With this launch, Amazon OpenSearch Serverless introduces two named architectures. Existing collections are now referred to as Classic collections. The new architecture is called NextGen and is now the default when you create a new collection via the AWS Console. You can use NextGen architecture in the API by specifying --generation NEXTGEN in the CLI. To continue using the Classic architecture, specify --generation CLASSIC in the CLI or omit the optional --generation parameter.

What this means for your applications

The new architecture delivers improvements across three pillars: performance, cost, and a simplified user experience.

Performance: Autoscaling in seconds

An OpenSearch Compute Unit (OCU) is the unit of compute capacity that powers your indexing and search workloads. Amazon OpenSearch Serverless now provisions additional OCUs in seconds. When traffic arrives, the service adds resources in line with demand instead of reacting after a worker is already under pressure. The same mechanism scales the infrastructure back down quickly when traffic drops. The new architecture scales capacity up to 20 times faster than the previous architecture, so your users experience consistent performance during traffic surges, and you stop paying for capacity when you no longer need it.

Cost efficiency: Pay only for what you use

Indexing, search, storage, and Vector Index GPU-Acceleration are metered and billed independently, so you can see and optimize each dimension of your workload separately.

Decoupled compute and storage: OpenSearch Serverless now has full decoupling between compute and storage, allowing OCUs to scale up and down irrespective of the amount of data stored in a collection. This is powered by a new storage layer that is accessible to both indexing and search OCUs. You can now have multiple indices with data indexed in them but not pay any compute costs if you are not actively indexing or searching data. For workloads with significant idle time, the new architecture can reduce infrastructure costs by up to 60% compared to the cost of provisioning OpenSearch Service domains for peak capacity.

Scale to zero: When no requests arrive within the idle timeout window (10 minutes), the service releases compute resources and your OCU usage scales to 0. When traffic resumes, capacity is back in approximately 10 seconds. During this window, the service queues incoming requests and serves them once capacity is available; it does not drop them. If you anticipate a burst of traffic, for example before a scheduled batch job or a marketing campaign, you can send a lightweight query (such as a match_all with size=1) to warm the collection before your application starts sending production traffic. This reduces the latency your users experience on the first real request. Indexing and search scale independently. If you have no search requests, search OCUs scale to zero, even while OpenSearch Serverless maintains indexing OCUs for indexing requests, and vice versa.

GPU acceleration for vector workloads: For vector collections created in the new architecture, OpenSearch Serverless automatically uses GPU-backed compute to accelerate Hierarchical Navigable Small World (HNSW) vector index construction, significantly reducing indexing time compared to CPU-only builds. GPU acceleration kicks in automatically whenever there is an opportunity to leverage GPUs to reduce overall indexing time and cost. In the Classic architecture, you had to opt in or out of GPU acceleration at the collection level through the API. If you want to disable GPU acceleration for NextGen collections for a specific index, you can turn off the remote index build setting at the index level. GPU usage appears as a separate line item on your bill, so you have full visibility into when acceleration was active and what it cost. For more details on how GPU acceleration works and performance benchmarks, refer to Build billion-scale vector databases in under an hour with GPU acceleration on Amazon OpenSearch Service.

Simplified experience: Fewer steps to production

We also simplified the day-to-day experience of running OpenSearch Serverless:

With the new architecture, you can provision a collection and start sending requests in seconds. There is no need for capacity planning, no sizing decisions, and no waiting for infrastructure to warm up. This makes Amazon OpenSearch Serverless a natural fit for agentic workloads, where an AI agent can spin up a vector search or retrieval step on demand and expect a response without delay.

To make getting started even faster, we have introduced Express Create on the console. You supply a collection name and a collection type, choose Express Create, and your collection is active in seconds with no upfront network, encryption, or access policies to configure. You can add those later if your workload requires them.

Collection groups and collections can also be created programmatically using the AWS Command Line Interface (AWS CLI) and AWS SDKs. AWS CloudFormation support is coming soon.

The new architecture introduces two endpoint formats on the on.aws domain. The per-collection endpoint (<collectionId>.aoss.<region>.on.aws) works the same way as before with one endpoint per collection. The per-account Regional endpoint (<accountId>.aoss.<region>.on.aws) is new: it serves all of your collections through a single hostname, with the target collection identified in each request using the x-amz-aoss-collection-name or x-amz-aoss-collection-id header. This means one connection pool, one Transport Layer Security (TLS) session, and one endpoint to manage regardless of how many collections you have — a significant improvement for multi-tenant workloads where each tenant maps to its own collection. Both endpoints use standard AWS PrivateLink, so you create virtual private cloud (VPC) endpoints from the VPC console or the EC2 API just like any other AWS service. Private Domain Name System (DNS) is configured automatically, eliminating the Amazon Route 53 Private Hosted Zones, forwarding rules, and custom DNS infrastructure that were required with the original architecture. Cross-VPC, cross-account, and on-premises access all work using standard vpce-* DNS names with no additional setup.

Collection groups are the new unit of organization for your collections. You can share compute capacity across multiple collections with Collection Groups, which reduces cost for smaller collections that have complementary traffic patterns. You can also assign different AWS Key Management Service (AWS KMS) keys to collections within the same group, so you get both cost efficiency and per-collection encryption isolation. Collection groups are required when creating collections with the new architecture.

You also get the benefits of OpenSearch open-source releases without needing to manage versions and upgrades. The service tracks upstream releases automatically.

Amazon OpenSearch Serverless is also available on the Vercel Marketplace, making it straightforward for developers to add search infrastructure directly from their Vercel projects. You can link an existing AWS account through delegated access, or get started through a Limited Scope Account with USD $100 in AWS credit if you are new to AWS.

The integration creates a collection with sensible defaults, scale-to-zero billing, public endpoints, and AWS-managed encryption, and automatically sets connection details as environment variables in your Vercel project. You can choose from Search or Vector Search collection types depending on your use case, whether that is full-text search or semantic and AI-powered search.

How the architecture works

The new Amazon OpenSearch Serverless architecture separates compute from storage entirely. OCUs are stateless and read from and write to a distributed shared storage layer that is accessible to both indexing and search OCUs. The storage layer is designed for high durability, keeping your data available independently of the compute nodes that process it.

This design has two practical consequences:

Fast provisioning. New OCUs start serving requests in seconds because there is no local disk to bootstrap. The OCU mounts the shared storage layer and begins processing immediately.
Efficient scale down. Idle capacity can be released with no impact to your stored data, because the data never lived on the OCU. When traffic subsides, compute resources are released and your cost drops accordingly.

Architecture comparison

The following table summarizes the key differences between the original and new architectures:

Capability	Classic Architecture	NextGen Architecture
Minimum capacity	2 OCUs (always on)	0 OCUs (scale to zero)
Scaling speed	Minutes	Seconds
Storage	Local storage per compute node	Distributed shared storage (decoupled)
Collection organization	Individual collections (Default) Collection groups (Optional)	Collection groups (required)
Cold start from zero	N/A (always on)	~10 seconds
Endpoint	Per-collection endpoint	Regional endpoint (static per account)
Cost vs. OpenSearch Service domain	Baseline	Up to 60% lower cost
Scaling speed (vs. Classic)	Baseline	Up to 20 times faster than baseline

Walkthrough: Create a vector collection and observe scale to zero

In this walkthrough, you create a vector search collection with Express Create, index a few sample documents with embeddings, run a k-nearest neighbor (k-NN) query, and watch the collection scale to zero in Amazon CloudWatch. The entire process takes about 10 minutes.

Prerequisites

An AWS account with permissions to create Amazon OpenSearch Serverless collections.
AWS Command Line Interface (AWS CLI) configured with appropriate credentials.
curl 7.75 or later (for built-in --aws-sigv4 support).

Step 1: Configure security policies

Create encryption, network, and data access policies. These must exist before the collection can be created.

# Create an encryption policy
aws opensearchserverless create-security-policy \
    --name product-vectors-encryption \
    --type encryption \
    --policy '{"Rules":[{"ResourceType":"collection","Resource":["collection/product-vectors"]}],"AWSOwnedKey":true}' \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

# Create a network policy (public access for this tutorial)
aws opensearchserverless create-security-policy \
    --name product-vectors-network \
    --type network \
    --policy '[{"Rules":[{"ResourceType":"collection","Resource":["collection/product-vectors"]},{"ResourceType":"dashboard","Resource":["collection/product-vectors"]}],"AllowFromPublic":true}]' \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

# Get your principal ARN
PRINCIPAL_ARN=$(aws sts get-caller-identity --query 'Arn' --output text)

# Create a data access policy
aws opensearchserverless create-access-policy \
    --name product-vectors-data \
    --type data \
    --policy "[{\"Rules\":[{\"ResourceType\":\"index\",\"Resource\":[\"index/product-vectors/*\"],\"Permission\":[\"aoss:CreateIndex\",\"aoss:DescribeIndex\",\"aoss:UpdateIndex\",\"aoss:DeleteIndex\",\"aoss:ReadDocument\",\"aoss:WriteDocument\"]}],\"Principal\":[\"\${PRINCIPAL_ARN}\"]}]" \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

Note: If you use the AWS console’s Express Create workflow, these policies are created automatically.

Important: After creating the data access policy, wait approximately 30 to 60 seconds for the policy to propagate before making API calls to the collection. If you receive a 403 Forbidden error, wait and retry.

Step 2: Create a collection group and collection

Create a collection group with scale-to-zero capacity limits, then create a vector search collection within it.

# Create a collection group with scale-to-zero enabled (min OCU = 0)
aws opensearchserverless create-collection-group \
    --name product-search-cg \
    --generation NEXTGEN \
    --standby-replicas ENABLED \
    --capacity-limits "minIndexingCapacityInOCU=0,maxIndexingCapacityInOCU=4,minSearchCapacityInOCU=0,maxSearchCapacityInOCU=4" \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

# Create a vector search collection in the group
aws opensearchserverless create-collection \
    --name product-vectors \
    --type VECTORSEARCH \
    --collection-group-name product-search-cg \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

The collection status transitions to ACTIVE within seconds.

Step 3: Create a vector index

Retrieve the collection endpoint and create a k-NN index using 3-dimensional vectors:

ENDPOINT=$(aws opensearchserverless batch-get-collection \
    --names product-vectors \
    --query 'collectionDetails[0].collectionEndpoint' \
    --output text \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2")

awscurl --service aoss --region us-east-2 \
    -XPUT "${ENDPOINT}/items" \
    -H "Content-Type: application/json" \
    -d '{
      "settings": {"index.knn": true},
      "mappings": {
        "properties": {
          "description": {"type": "text"},
          "embedding": {"type": "knn_vector", "dimension": 3,
            "method": {"name": "hnsw", "space_type": "cosinesimil", "engine": "faiss"}}
        }
      }
    }'

Note: If the collection has scaled to zero, the first request might take a few seconds while capacity scales up. If the request times out, wait 10 to 15 seconds and retry.

Step 4: Index sample documents with embeddings

awscurl --service aoss --region us-east-2 \
    -XPOST "${ENDPOINT}/items/_bulk" \
    -H "Content-Type: application/json" \
    -d '
{ "index": { "_id": "1" } }
{ "description": "Wireless noise-cancelling headphones", "embedding": [0.8, 0.2, 0.1] }
{ "index": { "_id": "2" } }
{ "description": "Portable Bluetooth speaker", "embedding": [0.7, 0.3, 0.2] }
{ "index": { "_id": "3" } }
{ "description": "Over-ear studio monitor headphones", "embedding": [0.9, 0.1, 0.05] }
'

Step 5: Run a k-NN query

Search for the two nearest neighbors to a query vector. Wait 30 seconds after indexing to allow the vector index to build before running this query:

awscurl --service aoss --region us-east-2 \
    -XGET "${ENDPOINT}/items/_search" \
    -H "Content-Type: application/json" \
    -d '{
      "query": {
        "knn": {
          "embedding": {
            "vector": [0.85, 0.15, 0.08],
            "k": 2
          }
        }
      }
    }'

The response returns the two most similar items, in this case, the headphone documents whose embeddings are closest to your query vector.

You can also run this query in OpenSearch UI by navigating to your collection in the Amazon OpenSearch Service console and choosing the OpenSearch UI Application URL. Then follow the steps outlined in this blog to create a workspace. Then navigate to Dev Tools and paste and run the following query.

GET items/_search
{
  "query": {
    "knn": {
      "embedding": {
        "vector": [0.85, 0.15, 0.08],
        "k": 2
      }
    }
  }
}

Step 6: Observe scale to zero

After a period of inactivity (no indexing or search traffic), the collection group scales down to 0 OCU. Verify with:

aws opensearchserverless batch-get-collection-group \
    --names product-search-cg \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

In the response, currentCapacity.search.capacityInOcu and currentCapacity.indexing.capacityInOcu will show 0 after the collection has scaled down.

You can also navigate to the Collection groups page in the Amazon OpenSearch Service console. Choose your collection group, then scroll down to the Monitoring section. Here you can see two charts: Indexing capacity (OCUs) and Search capacity (OCUs). After 10 minutes of idle time (no indexing or search requests), both metrics drop to zero, confirming that the service has released all compute resources for your collection.

Clean up

To avoid ongoing charges, delete the resources you created in this walkthrough when you are done. Delete the collection first so the collection group becomes empty, then delete the group, then remove the security and access policies.

# Look up the collection ID, then delete the collection
COLLECTION_ID=$(aws opensearchserverless batch-get-collection \
    --names product-vectors \
    --query 'collectionDetails[0].id' \
    --output text \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2")

aws opensearchserverless delete-collection \
    --id "${COLLECTION_ID}" \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

# Look up the collection group ID, then delete the collection group
GROUP_ID=$(aws opensearchserverless batch-get-collection-group \
    --names product-search-cg \
    --query 'collectionGroupDetails[0].id' \
    --output text \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2")

aws opensearchserverless delete-collection-group \
    --id "${GROUP_ID}" \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

# Delete the security and access policies
aws opensearchserverless delete-security-policy \
    --name product-vectors-encryption \
    --type encryption \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

aws opensearchserverless delete-security-policy \
    --name product-vectors-network \
    --type network \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

aws opensearchserverless delete-access-policy \
    --name product-vectors-data \
    --type data \
    --endpoint-url "https://aoss.us-east-2.amazonaws.com" \
    --region "us-east-2"

Upgrading existing collections

To move to the new architecture, create a new collection group and collection, then reindex your data into it. For a step-by-step walkthrough of the reindexing process, refer to Perform reindexing in Amazon OpenSearch Serverless using Amazon OpenSearch Ingestion. Your queries and index mappings remain the same. Only the collection endpoint changes. With the new static Regional endpoint, that is a one-time update.

The new architecture supports SEARCH and VECTORSEARCH collection types. TIMESERIES is not supported at launch.

Conclusion

The new Amazon OpenSearch Serverless architecture is available today. You can create your first OpenSearch Serverless collection in seconds with Express Create, scale it to handle production traffic, and your OpenSearch Serverless compute costs drop to zero when it sits idle.

To learn more:

If you have questions or feedback, open a support case or reach out through your AWS account team. We look forward to seeing what you build.

About the authors

How Buildkite Operates Test Analytics at Massive Scale with Amazon MSK and Amazon Managed Service for Apache Flink

James Hill — Wed, 27 May 2026 18:22:34 +0000

When engineering teams at Slack, Reddit, Canva, Airbnb, Shopify, and Uber need to ship code with confidence, they rely on Buildkite. As a CI/CD platform, Buildkite orchestrates complex build, test, and deployment pipelines for some of the most demanding engineering organizations in the world. It handles everything from routine code commits to artificial intelligence (AI) model-training workloads, processing over 50 billion requests per month.

At the heart of Buildkite’s test orchestration portfolio is Test Engine, a specialized analytics product designed to help engineering teams understand and optimize their test suites at scale. Test Engine aggregates results across thousands of builds, flags flaky tests, runs parallel test execution across machine fleets, and delivers interactive analytics on test execution data. It supports arbitrary metadata tagging for dimensions like instance type, architecture, language version, cloud provider, and feature flags.

The challenge? Delivering all of this in real time, across multiple enterprise tenants, at a volume that would stress even the most robust data infrastructure. In this post, we explore how Buildkite uses Amazon Managed Streaming for Apache Kafka (Amazon MSK) and Amazon Managed Service for Apache Flink to power Test Engine’s streaming-first analytics architecture at scale.

The problem: When scale breaks traditional architectures

Buildkite’s Test Engine must ingest and serve analytics on test telemetry from thousands of distributed pipelines simultaneously, for multiple enterprise customers. The scale is unforgiving: 50 billion test executions per month, 500K events per second at peak ingestion, and webhook payloads reaching 21 MB.

The architectural evolution and its limits

The original Rails and PostgreSQL stack couldn’t sustain this growth. In 2024, the team re-architected around a distributed streaming layer, a stateful stream processor for pre-aggregations, and multiple specialized stores: a key-value store for fast lookups, a relational database for pre-computed aggregates, and an open table format (Iceberg) with a distributed query engine (Trino) for flexible querying.

Yet the core tension remained unsolved. Enterprise customers demanded interactive, arbitrary slicing of billions of records across high-cardinality dimensions, not canned reports. The stream processor couldn’t handle ad hoc aggregations at query time. The key-value store was blind to analytical queries. The distributed query engine offered flexibility but was too slow for interactive use.

The result was a system that was expensive and operationally complex. It included nine relational database clusters, sprawling ETL pipelines, and 24/7 pre-aggregation jobs running regardless of demand. It still couldn’t deliver the one thing customers needed most: fast, flexible, interactive analytics at scale.

Architecture and implementation: MSK and Amazon Managed Service for Apache Flink as the streaming backbone

The solution Buildkite arrived at centers on Amazon Managed Streaming for Apache Kafka (Amazon MSK) and Amazon Managed Service for Apache Flink as the real-time data streaming and processing layers, decoupling high-throughput ingestion from downstream analytics.

The data pipeline

The following diagram shows the end-to-end data flow from CI/CD agents through Amazon MSK and Amazon Managed Service for Apache Flink to the analytics layer.

Amazon MSK sits at the critical junction between data producers (the distributed CI/CD agents and test collectors running across customer infrastructure) and the downstream processing and analytics layers. Amazon Managed Service for Apache Flink then transforms those raw event streams into enriched, queryable data before it reaches the analytics store.

High-throughput ingestion from CI/CD pipelines

Amazon MSK’s role begins at ingestion. Test collectors embedded in CI/CD pipelines publish test execution events directly to Kafka topics. The existing Amazon MSK cluster handles between 5 MB/sec and 100 MB/sec of inbound data under normal operating conditions. The architecture is designed to absorb the significant variance inherent in CI/CD workloads, where pipeline activity is bursty and correlated with engineering team working hours across global time zones.

When the Buildkite project was initiated, MSK Express Brokers were not yet available, leading the team to adopt MSK Tiered Storage as the primary mechanism for scaling and recovery. With MSK Express Brokers now generally available, the team is evaluating a migration of its most critical log ingestion workload, which sustains up to 1 GB/s at peak ingestion. MSK Express Brokers bring automatic storage scaling with zero storage management overhead, up to 20x faster scaling and 90% faster broker recovery, 3x higher per-broker throughput, 5x more partitions per broker, and built-in Intelligent Rebalancing.

Real-time stream processing with Amazon Managed Service for Apache Flink

Sitting between Amazon MSK and the analytics layer, Amazon Managed Service for Apache Flink acts as the stateful stream processing engine that transforms raw event streams before they reach downstream systems. Buildkite selected Flink for its exactly-once processing, mature stateful computation model, and deep Kafka integration. Handling sustained peaks of over 25,000 events per second, Amazon Managed Service for Apache Flink eliminates the operational overhead of cluster provisioning, version upgrades, checkpointing, and job recovery. This frees engineering teams to focus on application logic.

Amazon Managed Service for Apache Flink powers key stateful processing tasks, including flaky test detection through time-windowed pattern matching, enriching execution events with pipeline and customer metadata, and routing processed data to downstream systems such as ClickHouse for analytics, PostgreSQL for operational workloads, and Amazon Simple Storage Service (Amazon S3) for long-term archival.

Reliability and fault tolerance

Amazon MSK’s three-replica configuration ensures that no single broker failure can cause data loss or ingestion interruption. Combined with flexible data retention, the architecture provides a meaningful replay window. If a downstream consumer (Amazon Managed Service for Apache Flink, ClickHouse, or another service) experiences an outage, it can resume processing from its last committed offset without data loss.

During the migration to the current architecture, Buildkite employed a dual-write strategy: simultaneously writing to both the existing PostgreSQL pipeline and the new Amazon MSK/ClickHouse path. This approach allowed the team to validate data consistency and gradually shift traffic without risking customer-facing disruption. This pattern speaks to the operational maturity Amazon MSK provides.

Operational efficiency gains

The shift to a streaming-first architecture, combined with the downstream simplification of the analytics engine, produced significant operational improvements:

Flink workloads reduced by 60%+: Eliminating pre-aggregation jobs that ran continuously regardless of demand.
Key/value store completely retired: Amazon MSK’s buffering capability, combined with ClickHouse’s query performance, eliminated the need for a separate fast-lookup store.
PostgreSQL capacity cut in half: Nine separate database clusters consolidated and right-sized.
Thousands of lines of application code deleted: Simpler architecture means less ETL code, fewer failure modes, and faster onboarding for new engineers.

Platform performance at a glance

Metric	Value
Monthly test executions (for test engine platform)	50 billion (4x growth from 3B)
Sustained peak ingestion	500K events/second
Total records in analytics store	200 billion
Log ingestion requests	70,000+ per second
Peak webhook throughput	1.7 GB/second
MSK inbound throughput range	5 MB/sec – 100 MB/sec

Business and developer impact

The technical architecture ultimately exists to serve one purpose: helping developers ship better software faster. The streaming-first architecture built on Amazon MSK and Amazon Managed Service for Apache Flink delivers on that promise across four dimensions.

On-demand analytics replaced pre-computed reports. Customers can now interactively slice and dice 70 billion records across arbitrary metadata dimensions. They get answers to queries like “Show me P50 test durations by instance type and architecture for the last 30 days” in seconds, not hours. Real-time log streaming through the “live tail” feature means developers no longer wait for a build to complete before diagnosing failures. At 25,000 events per second, this experience scales across thousands of concurrent enterprise pipelines without degradation.

Smarter test intelligence comes from Amazon Managed Service for Apache Flink’s stateful flaky test detection: when a test begins exhibiting intermittent failure patterns, Amazon Managed Service for Apache Flink identifies it as it happens, not after the fact. This is what separates a proactive analytics platform from a reactive one. It requires publishing data to Kafka, processing with Flink, and letting ClickHouse handle the complex read requests.

Conclusion: Streaming as a strategic foundation

Buildkite’s journey from a Rails/Postgres monolith to a streaming-first analytics platform reflects a pattern increasingly common among enterprise SaaS companies: a reliable, high-throughput streaming and processing layer is not an optimization. It is a prerequisite for operating at scale.

Amazon MSK and Amazon Managed Service for Apache Flink form the backbone that helps Buildkite ingest 50 billion test executions per month, serve real-time interactive analytics to enterprise customers, and do so at lower cost than the more complex architecture it replaced. Amazon MSK handles durable, elastic event buffering. Amazon Managed Service for Apache Flink transforms raw streams into enriched, queryable data. Together they absorb the operational complexity that would otherwise consume engineering capacity.

For platform engineers evaluating streaming infrastructure for multi-tenant SaaS workloads, the signal is clear: invest in the streaming backbone early, and let managed services handle the operational complexity.

To learn more about Amazon MSK and Amazon Managed Service for Apache Flink, visit aws.amazon.com/msk and aws.amazon.com/managed-service-apache-flink.

About the authors

How Zynga scaled multi-warehouse data governance with Amazon Redshift federated permissions

Johan Eklund, Matthew Wongkee, Noelia Tardón — Wed, 27 May 2026 18:19:35 +0000

Zynga, a global leader in interactive entertainment operates a portfolio of mobile game studios including Socialpoint, the creators of Dragon City and Monster Legends. Zynga’s analytics platform processes telemetry and revenue data across studios using Amazon Redshift as its central data warehouse.

As Zynga expanded its analytics architecture to include individual studios with their own compute environments, the team faced a challenge: how to maintain centralized data governance while granting studios independent query capacity. Their existing approach to permission management introduced lag and required custom infrastructure if scaled to multiple warehouses.

In this post, we walk through how Zynga adopted Amazon Redshift federated permissions and AWS IAM Identity Center to enforce consistent, tiered data access across provisioned and serverless Amazon Redshift environments without building custom synchronization pipelines.

The challenge

Zynga needed to onboard Socialpoint’s current Amazon Redshift workloads and make Zynga’s central data available to them. Zynga’s existing production cluster would house the Socialpoint raw data, but the compute would come from another warehouse set up as a consumer. At the same time, Zynga’s data access control policies would need to be enforced across all warehouses. Zynga uses a tiered access control policy which would need to be synced across all consumers with no permission lag or manual grant synchronization.

During the migration, Socialpoint’s specific extract, transform, and load (ETL) processes would be included in Zynga’s central ETLs and their data ingestion pipeline would be replaced by Zynga’s latest generation of data ingestion infrastructure. Because the migration process happens in stages, Amazon Redshift sizing would also gradually need to increase.

The team evaluated two alternatives before arriving at a solution:

AWS Lake Formation couldn’t manage local and cross-cluster permissions using the same interface, and required AWS Access and Identity Management (IAM) or IAM Identity Center authentication for all users including service accounts.
Manual grants on consumer clusters introduced a delay between when permissions were updated on the producer and when they took effect on the consumer. This approach would also require an external job that synced permissions and would be unlikely to scale well beyond 2–3 consumers.

Solution overview

Zynga implemented a solution using three AWS services working together:

Amazon Redshift federated permissions enabled cross-cluster queries without explicit data shares. Permissions granted on the producer cluster propagate immediately to consumer workgroups through AWS Glue Data Catalog registration.
AWS IAM Identity Center provides unified authentication through federation with Okta. When users sign in, their Okta group memberships are provisioned through a System for Cross-domain Identity Management (SCIM) and automatically map to Amazon Redshift roles, removing the need for external synchronization jobs.
Amazon Redshift Serverless provides the compute layer for Socialpoint, scaling to zero when idle and avoiding the need to pre-size a provisioned cluster during the migration period.

The architecture uses a dual-grant approach where every permission is granted to both an IAM Identity Center group (for users) and a federated IAM role (for service accounts). This gives both authentication paths the same access.

How it works

Authentication with IAM Identity Center

Zynga’s existing Okta directory syncs to IAM Identity Center, which is connected to the Amazon Redshift Serverless workgroup. When a user authenticates, Amazon Redshift automatically creates a user mapped to their email address and assigns them to roles based on their Okta group membership.

For example, an analyst in the Gamma Tier group signs in and is automatically assigned the AWSIDC:role.sso.gamma role in Amazon Redshift. No manual role assignment or synchronization job is required.

Service accounts, used for programmatic access, authenticate differently. Either using their IAM role and calling the get-credentials API, or by using the new federated permissions feature. Each service account assumes a federated IAM role, which creates a corresponding federated user in Amazon Redshift (for example, IAMR:role_iam_gamma).

Figure 1 – Identity layer

The dual-grant approach

To ensure that both users and service accounts can access the same data, every read permission is granted to both the IAM Identity Center group and the federated IAM role in a single statement:

GRANT SELECT ON schema.table TO
	'IAMR:role_iam_gamma',
	ROLE 'AWSIDC:role.sso.gamma';

Transitioning the producer cluster

The shared provisioned cluster already had active users with local grants. To avoid disruption, Zynga implemented a tri-grant approach on the producer during the transition period. Existing stored procedures were modified to grant permissions to three targets: the legacy local role, the IAM Identity Center group, and the federated IAM role.

GRANT SELECT ON schema.table TO
	ROLE role_rs_gamma,
	ROLE 'AWSIDC:role.sso.gamma','IAMR:role_iam_gamma';

This approach maintains backward compatibility for existing users on the producer while enabling immediate access from the new serverless workgroup. The long-term plan includes migration of all producer users to IAM Identity Center and retire the legacy local grants.

Stored procedures for consistent governance

Rather than requiring users to construct dual-grant statements manually, Zynga created stored procedures that encapsulate the grant logic:

grant_read accepts a table name and access tier, then issues the appropriate dual-grant for both the IAM Identity Center group and the federated IAM role.
grant_write grants data definition language (DDL) and data manipulation language (DML) permissions to the appropriate team-based role.

This provides a consistent interface for permission management regardless of which cluster or workgroup that the user is on.

Figure 2 – Data and compute layer

Results

The migration delivered measurable improvements:

Immediate permission propagation – Grants on the producer cluster took effect on the consumer workgroup instantly, replacing a process that previously required manual intervention and introduced lag.
Zero additional infrastructure cost – Federated permissions, federated queries, and IAM Identity Center added no incremental costs to the architecture.
Removed custom synchronization – The team removed the need for Lake Formation configurations, external AWS Lambda functions, and Airflow workflows for permission management.
Scalable pattern – The same architecture can be extended to additional studio workgroups without duplicating permission management logic.

Lessons learned

Zynga adopted federated permissions shortly after the feature launched in US West Oregon (us-west-2) in January 2026. The team shared several observations from their early adoption:

Run a proof of concept first. The team validated the full permission model in a test environment before deploying to production, including testing that existing data shares to other clusters were not disrupted.
Plan for the dual-grant requirement. Because IAM Identity Center users and federated IAM roles are distinct identity types, every read permission requires two grants. Encapsulating this in stored procedures prevents errors and reduces cognitive overhead.
Start with serverless for new workloads. With Amazon Redshift Serverless, the team can avoid sizing decisions during the migration period. If usage patterns later justify it, they can migrate to a provisioned cluster from a serverless snapshot with minimal downtime.
Engage with AWS. As an early adopter of a new feature, Zynga maintained regular contact with the Amazon Redshift team through their AWS Technical Account Manager to report issues and request enhancements.

Conclusion

Zynga’s adoption of Amazon Redshift federated permissions demonstrates how organizations with multi-cluster Amazon Redshift architectures can enforce centralized data governance without building custom synchronization infrastructure. By combining federated permissions with IAM Identity Center and Amazon Redshift Serverless, the team established a pattern that scales to additional studios while maintaining consistent access controls and reducing operational overhead.

To learn more about the services used in this post, see the following resources:

AWS Big Data Blog

Access Amazon S3 data files directly using AWS Lake Formation permissions

Key use cases for Lake Formation permissions to S3 locations

What customers are saying

Lake Formation Credential Vending Plugin for AWS SDK v2 for Java

Solution overview

Prerequisites

Solution walkthrough

Step 1 – Create a parquet table in Data Catalog

Step 2 – Register S3 location and grant table permission to IAM roles in Lake Formation

Step 3 – Run ETL script in EMR

Step 4 – Run query as Data-Analyst using Athena

Clean up

Conclusion

About the authors

Building AI shopping agent using Amazon Bedrock AgentCore Runtime and Amazon OpenSearch Service

Solution overview

Walkthrough

Prerequisites

Step 1: Configure IAM permissions

Step 2: Connect to the model by using OpenSearch ML Connectors

Step 3: Create an ingest pipeline for data indexing

Step 4: Create an index for storing data

Step 5: Ingest sample data

Step 6: Query the index

Step 7: Create an agent with Strands and Bedrock AgentCore Runtime

Step 8: Configure and launch your agent to Bedrock AgentCore Runtime

Step 9: Configure OpenSearch Service access

Step 10: Invoke the Bedrock AgentCore Runtime

Cleanup

Conclusion

About the authors

Choosing the right workflow orchestration service for your use case: Amazon MWAA and AWS Step Functions

Understanding workflow orchestration requirements

Use case: Enterprise data analytics pipeline

Business challenge

Workflow characteristics

Solution: Amazon Managed Workflows for Apache Airflow (Amazon MWAA)

Stateful workflow management

Built-in scheduling capabilities

Granular recovery and resume control

Comprehensive monitoring and operational control

Implementation considerations

Use case: Real-time serverless application orchestration

Business challenge

Workflow characteristics

Solution: AWS Step Functions

Serverless architecture and automatic scaling

Event-driven workflow execution

Native AWS service integration

Cost-effective pay-per-use model

Human approval workflow support

Implementation considerations

Choosing the right workflow orchestration service

Consider Amazon MWAA when your use case involves:

Consider AWS Step Functions when your use case involves:

Decision framework

Conclusion

About the authors

Real-time CDC from Aurora PostgreSQL to Amazon S3 Tables using Debezium and Firehose

Solution overview

Debezium event transformation

Walkthrough

Prerequisites

Step 1: Enable CDC in Aurora PostgreSQL

Step 2: Build and register the Debezium plugin

Step 3: Configure the CDK project

Step 4: Deploy the CDK stacks

Step 5: Enable MSK VPC connectivity, grant Lake Formation permissions, and apply MSK cluster policy

Step 6: Create the Debezium connector

Step 7: Verify the Debezium connector

Step 8: Test the pipeline

Step 9: Verify data delivery

Step 10: Query data using Amazon Athena

Cleaning up

Conclusion

Related posts

About the author

Beyond JSON blobs: Implementing the VARIANT data type in Apache Iceberg V3

Solution overview