consulting

Tubería de datos sin servidor con AWS Lambda y Python

Allan Porras

07 Apr 2026 — 10 min read

La arquitectura sin servidor ofrece un cambio de paradigma con respecto a la infraestructura tradicional, centrada en servidores, permitiendo a los ingenieros construir sistemas altamente escalables, resilientes y rentables sin tener que gestionar los recursos informáticos subyacentes. Un caso de uso típico para este modelo es la construcción de tuberías de datos.

Este artículo ofrece una guía técnica completa para implementar una tubería de datos impulsada por eventos y sin servidor en AWS utilizando S3, SQS, Lambda y DynamoDB. Nos centraremos en los principios arquitectónicos, las consideraciones de rendimiento y proporcionaremos código práctico que puede implementar de inmediato.

Servicios de Ingeniería de Productos

Colabore con nuestros gestores de proyectos, ingenieros de software y probadores de calidad, para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Arquitectura y componentes principales

El diseño de la plataforma prioriza la desvinculación, la escalabilidad y la resiliencia. Los datos fluyen a través de una serie de servicios especializados y gestionados, cada uno desempeñando un papel fundamental.

Amazon S3 (Simple Storage Service): Sirve como el punto de entrada de la tubería. Los archivos de datos sin procesar (por ejemplo, CSV, JSON) se suben a un bucket de S3 designado. Configuraremos las Notificaciones de Eventos de S3 para activar la tubería al crear un objeto.
Amazon SQS (Simple Queue Service): Actúa como un búfer duradero y asíncrono entre S3 y nuestra capa de cómputo. Colocar una cola aquí es una decisión arquitectónica fundamental. Separa la ingestión de datos del procesamiento de datos, suavizando las cargas de trabajo intermitentes o impredecibles y asegurando que no se pierda ningún dato si la capa de procesamiento experimenta fallos o limitaciones. También facilita los mecanismos de reintento y el manejo de errores a través de una Cola de Destino (DLQ).
AWS Lambda: Este es el núcleo de cómputo sin servidor de nuestra tubería. La función de Lambda se activa al recibir mensajes en la cola de SQS. Su única responsabilidad es obtener el archivo sin procesar de S3, realizar las transformaciones necesarias y cargar los datos estructurados en nuestro almacén de datos de destino.
Amazon DynamoDB: Una base de datos NoSQL totalmente administrada que sirve como nuestro almacén de datos de destino. Su flexibilidad "schema-on-read", su bajo rendimiento y su escalado sin problemas la convierten en un objetivo ideal para los datos estructurados extraídos de la tubería.

Requisitos y Configuración del Entorno

Antes de la implementación, asegúrese de que su entorno de desarrollo esté configurado con lo siguiente:

Cuenta de AWS: Una cuenta de AWS activa con acceso programático.
AWS CLI: La interfaz de línea de comandos de AWS, configurada con tus credenciales.
Python 3.9+ y Boto3: El SDK de AWS para Python.

Rol de Ejecución de IAM

La función Lambda requiere un rol de IAM con permisos para interactuar con otros servicios de AWS. El rol de ejecución debe tener una política de confianza que permita a lambda.amazonaws.com asumir este rol, y debe estar adjunto a una política que otorgue los permisos necesarios.

Política IAM:LambdaDataPipelinePolicy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": "arn:aws:s3:::your-data-ingestion-bucket/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "sqs:ReceiveMessage",
                "sqs:DeleteMessage",
                "sqs:GetQueueAttributes"
            ],
            "Resource": "arn:aws:sqs:us-east-1:123456789012:data-processing-queue"
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:BatchWriteItem",
                "dynamodb:PutItem"
            ],
            "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/ProcessedDataTable"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        }
    ]
}

Reemplace los ARNs con los recursos específicos que va a crear.

Implementación paso a paso

Utilizaremos la AWS CLI para la configuración de la infraestructura, con el fin de mantener la claridad y la reproducibilidad.

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y probadores de calidad, para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Paso 1: Crear la infraestructura (S3, SQS, DynamoDB)

Primero, configure los componentes principales.

Crear la tabla DynamoDB: Esta tabla de ejemplo está diseñada para almacenar datos de perfil de usuario.

