Why This Matters

SQS and SNS are the foundational messaging services that enable decoupled, resilient data architectures on AWS. SQS (Simple Queue Service) is a pull-based message queue that buffers workloads between producers and consumers. SNS (Simple Notification Service) is a push-based pub/sub system that broadcasts events to multiple subscribers simultaneously. Together, they form the backbone of event-driven data pipelines that can handle millions of messages per second with independent scaling, retry logic, and dead letter queue isolation.

Understanding when to use each service, or how to combine them, is essential for building scalable data platforms. SQS excels at task distribution and load leveling. SNS excels at fan-out and broadcasting. The SNS-to-SQS fan-out pattern is the recommended architecture for multi-consumer data pipelines, providing independent buffering, retry, and failure isolation per subscriber.

Architecture Overview

Real-World Project Structure

sqs-sns-data-pipeline/
├── infrastructure/
│   ├── cdk/
│   │   ├── app.py
│   │   ├── stacks/
│   │   │   ├── sns_stack.py
│   │   │   ├── sqs_stack.py
│   │   │   ├── lambda_consumers_stack.py
│   │   │   └── monitoring_stack.py
│   │   └── cdk.json
│   └── terraform/
│       ├── main.tf
│       ├── sns_topics.tf
│       ├── sqs_queues.tf
│       └── variables.tf
├── lambda_functions/
│   ├── analytics_consumer/
│   │   ├── app.py
│   │   └── requirements.txt
│   ├── inventory_consumer/
│   │   ├── app.py
│   │   └── requirements.txt
│   ├── ml_feature_consumer/
│   │   ├── app.py
│   │   └── requirements.txt
│   └── dlq_processor/
│       ├── app.py
│       └── requirements.txt
├── filter_policies/
│   ├── analytics_filter.json
│   ├── inventory_filter.json
│   └── ml_feature_filter.json
├── tests/
│   ├── unit/
│   ├── integration/
│   └── test_fan_out.py
├── monitoring/
│   ├── dashboards/
│   └── alarms/
└── README.md

SQS Queue Configuration

import boto3
import json

sqs_client = boto3.client('sqs')

def create_data_pipeline_queue(queue_name, dlq_name):
    """
    Create a production-ready SQS queue for data pipeline processing.
    """
    # Create DLQ first
    dlq_response = sqs_client.create_queue(
        QueueName=dlq_name,
        Attributes={
            'MessageRetentionPeriod': '1209600',  # 14 days
        }
    )
    dlq_arn = sqs_client.get_queue_attributes(
        QueueUrl=dlq_response['QueueUrl'],
        AttributeNames=['QueueArn']
    )['Attributes']['QueueArn']

    # Create main queue with DLQ
    queue_response = sqs_client.create_queue(
        QueueName=queue_name,
        Attributes={
            'VisibilityTimeout': '300',  # 5 minutes
            'MessageRetentionPeriod': '1209600',  # 14 days
            'ReceiveMessageWaitTimeSeconds': '20',  # Long polling
            'RedrivePolicy': json.dumps({
                'deadLetterTargetArn': dlq_arn,
                'maxReceiveCount': '3'
            })
        }
    )

    return {
        'queue_url': queue_response['QueueUrl'],
        'dlq_url': dlq_response['QueueUrl']
    }

Batch Operations

def send_batch_messages(queue_url, messages):
    """
    Send messages in batches of 10 for efficiency.
    """
    batch_size = 10
    total_sent = 0

    for i in range(0, len(messages), batch_size):
        batch = messages[i:i + batch_size]

        entries = [
            {
                'Id': str(idx),
                'MessageBody': json.dumps(msg),
                'MessageAttributes': {
                    'source': {'DataType': 'String', 'StringValue': 'data-pipeline'},
                    'priority': {'DataType': 'String', 'StringValue': 'high'}
                }
            }
            for idx, msg in enumerate(batch)
        ]

        try:
            response = sqs_client.send_message_batch(
                QueueUrl=queue_url,
                Entries=entries
            )
            total_sent += len(response.get('Successful', []))

            if response.get('Failed'):
                print(f"Failed messages: {response['Failed']}")

        except Exception as e:
            print(f"Batch send error: {str(e)}")
            raise

    return total_sent

SNS Fan-Out Pattern

The SNS-to-SQS fan-out pattern is the recommended architecture for multi-consumer data pipelines:

import boto3
import json

sns_client = boto3.client('sns')

