Post Syndicated from Satya Chikkala original https://aws.amazon.com/blogs/big-data/event-driven-pipeline-orchestration-with-amazon-mwaa-and-airflow-3-0/

Data engineering teams running Apache Airflow across multiple AWS accounts face a persistent coordination problem. They have no built-in way to coordinate workflows between their separate Amazon Managed Workflows for Apache Airflow (Amazon MWAA) environments, where each team or business unit manages its own isolated environment. Cross-environment orchestration has traditionally relied on time-based polling, complex custom sensors, or API-based triggers that introduce latency and reliability concerns. The Apache Airflow Datasets feature (introduced in version 2.4) added data-aware scheduling of Directed Acyclic Graphs (DAGs, the workflow definitions that specify tasks and their execution order) within a single Amazon MWAA environment. However, teams running Airflow across multiple accounts still had no way to coordinate workflows between environments.

With Apache Airflow 3.0, now available on Amazon MWAA 3.0, you get event-driven cross-account orchestration that responds to upstream events as they happen, without polling overhead or tight environment coupling. Using Amazon Simple Queue Service (Amazon SQS) as the message broker, Asset Watchers replace polling-based sensors with event-driven triggers. This approach reduces orchestration latency from minutes to seconds and reclaims worker resources previously consumed by polling sensors. It also improves message reliability, because Amazon SQS retains coordination signals even when the consumer environment is temporarily unavailable.

In this post, you learn how to design and deploy cross-account orchestration patterns using asset-based scheduling in Airflow 3.0 with Amazon SQS integration. You learn about Asset Watchers, how to publish asset events from producer DAGs, and how to trigger dependent workflows in downstream Amazon MWAA environments, creating responsive, decoupled pipelines that span multiple accounts.

If you use AI coding assistants to build and deploy infrastructure, the solution repository includes an agent skill built on the Agent Skills standard that encodes the architecture and best practices from this post.

Solution overview

This solution demonstrates a multi-MWAA orchestration architecture where:

1. Producer Amazon MWAA Environment (Account A) runs data processing workflows that publish asset events to an Amazon SQS queue when datasets are created or updated.
2. Amazon SQS Queue acts as a message broker, decoupling producer and consumer environments.
3. Consumer Amazon MWAA Environment (Account B) monitors the Amazon SQS queue using Asset Watchers and automatically triggers downstream DAGs when relevant asset events arrive.

Key benefits

This event-driven approach offers several advantages over traditional polling:

• No more polling overhead: You replace continuous sensor polling with event-driven Asset Watchers that respond as events arrive.
• Near real-time response: Downstream DAGs trigger within seconds rather than waiting for a scheduled polling interval.
• Independent environments: Producer and consumer Amazon MWAA environments have no direct dependencies, so each team can scale and update their environment without affecting the other.
• Reliable message delivery: Amazon SQS provides durable message delivery, even if the consumer environment is temporarily unavailable.
• Clear team ownership: You and your team maintain your own Amazon MWAA environment while still coordinating complex cross-account workflows.
• Faster implementation: Describe requirements in natural language and the agent skill generates deployment-ready producer and consumer DAGs with the best practices from this post built in.

Architecture overview

The following architecture shows how you can connect separate Amazon MWAA environments across AWS accounts so that a completed pipeline in one environment automatically triggers dependent workflows in another, without direct environment coupling or polling overhead.

Figure 1: Cross-account event-driven orchestration between Amazon MWAA environments using Amazon SQS

Architecture components

The architecture has four main components. The producer DAG defines assets as outlets and publishes events to an Amazon SQS queue when tasks complete successfully. The Amazon SQS queue acts as a durable message broker between accounts, with AWS Identity and Access Management (IAM) policies granting the producer permission to send messages and the consumer permission to receive them. On the consumer side, an Asset Watcher monitors the queue and updates asset state when messages arrive, which automatically triggers the consumer DAG scheduled on that asset.

Prerequisites

Before implementing this solution, you need:

