Part 4: Event-Driven Architecture with Lambda

Understanding Event-Driven Systems

One of my biggest realizations working with Lambda was that serverless isn't just about replacing servers—it's about rethinking application architecture around events. This mindset shift transformed how I design systems.

What is Event-Driven Architecture?

In event-driven architecture, components communicate through events rather than direct calls. When something happens (an event), interested parties (consumers) react to it.

Traditional vs Event-Driven Flow

AWS Lambda Event Sources

Lambda integrates with numerous AWS services. Here are the ones I use most frequently:

Direct Invocation Sources

1. API Gateway - HTTP/REST APIs

2. Application Load Balancer - HTTP requests

3. Lambda Function URLs - Direct HTTP endpoints

Asynchronous Sources

1. S3 - Object operations

2. SNS - Pub/sub messages

3. EventBridge - Event routing

4. SES - Email events

Stream-Based Sources

1. DynamoDB Streams - Table changes

2. Kinesis - Real-time data streams

3. SQS - Message queues

4. Kafka - Event streaming

Building an Event-Driven Workflow

Let me show you a real system I built: a Document Processing Pipeline that handles uploaded documents through multiple stages.

System Architecture

Event Source 1: S3 Events

S3 is my go-to trigger for file processing workflows.

S3 Event Handler

Create src/handlers/s3_processor.py:

import boto3
import json
import logging
from urllib.parse import unquote_plus
from typing import Dict, List
import os

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3_client = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
sns_client = boto3.client('sns')

table = dynamodb.Table(os.environ['DOCUMENTS_TABLE'])
topic_arn = os.environ['PROCESSING_TOPIC_ARN']


def lambda_handler(event: Dict, context) -> Dict:
    """
    Process S3 object creation events
    
    Event structure from S3:
    {
        "Records": [{
            "eventName": "ObjectCreated:Put",
            "s3": {
                "bucket": {"name": "my-bucket"},
                "object": {"key": "path/file.pdf", "size": 1024}
            }
        }]
    }
    """
    logger.info(f"Processing {len(event['Records'])} records")
    
    processed = []
    
    for record in event['Records']:
        try:
            result = process_s3_record(record)
            processed.append(result)
            
        except Exception as e:
            logger.error(f"Failed to process record: {str(e)}", exc_info=True)
            # Don't fail entire batch for one error
            continue
    
    return {
        'statusCode': 200,
        'processed': len(processed),
        'results': processed
    }


def process_s3_record(record: Dict) -> Dict:
    """Process individual S3 record"""
    
    # Extract S3 details
    bucket = record['s3']['bucket']['name']
    key = unquote_plus(record['s3']['object']['key'])
    size = record['s3']['object']['size']
    
    logger.info(f"Processing s3://{bucket}/{key}")
    
    # Get object metadata
    metadata = s3_client.head_object(Bucket=bucket, Key=key)
    
    # Extract file information
    file_info = {
        'document_id': key.split('/')[-1].split('.')[0],
        'bucket': bucket,
        'key': key,
        'size': size,
        'content_type': metadata.get('ContentType', 'unknown'),
        'uploaded_at': metadata['LastModified'].isoformat(),
        'status': 'processing'
    }
    
    # Store in DynamoDB
    table.put_item(Item=file_info)
    logger.info(f"Stored metadata for {file_info['document_id']}")
    
    # Publish event for downstream processing
    sns_client.publish(
        TopicArn=topic_arn,
        Message=json.dumps(file_info),
        Subject=f"Document Uploaded: {file_info['document_id']}",
        MessageAttributes={
            'event_type': {
                'DataType': 'String',
                'StringValue': 'document_uploaded'
            },
            'document_type': {
                'DataType': 'String',
                'StringValue': file_info['content_type']
            }
        }
    )
    
    return file_info

SAM Template for S3 Trigger

DocumentProcessorFunction:
  Type: AWS::Serverless::Function
  Properties:
    CodeUri: src/
    Handler: handlers.s3_processor.lambda_handler
    Runtime: python3.12
    Environment:
      Variables:
        DOCUMENTS_TABLE: !Ref DocumentsTable
        PROCESSING_TOPIC_ARN: !Ref ProcessingTopic
    Policies:
      - S3ReadPolicy:
          BucketName: !Ref DocumentBucket
      - DynamoDBCrudPolicy:
          TableName: !Ref DocumentsTable
      - SNSPublishMessagePolicy:
          TopicName: !GetAtt ProcessingTopic.TopicName
    Events:
      S3Upload:
        Type: S3
        Properties:
          Bucket: !Ref DocumentBucket
          Events: s3:ObjectCreated:*
          Filter:
            S3Key:
              Rules:
                - Name: prefix
                  Value: uploads/
                - Name: suffix
                  Value: .pdf

Event Source 2: DynamoDB Streams

DynamoDB Streams let you react to table changes. I use this for audit trails and data replication.

