Construir Lambda Event-Driven con Triggers SQS y Batch

Descripcion general

Los triggers SQS permiten a Lambda consumir mensajes de una cola automaticamente. Lambda hace polling de SQS, agrupa mensajes e invoca la funcion. Con batch windows, partial batch responses y dead-letter queues, puedes construir pipelines event-driven resilientes que manejan fallos graceful sin perder mensajes. A continuacion: configurar triggers SQS, batch processing, manejo de errores con partial batch responses, configuracion de DLQ y monitoreo.

Cuando Usar Esto

• For alternatives, see Serverless Architecture — Patterns and Anti-Patterns.

• Procesamiento asincrono de tareas (thumbnails de imagenes, envio de emails, generacion de reportes)

• Desacoplar productores de consumidores en arquitecturas event-driven

• Manejar picos de trafico con buffering basado en colas

• Cualquier workload donde los mensajes deben sobrevivir fallos de funcion

Prerrequisitos

• Python 3.11+
• Cuenta AWS con acceso a Lambda y SQS
• Paquete boto3

Solucion

1. Crear Cola SQS con DLQ

# Crear dead-letter queue
aws sqs create-queue --queue-name my-dlq

DLQ_URL=$(aws sqs get-queue-url --queue-name my-dlq --query 'QueueUrl' --output text)
DLQ_ARN=$(aws sqs get-queue-attributes --queue-url $DLQ_URL --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

# Crear cola principal con DLQ configurado
aws sqs create-queue --queue-name my-queue --attributes file://queue-attributes.json

# queue-attributes.json
{
  "RedrivePolicy": "{\"deadLetterTargetArn\":\"'${DLQ_ARN}'\",\"maxReceiveCount\":\"3\"}",
  "VisibilityTimeout": "120",
  "MessageRetentionPeriod": "1209600"
}

2. Handler Lambda con Batch Processing

import json
import boto3
import logging

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

sqs = boto3.client('sqs')

def lambda_handler(event, context):
    records = event.get('Records', [])
    logger.info(f"Processing {len(records)} messages")

    batch_item_failures = []

    for record in records:
        message_id = record['messageId']
        receipt_handle = record['receiptHandle']

        try:
            body = json.loads(record['body'])
            process_message(body)
            logger.info(f"Successfully processed message {message_id}")

        except Exception as e:
            logger.error(f"Failed to process message {message_id}: {e}")
            batch_item_failures.append({
                'itemIdentifier': message_id,
            })

    # Retornar partial batch response — solo los mensajes fallidos vuelven a SQS
    return {
        'batchItemFailures': batch_item_failures,
    }

def process_message(body: dict):
    order_id = body.get('orderId')
    if not order_id:
        raise ValueError("Missing orderId")

    # Logica de negocio aqui
    logger.info(f"Processing order {order_id}")
    # ej., guardar en DynamoDB, llamar API externa, generar reporte

3. Template SAM con Trigger SQS

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  ProcessQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: my-queue
      VisibilityTimeout: 120
      MessageRetentionPeriod: 1209600
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt DeadLetterQueue.Arn
        maxReceiveCount: 3

  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: my-dlq

  ProcessFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda_function.lambda_handler
      Runtime: python3.11
      CodeUri: src/
      Timeout: 120
      MemorySize: 512
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt ProcessQueue.Arn
            BatchSize: 10
            MaxBatchingWindow: 30
            FunctionResponseTypes:
              - ReportBatchItemFailures
      Policies:
        - SQSPollerPolicy:
            QueueName: !GetAtt ProcessQueue.QueueName

4. Productor: Enviar Mensajes a SQS

import json
import boto3

sqs = boto3.client('sqs', region_name='us-east-1')
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'

def send_order_message(order: dict):
    response = sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps(order),
        MessageAttributes={
            'OrderType': {
                'DataType': 'String',
                'StringValue': order.get('type', 'standard'),
            },
        },
    )
    return response['MessageId']

def send_batch_orders(orders: list):
    entries = [
        {
            'Id': str(i),
            'MessageBody': json.dumps(order),
            'MessageAttributes': {
                'OrderType': {
                    'DataType': 'String',
                    'StringValue': order.get('type', 'standard'),
                },
            },
        }
        for i, order in enumerate(orders)
    ]

    # Enviar en lotes de 10 (limite SQS)
    for i in range(0, len(entries), 10):
        batch = entries[i:i+10]
        response = sqs.send_message_batch(QueueUrl=QUEUE_URL, Entries=batch)
        if response.get('Failed'):
            for failure in response['Failed']:
                logger.error(f"Failed to send message {failure['Id']}: {failure['Message']}")

5. Cola FIFO con Message Groups

# Enviar a cola FIFO con message group ID para ordenamiento
def send_ordered_message(order: dict, group_id: str):
    response = sqs.send_message(
        QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/my-queue.fifo',
        MessageBody=json.dumps(order),
        MessageGroupId=group_id,  # Mensajes en mismo grupo se procesan en orden
        MessageDeduplicationId=f"order-{order['id']}",  # Idempotencia
    )
    return response['MessageId']

6. Ajuste de Visibility Timeout

import boto3

sqs = boto3.client('sqs')