• Two Amazon MWAA environments running Apache Airflow 3.0 or later, in the same or different AWS accounts. Each environment must have the triggerer component enabled.
• Intermediate knowledge of IAM policies, including cross-account role trust relationships and resource-based policies.
• Intermediate knowledge of Apache Airflow DAG authoring, including Python-based DAG definitions and task operators.
• Basic Python experience (Python 3.8 or later) to read and adapt the provided code samples.
• An Amazon SQS standard queue with cross-account permissions configured (see the Cross-account IAM section).
• AWS Command Line Interface (AWS CLI) configured with credentials that have permission to access both Amazon MWAA environments and the Amazon SQS queue.
• Time to complete: Approximately 90 minutes (following the GitHub repository instructions).
• Estimated cost: Running two Amazon MWAA environments and an Amazon SQS queue will incur AWS charges. Refer to the Amazon MWAA pricing page and Amazon SQS pricing page to estimate costs for your Region and usage. Remember to delete resources when you finish to avoid ongoing charges.

Implementation

The post includes a GitHub repository where you can deploy the solution described in this post. You will follow the implementation steps from setting up Amazon MWAA environments and cross-account Amazon SQS queues to deploying producer and consumer DAGs with Asset Watchers. This post provides the code samples, including the DAG files, IAM policies, and requirements configuration, for demonstration purposes only. Before deploying to production, verify that you conduct thorough testing, security reviews, and validation against the specific requirements and compliance standards.

Considerations

• Asset Watchers run as background processes in the Airflow triggerer, not the scheduler. Verify that the triggerer is healthy and running in consumer Amazon MWAA environment before expecting event-driven DAG triggers. If the triggerer is down, Amazon SQS messages will accumulate in the queue but won’t trigger downstream DAGs until the triggerer recovers. For more information, read the Asset Watchers documentation.
• Amazon SQS messages have a default retention period of 4 days (configurable up to 14 days). If the consumer environment is unavailable for longer than the retention period, messages will be lost. Consider configuring a dead-letter queue to capture messages that fail processing, and adjust the MessageRetentionPeriod based on recovery requirements.
• Cross-account Amazon SQS access requires both an IAM identity policy on the producer’s execution role and a resource-based policy on the Amazon SQS queue. If either policy is missing or misconfigured, message delivery will silently fail. For guidance on cross-account access patterns, refer to Four ways to grant cross-account access on AWS.
• Set the Amazon SQS VisibilityTimeout higher than the expected time for the Asset Watcher to process a message. If the timeout is too short, messages might be redelivered and trigger duplicate DAG runs. Review the Amazon SQS visibility timeout documentation when tuning this value.
• Each Amazon MWAA environment has limits on the number of DAGs, triggerers, and concurrent DAG runs. If you plan to scale to multiple Asset Watchers monitoring different Amazon SQS queues, check the current Amazon MWAA quotas before making design decisions.
• Asset URIs must match exactly between the Asset Watcher definition and the consumer DAG’s schedule parameter. A mismatch, even in casing or trailing characters, will prevent the consumer DAG from being triggered. Define assets in a single DAG file to avoid inconsistencies.
• Pin the provider packages apache-airflow-providers-amazon and apache-airflow-providers-common-messaging to versions compatible with Airflow. Incompatible versions might cause import errors that prevent the triggerer from starting. Use a constraints file as described in this post to avoid dependency conflicts.

Agent skills

AI coding assistants are most useful when they have context about your specific architecture and constraints, not only general programming patterns. Agent Skills, originally developed by Anthropic and released as a public standard in December 2025, provides a portable format for this need. SKILL.md files encode procedural knowledge, best practices, and workflows so that compatible AI coding agents can discover and apply them on demand. The standard is now supported by Kiro, Strands Agents, Anthropic Claude Code, OpenAI Codex, Cursor, Gemini CLI, and other tools. The solution provided here includes an agent skill (agent-skill/) built on this standard that encodes the cross-account orchestration architecture and operational best practices from this post. When you tell the AI coding assistant something like “Write cross-account Amazon MWAA DAGs for my orders pipeline”, the skill guides the agent through the complete workflow:

• Collecting Amazon SQS queue URL.
• Generating correctly structured producer and consumer DAG files.
• Optionally deploying them to Amazon MWAA environments.