def create_fan_out_architecture():
    """
    Create SNS topic with SQS subscribers for fan-out pattern.
    """
    # Create SNS topic
    topic = sns_client.create_topic(Name='order-events')
    topic_arn = topic['TopicArn']

    # Subscriber configurations
    subscribers = [
        {
            'name': 'analytics-queue',
            'filter': {'event_type': ['order_completed', 'order_shipped']},
            'max_receive_count': 3
        },
        {
            'name': 'inventory-queue',
            'filter': {'event_type': ['order_completed']},
            'max_receive_count': 5
        },
        {
            'name': 'ml-features-queue',
            'filter': {'event_type': ['order_completed', 'order_cancelled']},
            'max_receive_count': 3
        },
        {
            'name': 'compliance-queue',
            'filter': {'event_type': ['order_completed', 'order_refunded']},
            'max_receive_count': 3
        }
    ]

    for sub in subscribers:
        # Create SQS queue with DLQ
        queue = create_data_pipeline_queue(
            f"{sub['name']}-main",
            f"{sub['name']}-dlq"
        )

        # Subscribe with filter policy
        sns_client.subscribe(
            TopicArn=topic_arn,
            Protocol='sqs',
            Attributes={
                'FilterPolicy': json.dumps(sub['filter'])
            },
            Endpoint=queue['queue_url']
        )

    return topic_arn

Message Filtering

SNS message filtering allows subscribers to receive only messages matching specific criteria:

{
  "FilterPolicy": {
    "event_type": ["order_completed", "order_shipped"],
    "region": ["us-east-1", "us-west-2"],
    "amount": [{"numeric": [">", 100]}]
  }
}

Dead Letter Queue Configuration

Performance Considerations

Security Considerations

• IAM Policies: Use least-privilege policies for queue and topic access. Separate producers and consumers.
• Encryption: Enable server-side encryption (SSE-SQS or SSE-KMS) for sensitive data in queues.
• VPC Endpoints: Use VPC endpoints for SQS and SNS to keep traffic off the public internet.
• Queue Policies: Use resource-based policies to restrict cross-account access.
• Message Attributes: Never include sensitive data in message attributes (visible in console).
• CloudTrail: Log all SQS/SNS API calls for audit and compliance.
• DLQ Access Control: Restrict DLQ access to only authorized recovery processes.
• FIFO Deduplication: Use message deduplication ID to prevent duplicate processing.

Interview Questions & Answers

Q1: When would you choose SQS over SNS?

Answer: Choose SQS when you need point-to-point messaging where one consumer processes each message (task distribution across worker nodes). SQS excels at buffering workloads where producers and consumers operate at different rates, providing message persistence, retry logic with visibility timeouts, and dead letter queues. Use SNS when you need one event to reach multiple subscribers simultaneously (fan-out for analytics, notifications, and archival). SNS is push-based, so subscribers do not need to poll. In practice, they are often combined: SNS fans out to multiple SQS queues, each serving a different downstream system.

Q2: Explain the SQS visibility timeout and why it matters for data pipelines.

Answer: When a consumer receives a message, SQS hides it from other consumers for the visibility timeout duration. This prevents duplicate processing. For data pipelines, set the timeout to at least 6x your maximum processing time. If processing takes longer than the timeout, the message becomes visible again and could be processed twice, causing duplicate records in your data warehouse. For example, if your ETL job takes 5 minutes, set the visibility timeout to 30 minutes. Use the ChangeMessageVisibility API to extend the timeout if processing is still in progress.

Q3: How do you handle duplicate messages in a data pipeline?

Answer: Use FIFO queues for exactly-once processing when message order matters. For standard queues (at-least-once delivery), implement idempotency in your consumer: store processed message IDs in DynamoDB with conditional writes, check before processing. Use message attributes to include a unique event ID for deduplication. For batch processing, maintain a sliding window of processed IDs with TTL. Monitor CloudWatch metrics for duplicate processing patterns and tune visibility timeout accordingly.

Q4: Describe a fan-out architecture for a real-time analytics pipeline.

Answer: Events publish to an SNS topic. Four SQS queues subscribe with filter policies: one for real-time dashboard updates via Kinesis, one for batch analytics via Glue/Redshift, one for ML feature updates via SageMaker, and one for archival in S3 via Glacier. Each subscriber has independent DLQs and scaling policies. Message filtering ensures each queue receives only relevant events (e.g., analytics queue filters for event_type = ["order_completed"]). This architecture handles 100k+ events per second with independent scaling per subscriber, and one subscriber's failure does not affect others.

Q5: What metrics should you monitor for SQS-based data pipelines?

Answer: Monitor ApproximateNumberOfMessagesVisible for queue depth (growing queue means consumers cannot keep up). Track ApproximateAgeOfOldestMessage to detect processing delays (if oldest message age exceeds your SLA, investigate immediately). Set alarms on DLQ ApproximateNumberOfMessagesVisible for failed messages (even one DLQ message deserves investigation). Monitor NumberOfMessagesSent and NumberOfMessagesReceived for throughput parity. Use ApproximateNumberOfMessagesNotVisible to detect stuck consumers (messages received but not deleted). For FIFO queues, monitor NumberOfMessagesDeleted for throughput.

Q6: How does SNS message filtering reduce costs in fan-out architectures?