DynamoDB Stream Handler

Create src/handlers/stream_processor.py:

import boto3
import json
import logging
from typing import Dict, List
from decimal import Decimal

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Custom JSON encoder for DynamoDB types
class DecimalEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return super(DecimalEncoder, self).default(obj)


def lambda_handler(event: Dict, context) -> Dict:
    """
    Process DynamoDB stream events
    
    Event structure:
    {
        "Records": [{
            "eventName": "INSERT|MODIFY|REMOVE",
            "dynamodb": {
                "Keys": {...},
                "NewImage": {...},
                "OldImage": {...}
            }
        }]
    }
    """
    logger.info(f"Processing {len(event['Records'])} stream records")
    
    for record in event['Records']:
        event_name = record['eventName']
        
        if event_name == 'INSERT':
            handle_insert(record)
        elif event_name == 'MODIFY':
            handle_modify(record)
        elif event_name == 'REMOVE':
            handle_remove(record)
    
    return {'statusCode': 200}


def handle_insert(record: Dict):
    """Handle new item insertion"""
    new_item = record['dynamodb']['NewImage']
    logger.info(f"New item created: {json.dumps(new_item, cls=DecimalEncoder)}")
    
    # Example: Send welcome email for new users
    if 'email' in new_item:
        send_welcome_email(new_item)


def handle_modify(record: Dict):
    """Handle item modification"""
    old_item = record['dynamodb'].get('OldImage', {})
    new_item = record['dynamodb']['NewImage']
    
    # Compare old vs new
    logger.info("Item modified")
    
    # Example: Notify if status changed
    old_status = old_item.get('status', {}).get('S')
    new_status = new_item.get('status', {}).get('S')
    
    if old_status != new_status:
        notify_status_change(new_item, old_status, new_status)


def handle_remove(record: Dict):
    """Handle item deletion"""
    old_item = record['dynamodb']['OldImage']
    logger.info(f"Item deleted: {json.dumps(old_item, cls=DecimalEncoder)}")


def send_welcome_email(item: Dict):
    """Send welcome email to new user"""
    logger.info(f"Sending welcome email to {item.get('email', {}).get('S')}")
    # Implementation here


def notify_status_change(item: Dict, old_status: str, new_status: str):
    """Notify about status change"""
    logger.info(f"Status changed from {old_status} to {new_status}")
    # Implementation here

Stream Processing Flow

Event Source 3: SQS Queues

SQS provides reliable asynchronous processing. I use it for decoupling services and handling high-throughput workloads.

SQS Message Handler

Create src/handlers/sqs_processor.py:

import boto3
import json
import logging
from typing import Dict, List

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event: Dict, context) -> Dict:
    """
    Process SQS messages
    
    Event structure:
    {
        "Records": [{
            "messageId": "abc-123",
            "body": "{\"key\": \"value\"}",
            "attributes": {...},
            "messageAttributes": {...}
        }]
    }
    """
    logger.info(f"Processing {len(event['Records'])} messages")
    
    # Track failures for partial batch response
    batch_item_failures = []
    
    for record in event['Records']:
        try:
            process_message(record)
            
        except Exception as e:
            logger.error(f"Failed to process message: {str(e)}", exc_info=True)
            
            # Add to failures for retry
            batch_item_failures.append({
                'itemIdentifier': record['messageId']
            })
    
    # Return failed messages for retry
    return {
        'batchItemFailures': batch_item_failures
    }


def process_message(record: Dict):
    """Process individual SQS message"""
    
    message_id = record['messageId']
    body = json.loads(record['body'])
    
    logger.info(f"Processing message {message_id}")
    
    # Extract message attributes
    attributes = record.get('messageAttributes', {})
    message_type = attributes.get('type', {}).get('stringValue', 'unknown')
    
    # Route based on message type
    if message_type == 'order':
        process_order(body)
    elif message_type == 'notification':
        process_notification(body)
    else:
        logger.warning(f"Unknown message type: {message_type}")


def process_order(order: Dict):
    """Process order message"""
    logger.info(f"Processing order {order.get('order_id')}")
    # Order processing logic


def process_notification(notification: Dict):
    """Process notification message"""
    logger.info(f"Sending notification to {notification.get('recipient')}")
    # Notification logic

SQS Integration Pattern

SAM Template for SQS

SQSProcessorFunction:
  Type: AWS::Serverless::Function
  Properties:
    CodeUri: src/
    Handler: handlers.sqs_processor.lambda_handler
    Runtime: python3.12
    ReservedConcurrentExecutions: 10  # Limit concurrency
    Events:
      SQSEvent:
        Type: SQS
        Properties:
          Queue: !GetAtt ProcessingQueue.Arn
          BatchSize: 10
          MaximumBatchingWindowInSeconds: 5
          FunctionResponseTypes:
            - ReportBatchItemFailures  # Enable partial batch responses