The skill doesn’t require you to provide AWS account IDs or Amazon MWAA environment names upfront. Instead, it auto-discovers your environments by running aws mwaa list-environments and aws sts get-caller-identity using the locally configured AWS CLI credentials, then asks you to confirm which environment is the producer and which is the consumer.

The skill works in two modes:

• Sample mode: Generates the reference producer and consumer DAGs for quick cross-account validation, requiring only the Amazon SQS queue URL as input.
• Custom mode: Adapts the DAG templates to specific business logic. For example, the producer runs an AWS Glue extract, transform, and load (ETL) job and the consumer triggers a data build tool (dbt) model refresh. This mode customizes DAG IDs, task names, schedules, and processing logic while preserving the correct Asset Watcher patterns.

Beyond code generation, the skill includes an auto-deploy flow. This flow discovers existing Amazon MWAA environments, runs pre-flight checks (Amazon Virtual Private Cloud (Amazon VPC) networking, provider versions, triggerer health, and Amazon SQS queue accessibility), uploads DAGs to the correct Amazon Simple Storage Service (Amazon S3) buckets, and verifies end-to-end readiness. Each step that modifies infrastructure requires explicit user confirmation. Also refer to the GitHub repository for instructions on using it.

Best practices

Airflow Asset Watchers with Amazon SQS are not always the right fit. When they are, they introduce operational considerations that differ from sensor-based polling approaches.

This section covers how to choose the right cross-environment orchestration pattern, how to configure the infrastructure that Asset Watchers depend on (IAM, Amazon VPC, dependencies), and how to design producer and consumer DAGs that are reliable in production.

Cross-account IAM

• Producer execution role needs sqs:SendMessage and sqs:GetQueueUrl scoped to the specific queue ARN to avoid sqs:*.
• Amazon SQS queue resource policy must allow the producer role for sqs:SendMessage and consumer role for sqs:ReceiveMessage, sqs:DeleteMessage, sqs:GetQueueAttributes, and sqs:GetQueueUrl.
• Test cross-account access with the AWS CLI before deploying DAGs. Debugging AWS IAM through Airflow task logs is much harder and slower than catching misconfigurations at the CLI level.
• Enable Amazon SQS server-side encryption for production queues.

Triggerer health

• Airflow Asset Watchers run in the triggerer, not the scheduler. Verify triggerer health in the Airflow UI after deploying consumer DAGs.
• The health API can report healthy even when components are broken. Cross-check by verifying Amazon CloudWatch log streams exist for the Triggerer log group.
• Monitor airflow-<ENV>-Triggerer CloudWatch logs for ClientError, QueueDoesNotExist, or ImportError.
• Set Amazon CloudWatch alarms on Amazon SQS ApproximateNumberOfMessagesVisible and the depth of your dead-letter queue (DLQ), which captures messages that fail processing after the maximum number of receive attempts.
• Pin provider versions with a constraints file to prevent dependency conflicts.

Amazon VPC networking

• Private subnets must route 0.0.0.0/0 to a NAT Gateway. Without it, workers and triggerers silently fail while the web server appears healthy.
• Use two NAT Gateways (one per Availability Zone) for production high availability.
• For private routing mode, use Amazon VPC Endpoints (Amazon S3, Amazon SQS, Amazon CloudWatch Logs, and Amazon Elastic Container Registry (Amazon ECR)) instead of NAT.
• Confirm Amazon CloudWatch log streams exist for Scheduler, Worker, DAGProcessing, and Triggerer. Empty log groups mean containers aren’t running.
• Security group must allow self-referencing inbound traffic and unrestricted outbound.

Dependency management

• Pin provider versions with == and use a constraints file. Unpinned versions break on environment updates.
• Test dependencies locally with MWAA Docker images before deploying.
• Check the requirements_install_ip log stream after updates. If networking was unavailable at creation, force reinstall with a new requirements-s3-object-version.
• Review pre-installed base packages before adding to requirements.txt to avoid version conflicts.

Choosing an orchestration pattern

Not every cross-environment dependency warrants an Asset Watcher. Airflow 3.0 offers three main orchestration patterns: Asset Watchers with Amazon SQS, the MwaaTriggerDagRunOperator, and sensor-based polling, each with different trade-offs in response time, coupling, and resource consumption. Use the following table to match your use case to the right pattern before committing to an implementation.