Answer: Without filtering, every subscriber receives every message and must discard irrelevant ones, wasting compute and increasing processing time. With filter policies on subscriptions, SNS only delivers messages matching the subscriber criteria. This reduces Lambda invocations, SQS costs, and processing time. For example, a subscription filtering for event_type = ["order_completed"] ignores inventory updates and user profile changes. In high-volume scenarios, filtering can reduce costs by 60-80% for subscribers that only need a subset of events.

Q7: Explain the SNS + SQS fan-out pattern with DLQs.

Answer: An SNS topic fans out to multiple SQS queues, each serving a different downstream system. Each SQS queue has its own DLQ. If the inventory processor fails, messages go to its DLQ without affecting the analytics pipeline's DLQ. This isolation means one system's failure does not cascade. You can set different maxReceiveCount values per queue (critical financial queues might retry 5 times while non-critical logging retries 3 times). Use StartMessageMoveTask API to replay DLQ messages after fixing the root cause.

Q8: How do you test and replay failed messages from a DLQ?

Answer: First, fix the root cause of the processing failure. Then use StartMessageMoveTask API to move DLQ messages back to the source queue. This processes messages with the corrected consumer logic. Alternatively, use the SQS console or AWS CLI to send individual messages from the DLQ to the source queue for testing. Monitor CloudWatch metrics to confirm messages process successfully after replay. Implement a DLQ processor Lambda that can inspect, filter, and selectively replay messages based on error type. Always set a maximum message age for replay to avoid processing stale data.

Common Pitfalls

Advanced: SQS Extended Client for Large Messages

SQS has a 256 KB message size limit. The Extended Client Library stores large payloads in S3 and sends a reference in the queue:

import boto3
import json

s3_client = boto3.client('s3')
sqs_client = boto3.client('sqs')

def send_large_message(queue_url, large_payload, bucket_name):
    """
    Send messages larger than 256 KB using S3 as backing store.
    """
    payload_size = len(json.dumps(large_payload).encode('utf-8'))

    if payload_size > 250000:  # Near the 256 KB limit
        # Store payload in S3
        s3_key = f"sqs-payloads/{context.aws_request_id}.json"
        s3_client.put_object(
            Bucket=bucket_name,
            Key=s3_key,
            Body=json.dumps(large_payload).encode('utf-8'),
            ContentType='application/json'
        )

        # Send S3 reference as the message
        message_body = json.dumps({
            's3_reference': True,
            'bucket': bucket_name,
            'key': s3_key,
            'payload_size': payload_size
        })
    else:
        message_body = json.dumps(large_payload)

    response = sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody=message_body,
        MessageAttributes={
            'payload_size': {
                'DataType': 'Number',
                'StringValue': str(payload_size)
            },
            'has_s3_reference': {
                'DataType': 'String',
                'StringValue': 'true' if payload_size > 250000 else 'false'
            }
        }
    )

    return response['MessageId']

Advanced: Priority-Based Processing with SQS

import boto3
import json

sqs_client = boto3.client('sqs')

def send_priority_message(queue_url, message, priority='normal'):
    """
    Send messages with priority using message group IDs (FIFO queues).
    """
    # FIFO queues with message group IDs enable priority processing
    # High-priority messages use a separate group processed first
    message_group_id = f"priority-{priority}"

    response = sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps(message),
        MessageGroupId=message_group_id,
        MessageDeduplicationId=f"{message.get('id')}-{priority}",
        MessageAttributes={
            'priority': {
                'DataType': 'String',
                'StringValue': priority
            },
            'timestamp': {
                'DataType': 'Number',
                'StringValue': str(int(__import__('time').time()))
            }
        }
    )

    return response['MessageId']

Advanced: Multi-Region SQS Replication

import boto3
import json

def setup_multi_region_messaging():
    """
    Configure SNS with SQS in multiple regions for disaster recovery.
    """
    # Primary region SNS topic
    sns_primary = boto3.client('sns', region_name='us-east-1')
    topic = sns_primary.create_topic(Name='global-events')
    topic_arn = topic['TopicArn']

    # Create SQS queues in multiple regions
    regions = ['us-east-1', 'us-west-2', 'eu-west-1']
    for region in regions:
        sqs = boto3.client('sqs', region_name=region)

        # Create queue
        queue = sqs.create_queue(
            QueueName=f'events-{region}',
            Attributes={
                'VisibilityTimeout': '300',
                'MessageRetentionPeriod': '1209600'
            }
        )

        # Subscribe to SNS topic
        sns_primary.subscribe(
            TopicArn=topic_arn,
            Protocol='sqs',
            Endpoint=queue['QueueArn']
        )

    print(f"Multi-region messaging configured across {len(regions)} regions")

See Also

• AWS Lambda
• AWS EventBridge
• AWS Step Functions
• AWS Kinesis