ProcessingQueue:
  Type: AWS::SQS::Queue
  Properties:
    QueueName: ProcessingQueue
    VisibilityTimeout: 300  # 5 minutes (6x function timeout)
    MessageRetentionPeriod: 1209600  # 14 days
    RedrivePolicy:
      deadLetterTargetArn: !GetAtt DeadLetterQueue.Arn
      maxReceiveCount: 3

DeadLetterQueue:
  Type: AWS::SQS::Queue
  Properties:
    QueueName: ProcessingQueue-DLQ
    MessageRetentionPeriod: 1209600  # 14 days

Event Source 4: API Gateway

We covered basic API Gateway in Part 3. Here's advanced patterns I use:

WebSocket API Handler

Create src/handlers/websocket_handler.py:

import boto3
import json
import logging
import os

logger = logging.getLogger()
logger.setLevel(logging.INFO)

dynamodb = boto3.resource('dynamodb')
connections_table = dynamodb.Table(os.environ['CONNECTIONS_TABLE'])
apigw_client = boto3.client('apigatewaymanagementapi',
                             endpoint_url=os.environ.get('WEBSOCKET_ENDPOINT'))


def lambda_handler(event: Dict, context) -> Dict:
    """
    Handle WebSocket events
    
    Route keys: $connect, $disconnect, sendMessage
    """
    route_key = event['requestContext']['routeKey']
    connection_id = event['requestContext']['connectionId']
    
    logger.info(f"Route: {route_key}, Connection: {connection_id}")
    
    if route_key == '$connect':
        return handle_connect(connection_id, event)
    elif route_key == '$disconnect':
        return handle_disconnect(connection_id)
    elif route_key == 'sendMessage':
        return handle_send_message(connection_id, event)
    
    return {'statusCode': 200}


def handle_connect(connection_id: str, event: Dict) -> Dict:
    """Handle new WebSocket connection"""
    
    # Extract user info from query params
    query_params = event.get('queryStringParameters', {})
    user_id = query_params.get('userId', 'anonymous')
    
    # Store connection
    connections_table.put_item(
        Item={
            'connectionId': connection_id,
            'userId': user_id,
            'connectedAt': event['requestContext']['requestTimeEpoch']
        }
    )
    
    logger.info(f"User {user_id} connected: {connection_id}")
    return {'statusCode': 200}


def handle_disconnect(connection_id: str) -> Dict:
    """Handle WebSocket disconnection"""
    
    connections_table.delete_item(
        Key={'connectionId': connection_id}
    )
    
    logger.info(f"Disconnected: {connection_id}")
    return {'statusCode': 200}


def handle_send_message(connection_id: str, event: Dict) -> Dict:
    """Handle sending message to all connections"""
    
    body = json.loads(event.get('body', '{}'))
    message = body.get('message', '')
    
    # Get all connections
    response = connections_table.scan()
    connections = response['Items']
    
    # Broadcast to all
    for connection in connections:
        try:
            apigw_client.post_to_connection(
                ConnectionId=connection['connectionId'],
                Data=json.dumps({
                    'from': connection_id,
                    'message': message
                })
            )
        except apigw_client.exceptions.GoneException:
            # Connection is dead, remove it
            connections_table.delete_item(
                Key={'connectionId': connection['connectionId']}
            )
    
    return {'statusCode': 200}

WebSocket Flow

Event Filtering and Routing

I use EventBridge for complex event routing patterns.

EventBridge Pattern Matching

# SAM Template
EventProcessorFunction:
  Type: AWS::Serverless::Function
  Properties:
    Handler: handlers.event_processor.lambda_handler
    Events:
      OrderCreated:
        Type: EventBridgeRule
        Properties:
          EventBusName: !Ref EventBus
          Pattern:
            source:
              - com.myapp.orders
            detail-type:
              - OrderCreated
            detail:
              amount:
                - numeric: [">=", 1000]  # Only orders >= $1000
      
      HighPriorityUser:
        Type: EventBridgeRule
        Properties:
          EventBusName: !Ref EventBus
          Pattern:
            source:
              - com.myapp.users
            detail:
              tier:
                - premium
                - enterprise

Complete Event-Driven System

Here's how all these pieces work together in a real system:

Key Takeaways

From building event-driven systems:

1. Decouple Services: Events enable independent scaling and deployment

2. Embrace Async: Not everything needs immediate response

3. Handle Failures: Always implement retries and dead-letter queues

4. Monitor Events: Track event flow through CloudWatch and X-Ray

5. Filter Early: Use event filtering to reduce Lambda invocations

6. Batch When Possible: Process multiple events together for efficiency

What's Next?

In Part 5: Deployment and Best Practices, we'll cover:

• CI/CD pipelines for Lambda

• Infrastructure as Code best practices

• Monitoring and observability

• Cost optimization strategies

• Security hardening

• Performance tuning

Get ready to take your Lambda functions to production!

---

This article is based on event-driven systems I've built handling millions of events daily.