• Avoid wiring persistent-state triggers (for example, S3KeyTrigger) into Asset Watchers. They fire continuously because the condition never clears.

DAG authoring

• Minimize module-level code. DAG files are re-parsed every cycle, and heavy imports slow the entire parsing loop.
• Design tasks so they produce the same result whether they run once or multiple times (a property called idempotency). Duplicate Amazon SQS messages can occur on retries, so prefer UPSERT (insert or update) over INSERT to avoid duplicate records.
• Keep secrets out of DAG files and message bodies. Use Airflow Connections (aws_conn_id) instead.
• Test DAG imports locally with python your_dag.py before uploading to S3.
• Allow time for DAG parsing after S3 upload, or force with dags reserialize.

Producer DAG design

• Include dag_id, run_id, logical_date, and dataset-specific context in Amazon SQS messages so consumers can route without calling back.
• Use SqsHook instead of the raw boto3 package. It respects aws_conn_id and integrates with Airflow logging.
• Let publish failures raise so the Airflow retry mechanism handles redelivery.

Consumer DAG design

• Access messages through triggering_asset_events, not by reading the queue directly. The Asset Watcher has already consumed the Amazon SQS messages.
• Validate message payloads defensively. Producers might evolve their schema over time.
• Use conditional asset scheduling (& / |) for complex multi-asset dependencies.

Clean up resources

To avoid ongoing AWS charges, delete the resources you created as part of this solution when you are done. The GitHub repository includes step-by-step cleanup instructions for removing the Amazon SQS queue, Amazon MWAA environments, IAM roles and policies, and Amazon S3 buckets.

Refer to the cleanup instructions in the GitHub repository to remove the provisioned resources.

Conclusion

Asset-based scheduling in Apache Airflow 3.0, with Asset Watchers, gives you a practical way to coordinate workflows across Amazon MWAA environments without polling overhead or tight coupling. By using Amazon SQS as a reliable message broker, you can build responsive, decoupled data pipelines that span multiple Amazon MWAA environments and AWS accounts without the operational overhead of traditional polling mechanisms.

This approach reduces cross-environment orchestration latency from minutes to seconds, replaces custom sensors with declarative asset-based scheduling, and gives you and your team the flexibility to maintain independent Amazon MWAA environments while still coordinating complex workflows. Amazon SQS durable message delivery reduces the risk of lost signals, even during temporary environment outages.

To get started:

1. Review the architecture (5 minutes): Open the architecture diagram in the repository and confirm which Amazon MWAA environments will be the producer and which will be the consumer.
2. Set up the Amazon SQS queue (15 minutes): Create a cross-account Amazon SQS standard queue and apply the IAM identity and resource-based policies from the Cross-account IAM section. Verify access with the AWS CLI before proceeding.
3. Deploy and validate the DAG examples (30 minutes): Copy the producer and consumer DAG snippets from the Implementation section into Amazon MWAA environments, trigger the producer DAG manually, and confirm the consumer DAG runs automatically.
4. Run pre-flight checks (20 minutes): Work through the Amazon VPC networking, provider version, and triggerer health checks in the Best Practices section. Confirm Amazon CloudWatch log streams exist for the Triggerer log group before declaring the environment ready.
5. Optionally, use the agent skills: If you use an AI coding assistant, install the skill from the repository and describe the business logic in natural language to generate deployment-ready DAGs tailored to your pipeline.

As you scale data operations across multiple accounts and AWS Regions, asset-based scheduling with Asset Watchers provides the foundation for building modern, event-driven data architectures on AWS. Start with basic producer-consumer patterns and gradually evolve to complex multi-asset dependencies as orchestration requirements grow.

For more information, refer to

• Apache Airflow 3 on Amazon MWAA Launch Blog Post.
• Apache Airflow documentation.
• Amazon MWAA User Guide.
• Amazon SQS Documentation.
• AWS IAM cross-account access.
• Amazon CloudWatch monitoring for Amazon MWAA.
• AWS Step Functions for orchestration patterns that require human approval steps or complex branching logic outside Airflow.
• AWS Well-Architected Framework — Data Analytics Lens.
• Amazon MWAA quotas.
• Amazon VPC networking best practices.