# Establecer visibility timeout por cola
sqs.set_queue_attributes(
    QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/my-queue',
    Attributes={
        'VisibilityTimeout': '300',  # 5 minutos — debe ser >= timeout de Lambda
    },
)

# El visibility timeout deberia ser al menos 6x el timeout de Lambda
# Si Lambda hace timeout a 60s, establece visibility timeout a 360s
# Esto da a SQS suficiente tiempo para hacer el mensaje visible despues de reintentos

7. Monitoreo y Alarmas

# Alarma de CloudWatch para profundidad de DLQ
Resources:
  DLQAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: DLQ-Messages-Alert
      AlarmDescription: Alert when DLQ has messages
      MetricName: ApproximateNumberOfMessagesVisible
      Namespace: AWS/SQS
      Dimensions:
        - Name: QueueName
          Value: my-dlq
      Statistic: Sum
      Period: 60
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      AlarmActions:
        - !Ref AlertTopic

Como Funciona

1. Polling: Lambda hace polling de SQS buscando mensajes. Cuando hay mensajes disponibles, los agrupa (hasta BatchSize) e invoca la funcion con todos los mensajes en un evento.
2. Batching window: MaxBatchingWindow (0-300 segundos) le dice a Lambda que espere por mas mensajes antes de invocar. Una ventana de 30 segundos recolecta hasta 10 mensajes antes de invocar, reduciendo invocaciones de funcion.
3. Partial batch responses: Con ReportBatchItemFailures, la funcion retorna una lista de IDs de mensajes fallidos. SQS solo re-envia esos mensajes — los exitosos se eliminan. Sin esto, un solo fallo rechaza todo el lote.
4. Visibility timeout: Cuando Lambda lee un mensaje, SQS lo oculta de otros consumidores por la duracion del visibility timeout. Si la funcion falla, el mensaje se vuelve visible despues del timeout. Establecelo a al menos 6x el timeout de Lambda.
5. DLQ: Despues de maxReceiveCount intentos de recepcion fallidos, SQS mueve el mensaje a la dead-letter queue. Inspecciona el DLQ para entender por que los mensajes fallan.

Variantes

Cola FIFO con Lambda

# Las colas FIFO requieren BatchSize=1 para ordenamiento estricto, o usa message groups
ProcessFunction:
  Type: AWS::Serverless::Function
  Properties:
    Events:
      SQSEvent:
        Type: SQS
        Properties:
          Queue: !GetAtt FifoQueue.Arn
          BatchSize: 10
          # Lambda procesa un message group a la vez para FIFO

Drenado Programado de Cola

# Regla de EventBridge dispara Lambda cada 5 minutos para drenar la cola
import boto3

def lambda_handler(event, context):
    sqs = boto3.client('sqs')
    queue_url = event['queueUrl']

    while True:
        response = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=5,
        )

        messages = response.get('Messages', [])
        if not messages:
            break

        for msg in messages:
            process_message(json.loads(msg['Body']))
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=msg['ReceiptHandle'],
            )

Trigger SQS Cross-Account

# El ARN de la cola debe estar en la misma region
# Agrega una politica basada en recursos a la cola permitiendo a Lambda consumir
ProcessFunction:
  Type: AWS::Serverless::Function
  Properties:
    Events:
      SQSEvent:
        Type: SQS
        Properties:
          Queue: arn:aws:sqs:us-east-1:999999999999:cross-account-queue
          BatchSize: 10

Mejores Practicas

• Usar ReportBatchItemFailures: Sin esto, un mensaje fallido rechaza todo el lote, causando reprocesamiento de mensajes exitosos. Siempre retorna partial batch responses.
• Establecer visibility timeout a 6x el timeout de Lambda: Si Lambda hace timeout, el mensaje deberia permanecer oculto el tiempo suficiente para el siguiente intento de retry.
• Usar DLQ con maxReceiveCount de 3-5: Muy bajo envia mensajes a DLQ por errores transitorios. Muy alto desperdicia tiempo de procesamiento en poison pills.
• Mantener batch size pequeno para procesamiento pesado: Batch size 10 es el default. Para procesamiento CPU-intensivo, usa 1-5 para evitar timeouts.
• Usar colas FIFO para ordenamiento: Las colas estandar entregan at-least-once sin ordenamiento. Las colas FIFO garantizan orden dentro de un message group.
• Monitorear profundidad de DLQ: Establece alarmas de CloudWatch en el conteo de mensajes del DLQ. Investiga mensajes del DLQ prontamente.

Errores Comunes

• No usar partial batch responses: Sin ReportBatchItemFailures, un mensaje fallido causa que todo el lote reintente. Los mensajes exitosos se reprocesan.
• Visibility timeout demasiado corto: Si el visibility timeout es mas corto que el timeout de Lambda, SQS hace el mensaje visible mientras Lambda aun lo procesa, causando procesamiento duplicado.
• Sin DLQ: Sin DLQ, los mensajes poison pill (que siempre fallan) se reintentan hasta que MessageRetentionPeriod expira, luego se pierden.
• Batch size demasiado grande: Un lote de 10 mensajes con 30 segundos de procesamiento cada uno excede un timeout de 60 segundos. Ajusta batch size al tiempo de procesamiento.
• No manejar errores de json.loads: Si el body del mensaje no es JSON valido, json.loads lanza una excepcion. Envuelve en try/except y retorna como batch item failure.