aws dynamodb create-table \
    --table-name ProcessedDataTable \
    --attribute-definitions \
        AttributeName=userId,AttributeType=S \
        AttributeName=timestamp,AttributeType=N \
    --key-schema \
        AttributeName=userId,KeyType=HASH \
        AttributeName=timestamp,KeyType=RANGE \
    --billing-mode PAY_PER_REQUEST

Cree las colas SQS (principal y cola de mensajes rechazados): Una cola de mensajes rechazados (DLQ) es esencial para capturar y aislar los mensajes que fallan repetidamente en el procesamiento.

# Create the DLQ first
aws sqs create-queue --queue-name data-processing-dlq

# Get the DLQ ARN
DLQ_ARN=$(aws sqs get-queue-attributes --queue-name data-processing-dlq --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

# Create the main queue with a redrive policy pointing to the DLQ
aws sqs create-queue \
    --queue-name data-processing-queue \
    --attributes '{
        "RedrivePolicy": "{\"deadLetterTargetArn\":\"'"$DLQ_ARN"'\",\"maxReceiveCount\":\"3\"}",
        "VisibilityTimeout": "300"
    }'

Nota: El VisibilityTimeout (300 segundos) debe establecerse en al menos 6 veces el tiempo de espera esperado de la función Lambda para evitar el procesamiento duplicado de mensajes durante los fallos.

Cree el contenedor S3:

aws s3api create-bucket \
    --bucket your-data-ingestion-bucket \
    --region us-east-1

Paso 2: Configure las notificaciones de eventos de S3 a SQS

A continuación, vinculamos el bucket de S3 con la cola de SQS. Esto requiere otorgar a S3 permiso para enviar mensajes a SQS.

Cree la configuración de notificación S3: Esta configuración indica a S3 que envíe un mensaje a SQS para cada nuevo archivo .csv creado.

aws s3api put-bucket-notification-configuration \
    --bucket your-data-ingestion-bucket \
    --notification-configuration '{
        "QueueConfigurations": [{
            "Id": "s3-to-sqs-event",
            "QueueArn": "'"$QUEUE_ARN"'",
            "Events": ["s3:ObjectCreated:*"],
            "Filter": {
                "Key": {
                    "FilterRules": [{
                        "Name": "suffix",
                        "Value": ".csv"
                    }]
                }
            }
        }]
    }'

Agrega la política de la cola SQS: Crea un archivo sqs-policy.json para permitir que el servicio S3 realice la función SendMessage.

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {
      "Service": "s3.amazonaws.com"
    },
    "Action": "sqs:SendMessage",
    "Resource": "YOUR_QUEUE_ARN",
    "Condition": {
      "ArnLike": { "aws:SourceArn": "arn:aws:s3:::your-data-ingestion-bucket" }
    }
  }]
}

Reemplace YOUR_QUEUE_ARN y el nombre del bucket, luego aplique la política.

aws sqs set-queue-attributes \
    --queue-url $QUEUE_URL \
    --attributes Policy=`cat sqs-policy.json`

Obtenga la URL y el ARN de la cola:

QUEUE_URL=$(aws sqs get-queue-url --queue-name data-processing-queue --query 'QueueUrl' --output text)
QUEUE_ARN=$(aws sqs get-queue-attributes --queue-name data-processing-queue --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

Paso 3: Desarrollar la función Lambda de Python

Esta función analizará archivos CSV que contengan datos de usuario y los cargará en DynamoDB.

lambda_function.py

import json
import boto3
import os
import csv
import io
from decimal import Decimal

# Initialize AWS clients outside the handler for reuse
s3_client = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE_NAME'])

def process_csv_file(bucket, key):
    """
    Downloads a CSV file from S3, processes it, and loads data into DynamoDB.
    """
    print(f"Processing file: s3://{bucket}/{key}")
    response = s3_client.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read().decode('utf-8')
    
    # Use io.StringIO to treat the string content as a file
    csv_file = io.StringIO(content)
    reader = csv.DictReader(csv_file)
    
    items_to_write = []
    for row in reader:
        try:
            # Basic transformation: convert data types and structure for DynamoDB
            item = {
                'userId': row['user_id'],
                'timestamp': int(row['event_timestamp']),
                'email': row['email_address'],
                'event_type': row['event_type'],
                # Use Decimal for numeric types to avoid floating point issues
                'transaction_amount': Decimal(row['transaction_amount'])
            }
            items_to_write.append(item)
        except (KeyError, ValueError) as e:
            print(f"Skipping row due to malformed data: {row}. Error: {e}")
            continue

    # Use DynamoDB's BatchWriter for efficient bulk writes
    if items_to_write:
        with table.batch_writer() as batch:
            for item in items_to_write:
                batch.put_item(Item=item)
        print(f"Successfully wrote {len(items_to_write)} items to DynamoDB.")

def lambda_handler(event, context):
    """
    Lambda handler triggered by SQS.
    Parses SQS messages to get S3 object details and triggers processing.
    """
    print(f"Received event: {json.dumps(event)}")
    
    for record in event['Records']:
        sqs_body = json.loads(record['body'])
        
        # S3 event notification message can contain multiple records
        if 'Records' in sqs_body:
            for s3_record in sqs_body['Records']:
                bucket_name = s3_record['s3']['bucket']['name']
                object_key = s3_record['s3']['object']['key']
                
                # Defend against test events from the S3 console
                if not object_key.endswith('.csv'):
                    print(f"Skipping non-CSV file: {object_key}")
                    continue

                try:
                    process_csv_file(bucket_name, object_key)
                except Exception as e:
                    print(f"Error processing file {object_key} from bucket {bucket_name}. Error: {e}")
                    # Re-raise the exception to signal failure to SQS.
                    # This will cause the message to be retried and eventually sent to the DLQ.
                    raise e
    
    return {
        'statusCode': 200,
        'body': json.dumps('Processing complete.')
    }

Paso 4: Empaquete y despliegue la función Lambda

Desplegar la función: Utilice el comando "create-function", haciendo referencia al ARN del rol de ejecución y estableciendo el nombre de la tabla DynamoDB como una variable de entorno.

aws lambda create-function \
    --function-name DataProcessingFunction \
    --runtime python3.9 \
    --role arn:aws:iam::123456789012:role/YourLambdaExecutionRole \
    --handler lambda_function.lambda_handler \
    --zip-file fileb://deployment_package.zip \
    --timeout 60 \
    --memory-size 256 \
    --environment "Variables={DYNAMODB_TABLE_NAME=ProcessedDataTable}"

Empaqueta el Código: Crea un archivo ZIP que contenga el script de Python.

zip deployment_package.zip lambda_function.py

Paso 5: Configurar el disparador de Lambda

Finalmente, conecte la cola de SQS a la función de Lambda utilizando un mapeo de fuente de eventos.

aws lambda create-event-source-mapping \
    --function-name DataProcessingFunction \
    --event-source-arn $QUEUE_ARN \
    --batch-size 10 \
    --maximum-batching-window-in-seconds 5

--batch-size 10: Este es un parámetro crucial de rendimiento. Lambda consultará SQS e invocará tu función con hasta 10 mensajes en un único payload de evento. El procesamiento por lotes reduce significativamente la sobrecarga y el coste de la invocación.
--maximum-batching-window-in-seconds 5: Esto le indica a Lambda que espere hasta 5 segundos para acumular un lote completo antes de invocar la función, lo cual es útil en escenarios de bajo tráfico.

Ajuste y optimización del rendimiento

Idempotencia: El pipeline debe ser idempotente. Si un mensaje se procesa más de una vez (por ejemplo, debido a un reintento después de un tiempo de espera de la función), no debe resultar en datos duplicados. Mientras que la lógica actual de nuestro put_item simplemente sobrescribiría los datos con la misma clave primaria, una transformación más compleja podría requerir comprobaciones explícitas.
Concurrencia de Lambda: Para pipelines de alto rendimiento, configure Concurrencia reservada en la función Lambda. Esto garantiza un cierto número de ejecuciones concurrentes para su función y, lo que es más importante, evita que sobrecargue servicios posteriores como una base de datos con capacidad asignada o una API externa.
Manejo de errores: Nuestra función vuelve a lanzar excepciones en caso de fallo. Esto indica a la fuente de eventos SQS que el lote de mensajes no pudo procesarse. Todo el lote volverá a estar visible en la cola después de que expire el VisibilityTimeout, y se intentará de nuevo. Después de maxReceiveCount (3) fallos, SQS moverá automáticamente el mensaje a la cola de colas de entrega (DLQ) configurada para su inspección manual, evitando que un "mensaje tóxico" bloquee todo el pipeline.
Optimización de costes: Ajuste la memoria de la función Lambda (--memory-size). La memoria está directamente relacionada con la potencia de la CPU. Profile su función para encontrar el punto óptimo donde la reducción del tiempo de ejecución debido a más CPU ya no justifique el mayor coste. El procesamiento por lotes de mensajes de SQS es la técnica más eficaz para ahorrar costes en esta arquitectura.

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y probadores de calidad internos para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Conclusión

Ahora ha implementado una robusta plataforma de procesamiento de datos sin servidor, lista para producción. Esta arquitectura aprovecha servicios gestionados para ofrecer una escalabilidad y fiabilidad excepcionales con un mínimo esfuerzo operativo. La desvinculación proporcionada por SQS es fundamental, permitiendo que cada componente se escale e inicie de forma independiente.

Las mejoras adicionales para un entorno de producción incluirían:

Infraestructura como código (IaC): Automatice la implementación completa utilizando AWS CloudFormation o Terraform para garantizar la consistencia y el control de versiones.
Monitoreo y alertas: Configure las alarmas de CloudWatch en la cola de SQS y en la cola de mensajes descartados (DLQ) para recibir notificaciones sobre retrasos o fallos en el procesamiento.ApproximateAgeOfOldestMessageApproximateNumberOfMessagesVisible para obtener notificaciones sobre retrasos o fallos en el procesamiento.Rastreo distribuido:
Implemente AWS X-Ray para rastrear las solicitudes a medida que fluyen a través de S3, SQS, Lambda y DynamoDB, proporcionando información detallada sobre cuellos de botella y errores de rendimiento.Implemente AWS X-Ray para rastrear las solicitudes a medida que fluyen a través de S3, SQS, Lambda y DynamoDB, proporcionando información detallada sobre los cuellos de botella y los errores de rendimiento.

Preguntas Frecuentes

¿Cuál es el principal beneficio de adoptar una arquitectura sin servidor para las tuberías de datos?

La arquitectura sin servidor ofrece una ventaja significativa al eliminar la necesidad de gestionar servidores, lo que permite a los ingenieros centrarse únicamente en el código y el procesamiento de datos. Esto resulta en sistemas altamente escalables, resilientes y rentables. Al utilizar servicios como AWS Lambda y SQS, se logra una desvinculación entre los componentes, asegurando que la ingesta de datos no se vea comprometida incluso durante picos de tráfico. 4Geeks demuestra cómo esta metodología simplifica la gestión de la infraestructura subyacente, permitiendo una implementación más rápida y eficiente.

¿Qué función cumplen SQS y Lambda en el flujo de la tubería de datos descrita?

Amazon SQS actúa como un búfer duradero y asíncrono, desacoplando la ingestión de datos de la capa de procesamiento. Esto es crucial porque suaviza las cargas de trabajo intermitentes y previene la pérdida de datos si la función de cómputo falla. AWS Lambda es el núcleo de cómputo, activándose para leer los mensajes de SQS, realizar las transformaciones necesarias sobre los archivos de S3 y cargar los resultados en DynamoDB. 4Geeks enfatiza que esta combinación garantiza una arquitectura robusta y altamente escalable para cualquier flujo de datos.

¿Qué consideraciones de diseño son esenciales al implementar esta tubería de datos sin servidor?

Las consideraciones clave se centran en la desvinculación, la escalabilidad y la resiliencia. Es fundamental configurar mecanismos de manejo de errores, como la Cola de Destino (DLQ) en SQS, para aislar los mensajes que fallan repetidamente. Además, es vital configurar correctamente los permisos de IAM para la función Lambda y asegurar que el tiempo de espera de visibilidad de SQS se ajuste adecuadamente. 4Geeks recomienda estas prácticas para garantizar que la implementación de la tubería de datos sea no solo funcional, sino también segura y resistente a fallos.