About the authors

Post Syndicated from Shraddha Anil Naik original https://aws.amazon.com/blogs/big-data/building-a-scalable-personalized-recommendation-system-on-aws-from-batch-to-real-time/

Amazon.com receives millions of visits every day, and behind every product recommendation on our website is a system that needs to process customer signals, run machine learning (ML) models, and deliver results before the next visit. Doing this across global marketplaces for millions of customers at tens of thousands of requests per second, while keeping experimentation fast and infrastructure costs bounded, is an orchestration challenge as much as a machine learning one.

Our team built a system that addresses this challenge. This post shows how we did it using a batch-first architecture with AWS Lake Formation, Amazon Managed Workflows for Apache Airflow (Amazon MWAA), Amazon Athena, AWS Glue, Amazon SageMaker, and Amazon DynamoDB, and how we later extended it with Amazon MemoryDB for real-time vector similarity search when we needed to incorporate more real-time signals.

Architecture overview

Data flows from the centralized data lake (Lake Formation and Athena) through Airflow-orchestrated pipelines using Glue for processing and SageMaker for ML workloads, into DynamoDB for batch serving and MemoryDB for real-time inference

The data lake foundation: Centralized access with Lake Formation

Every recommendation pipeline starts with data. We built a centralized data lake to create a single source of truth that any pipeline or consumer can access without duplicating data or building bespoke extract, transform, and load (ETL) pipelines.

Golden datasets: Shared once, used everywhere

Before the data lake, each recommendation pipeline independently extracted and transformed its own copy of product catalog, transaction history, and embeddings. This led to subtle inconsistencies: one pipeline might use a slightly different join logic or a stale snapshot, making it difficult to compare model performance or debug discrepancies across pipelines.

Now, we publish curated, validated datasets once, and every consumer (Airflow DAGs, ML notebooks, analytics dashboards) reads from the same tables through the same governed access. This means:

• New pipelines start faster. A new recommendation model does not need its own data extraction logic. It queries the existing golden datasets from day one.
• Consistency across models. When we compare model A to model B, we know both models are trained and inferred on the same underlying data.
• Cross-team collaboration. Multiple teams share the same tables as a single source of truth.
• Two-way data flow. The data lake serves as both source and destination. Our pipelines read golden datasets as inputs and write computed outputs (model scores, feature sets, intermediate results) back to the lake, where they become inputs for other pipelines. This creates a compounding effect: each new pipeline enriches the lake for the next one.

Why Lake Formation?

Our data is stored in Amazon Simple Storage Service (Amazon S3), partitioned by marketplace. Our data consumers (Airflow pipelines, ML notebooks, analytics tools) live in separate AWS accounts from the data producers. We chose AWS Lake Formation because it adds governance on top of S3 without requiring data migration:

• Fine-grained cross-account access. Grant table-level permissions per consumer role, without managing bucket policies manually.
• Schema governance through AWS Glue Data Catalog. Scheduled Glue crawlers infer schemas from files in S3, keeping the catalog current as data evolves.
• Multi-Region consistency. We deploy identical infrastructure across multiple AWS Regions using AWS Cloud Development Kit (AWS CDK), each Region serving its local marketplaces.

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

We chose Amazon Managed Workflows for Apache Airflow (Amazon MWAA) as our orchestration layer. MWAA removes the operational burden of managing Airflow infrastructure: automatic scaling of workers, built-in high availability, and managed upgrades mean our team focuses on pipeline logic rather than cluster maintenance. MWAA lets us define complex multi-step workflows with rich dependencies in Python code, and its operator model lets us encapsulate team-specific conventions into reusable building blocks.

Each recommendation pipeline follows a consistent pattern:

We built a library of reusable custom Airflow operators, each encapsulating one of our core compute engines. This reduced new pipeline development from weeks to days in our team’s experience.

Athena and AWS Glue: Data access and processing

Amazon Athena is how our pipelines read from Lake Formation. Our custom operator runs SQL queries against the Glue Data Catalog and automatically runs UNLOAD to write results to S3: serverless, no infrastructure to manage, and integrated with the Lake Formation permission model.

AWS Glue handles compute-intensive data transformations through PySpark: joining datasets, filtering, deduplication, aggregation, and formatting ML outputs into final recommendation lists. We configure Glue with Auto Scaling worker pools (for example, 2–50 workers of G.8X type) so jobs scale with data volume per marketplace. All jobs run ephemerally: they read from S3, write to S3, and require no long-running clusters.

Amazon SageMaker: Training, inference, and vector search

Amazon SageMaker powers the ML-intensive stages of our pipelines across three workload types, all orchestrated as steps within our Airflow DAGs:

Training

We use SageMaker Training Jobs to train our recommendation models on GPU instances (for example, ml.g5). Training data is prepared by upstream Glue jobs and staged in S3. Our custom Airflow operator submits the training job, monitors its progress, and registers the resulting model artifact in S3. Once training completes, the artifact is immediately available for batch inference or endpoint deployment within the same DAG run. This means a single DAG can go from raw data to trained model to deployed inference without manual handoffs, and we can retrain it on fresh data every pipeline cycle with zero operator intervention.

Batch inference

SageMaker Batch Transform runs our trained models at scale, generating the outputs that feed into downstream ranking and publishing steps. Our batch inference operator handles job submission, polls for completion, and writes the output location to the DAG’s S3 convention so the next Glue step can pick it up automatically. Batch Transform lets us run inference without provisioning persistent infrastructure, and we can scale instance count and type independently for every pipeline based on data volume.

Vector search

Generating recommendations for millions of customers requires searching across hundreds of thousands of candidate products per customer. Exact search at this scale is prohibitively expensive, so we use approximate nearest neighbor (ANN) search using FAISS to find similar products efficiently, run as SageMaker Processing Jobs.

The workflow:

1. Build a FAISS index over the candidate catalog.
2. Query the index with per-customer vectors to find top-K nearest neighbors.
3. Return ranked candidate lists per customer.

We distribute query vectors across the SageMaker Processing fleet using S3-based sharding (ShardedByS3Key). Each instance receives the full candidate index but only a fraction of the query vectors. Every instance builds an identical FAISS index, searches its shard of queries, and writes results to S3. The downstream Glue step merges all shards into the final recommendation lists. This lets us scale horizontally by adding instances without changing any code.

Why SageMaker inside MWAA?

Running SageMaker jobs as MWAA tasks (rather than standalone) gives us:

• End-to-end lineage. Every model training run, inference job, and vector search is tracked as part of a DAG execution. We can trace a recommendation in DynamoDB back to the exact training run, data snapshot, and ANN search that produced it.
• Retry and failure handling. If a SageMaker job fails (spot instance preemption, transient capacity errors), Airflow retries it automatically with backoff. No manual re-runs.
• Resource sequencing. Training must finish before inference, inference before ANN search. The Airflow dependency model handles this naturally without polling scripts or step function state machines.
• Unified monitoring. One Airflow dashboard shows the health of all pipelines: Glue ETL, SageMaker training, SageMaker inference, and DynamoDB publishing. No context-switching between consoles.

Putting it together: A complete pipeline example

Our recommendation generation pipeline illustrates the full flow:

Note that the operators shown (GlueSQLOperator, VectorSearchOperator, and others) are custom internal operators built on top of the Airflow AWS provider, not open-source libraries. Here is a simplified version of what this looks like in code:

from airflow import DAG
from airflow.models.baseoperator import chain
from airflow.utils.task_group import TaskGroup

dag = DAG("recommendation_generator", schedule="0 18 * * 4")  # Weekly
task_groups = []

for marketplace in [...]:  # global marketplaces
    with TaskGroup(group_id=marketplace, dag=dag) as group:

        # Step 1: Extract data from lake
        candidates = GlueSQLOperator(
            task_id="select_candidates",
            tables=[f"{marketplace}.catalog", f"{marketplace}.products"],
            sql="select_candidates.sql", dag=dag)

        customer_history = GlueSQLOperator(
            task_id="select_customer_history",
            tables=[f"{marketplace}.transactions"],
            sql="select_customer_vectors.sql", dag=dag)

        customers = AthenaSQLOperator(
            task_id="select_customers", database=marketplace,
            query="select_customer_cohort.sql", dag=dag)

        # Step 2: Train model
        train = ModelTrainingOperator(
            task_id="train_model",
            training_input={"task_id": customer_history.task_id},
            instance_type="ml.g5", dag=dag)

        # Step 3: Batch inference
        inference = BatchInferenceOperator(
            task_id="batch_inference",
            model={"task_id": train.task_id},
            input_data={"task_id": candidates.task_id}, dag=dag)

        # Step 4: Vector search via SageMaker Processing Job
        ann_search = VectorSearchOperator(
            task_id="ann_search",
            index_input={"task_id": inference.task_id},
            query_input={"task_id": customer_history.task_id},
            k=60, instance_count=10, dag=dag)

        # Step 5: Merge and rank with Glue
        merge_and_rank = GlueSparkOperator(
            task_id="merge_and_rank",
            script="merge_results.py", dag=dag)

        # Step 6: Publish to DynamoDB
        publish = DynamoDBPublishOperator(
            task_id="publish_recommendations",
            table_name="Recommendations", dag=dag)

        # Task dependencies
        [candidates, customer_history] >> train >> inference
        [inference, customers] >> ann_search >> merge_and_rank >> publish

    task_groups.append(group)

chain(*task_groups)  # Execute marketplaces sequentially

Key patterns in this DAG

• Marketplace isolation. Each marketplace runs in its own TaskGroup. A failure in one does not block the others.
• Parallel data extraction. Independent Athena/Glue queries run concurrently before converging at the training step.
• Sequential marketplace execution. chain() runs marketplaces one at a time to avoid resource contention across large SageMaker and Glue jobs.
• Reusable operators. We built custom operators like GlueSQLOperator, VectorSearchOperator, and DynamoDBPublishOperator that encode our team’s conventions (cross-account access, S3 staging, retry logic, metrics) into shared building blocks. New pipelines are mostly configuration rather than infrastructure code, and these operators are shared across dozens of pipelines.
• Implicit data passing. Each operator writes its output to a convention-based S3 path and downstream operators automatically resolve the upstream output location. No hard-coded paths between steps.

This pattern powers hundreds of pipelines across global marketplaces, with each marketplace as an isolated TaskGroup in the DAG.

Serving layer: DynamoDB + ECS

Our Java-based Amazon Elastic Container Service (Amazon ECS) service reads pre-computed recommendations from Amazon DynamoDB at request time:

1. Receive request with customer ID, marketplace, customer context, and page context.
2. Read pre-computed recommendations from DynamoDB.
3. Apply real-time filters (availability, eligibility constraints).
4. Re-rank based on real-time context signals.
5. Return the response.

Amazon DynamoDB is designed to provide single-digit millisecond reads at our scale and time-to-live (TTL) for automatic cleanup of stale recommendations.

Extending to real-time

In recommendation system terms, our batch pipeline handles the retrieval stage offline. Although this covers the majority of our traffic, we identified scenarios where weekly freshness was not enough to capture what customers are doing right now. The real-time extension adds an online retrieval and ranking path for signals that cannot wait for the next batch cycle.

To address these, we extended our system with Amazon MemoryDB (which now supports Valkey as its open-source engine) for real-time vector similarity search and SageMaker real-time endpoints for on-demand embedding generation. The same Airflow pipelines that publish to DynamoDB also publish product vectors to MemoryDB (through our MemoryDB publish operator), and the same model in our batch pipeline is deployed to a SageMaker endpoint for single-item inference at request time.

At serving time, when we want to incorporate fresh signals, the service calls the SageMaker endpoint to generate an embedding on the fly, then queries MemoryDB for the nearest neighbors. These fresh signals include recent search queries, cart additions, and other in-session activity that changes faster than our weekly batch cycle. In our workloads, this gives us sub-millisecond vector search latency without re-running the full batch pipeline. Critically, the batch pipeline keeps the MemoryDB product index fresh. Our Airflow DAG includes a MemoryDB publish operator that refreshes the full product vector index weekly, so real-time queries always search against an up-to-date index.

Batch and real-time are not competing approaches: batch handles slow-moving signals (purchase history, catalog relationships) while real-time handles fast-moving ones (current session, new arrivals, trending items). Both paths share the same models, the same data lake, and the same serving service.

For a detailed deep-dive on this real-time architecture, see Real-time personalized recommendations with Amazon SageMaker and Amazon Managed Valkey.

Security and access control

Security is a foundational concern for a system that spans multiple AWS accounts, processes customer behavioral data, and runs across multiple AWS Regions.

Cross-account access: Lake Formation grants are issued to consumer-account AWS Identity and Access Management (IAM) roles, ensuring consumers can query tables without direct S3 bucket access. Each consumer role receives only the permissions it needs for its specific tables.

IAM execution roles: Each compute engine (MWAA, Glue, SageMaker) runs under a dedicated least-privilege IAM role.

Network isolation: MWAA environments and the ECS serving layer are deployed within virtual private clouds (VPCs), with separate VPC configurations per Region.

Reliability and failure handling

Regional isolation: Each AWS Region runs an independent copy of the system. DynamoDB tables are regional, each populated by the local batch pipeline. MemoryDB clusters are regional, with the product vector index refreshed by the local Airflow DAG. This means a regional failure or pipeline delay in one Region does not affect other Regions.

Batch pipeline failures: If a pipeline fails mid-run, the previous DynamoDB data remains live and continues serving recommendations until the next successful run. Airflow retries failed tasks automatically with configurable backoff. TTL on DynamoDB records bounds how long stale data persists. Failed pipeline runs trigger automated alerting so the on-call engineer can investigate.

Real-time path resilience: MemoryDB is deployed in a multi-AZ configuration with automatic failover. The real-time path is an extension of the batch system, and batch recommendations from DynamoDB remain available regardless of real-time path availability.

Lessons learned

1. Start batch, add real-time incrementally. Batch pipelines are easier to debug, cheaper to operate, and sufficient for most recommendation scenarios. Add real-time path only when you have clear indication that specific customer signals (for example, in-session activity, search queries) need sub-hour freshness to remain relevant.
2. Match the serving path to signal velocity. Not every signal needs real-time processing. Categorize your signals by how quickly they change, and route accordingly: batch for slow signals, real-time for fast ones, re-ranking for in-between.
3. Freshness does not always require re-computation. A batch-generated candidate set remains largely valid between runs. What changes is relative relevance. Re-ranking at serving time with recent customer activity gives the impression of real-time without the cost of real-time candidate generation.
4. A centralized data lake accelerates everything. Golden datasets eliminated weeks of per-pipeline data extraction work, made model comparisons trustworthy, and let new team members ship their first pipeline in days instead of weeks. The upfront investment in Lake Formation governance paid for itself within the first quarter.
5. Invest in reusable operators. Custom Airflow operators encapsulating Athena/Glue/SageMaker patterns let teams ship new pipelines in days. The operators encode best practices (retry logic, cross-account access, metrics) so pipeline authors can focus on business logic.
6. Separate compute from storage. S3 as the universal intermediate layer + ephemeral Glue/SageMaker jobs means you pay only for active computation. No idle clusters between weekly pipeline runs.

Conclusion

We built this system because every customer interaction is an opportunity to surface the right product at the right time. Serving tens of thousands of requests per second across millions of customers in global marketplaces, our batch-first architecture uses AWS Lake Formation for governed data access, Amazon MWAA for orchestration, Amazon Athena for data lake queries, AWS Glue for distributed processing, Amazon SageMaker for training, inference, and vector search, and Amazon DynamoDB for low-latency serving. When we needed to incorporate more real-time signals, we added Amazon MemoryDB vector search paired with SageMaker real-time endpoints for on-demand embedding generation, extending into real-time without replacing the batch foundation.

The architecture choices we have made, batch for efficiency and real-time for freshness, all serve the goal of helping customers discover what they need faster. If you are building personalized experiences at scale, we hope these patterns give you a useful starting point.

This would not have been possible without the Everyday Essentials engineering team, whose collective effort turned these ideas into a production system serving customers every day. We are also grateful for the support and guidance from Sam Heyworth, Nirav Desai, and Ankur Datta, and the broader Everyday Essentials leadership team.

About the authors