lección 10

Step Functions + Lambda: pipelines serverless completos

Orquestar Lambdas con Step Functions: estados, transiciones, manejo de errores, paralelismo y el patrón de pipeline serverless completo.

⏱ 55 min

Ya conoces Step Functions de la Skill 15 (Orquestación). Ahora lo combinamos con Lambda para construir pipelines de datos completamente serverless. La idea es poderosa: cada paso del pipeline es una Lambda independiente, y Step Functions las coordina — maneja el orden de ejecución, los reintentos si algo falla, las bifurcaciones condicionales, y el paralelismo.

Este es el patrón más moderno y económico para pipelines de datos que no son extremadamente pesados. Si tu pipeline procesa archivos de < 1GB por paso y no necesita Spark, Step Functions + Lambda es más barato y fácil de mantener que Airflow o EMR.

### El patrón de pipeline serverless

01.S3 Event: llega un archivo nuevo a raw/
02.Lambda Validadora: verifica formato, tamaño, esquema
03.Lambda Transformadora: limpia, normaliza, convierte a Parquet
04.Lambda Cargadora: registra en Glue Catalog o carga en Redshift
05.Lambda Notificadora: envía Slack/email si todo OK o si falla

### Definir una máquina de estados (ASL)

1import boto3
2import json
3
4sfn = boto3.client('stepfunctions', endpoint_url='http://localhost:4566',
5                   aws_access_key_id='test', aws_secret_access_key='test',
6                   region_name='eu-west-1')
7
8# Definición de la máquina de estados (ASL - Amazon States Language)
9pipeline_definition = {
10    "Comment": "Pipeline ETL serverless: validar → transformar → cargar",
11    "StartAt": "ValidarArchivo",
12    "States": {
13        "ValidarArchivo": {
14            "Type": "Task",
15            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:validar-archivo-s3",
16            "Next": "EsValido",
17            "Catch": [{
18                "ErrorEquals": ["States.ALL"],
19                "Next": "NotificarError"
20            }]
21        },
22        "EsValido": {
23            "Type": "Choice",
24            "Choices": [{
25                "Variable": "$.valido",
26                "BooleanEquals": True,
27                "Next": "TransformarDatos"
28            }],
29            "Default": "NotificarError"
30        },
31        "TransformarDatos": {
32            "Type": "Task",
33            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:transformar-csv",
34            "Next": "CargarEnCatalogo",
35            "Retry": [{
36                "ErrorEquals": ["States.TaskFailed"],
37                "MaxAttempts": 2,
38                "IntervalSeconds": 30
39            }]
40        },
41        "CargarEnCatalogo": {
42            "Type": "Task",
43            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:registrar-catalogo",
44            "Next": "NotificarExito"
45        },
46        "NotificarExito": {
47            "Type": "Task",
48            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:notificar",
49            "End": True
50        },
51        "NotificarError": {
52            "Type": "Task",
53            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:notificar-error",
54            "End": True
55        }
56    }
57}
58
59# Crear la máquina de estados
60sfn.create_state_machine(
61    name='pipeline-etl-serverless',
62    definition=json.dumps(pipeline_definition),
63    roleArn='arn:aws:iam::000000000000:role/StepFunctionsRole'
64)
65print("✅ Pipeline serverless creado")

Un pipeline completo definido como máquina de estados con reintentos y manejo de errores

### Ejecutar el pipeline y monitorizar

1import time
2
3# Ejecutar el pipeline con un input (simula evento S3)
4execution = sfn.start_execution(
5    stateMachineArn='arn:aws:states:eu-west-1:000000000000:stateMachine:pipeline-etl-serverless',
6    input=json.dumps({
7        "bucket": "fashionstore-datalake",
8        "key": "raw/ventas/2024-01-15.csv",
9        "size": 15000
10    })
11)
12
13execution_arn = execution['executionArn']
14print(f"Ejecutando: {execution_arn}")
15
16# Poll estado
17while True:
18    desc = sfn.describe_execution(executionArn=execution_arn)
19    status = desc['status']
20    print(f"  Estado: {status}")
21    if status in ('SUCCEEDED', 'FAILED', 'TIMED_OUT', 'ABORTED'):
22        break
23    time.sleep(2)
24
25if status == 'SUCCEEDED':
26    print(f"✅ Pipeline completado. Output: {desc.get('output', '')}")
27else:
28    print(f"❌ Pipeline falló: {desc.get('error', 'Unknown')}")

Ejecutar y monitorizar una ejecución del pipeline

### Paralelismo: procesar varios archivos simultáneamente

Step Functions soporta el estado "Parallel" para ejecutar ramas en paralelo, y "Map" para iterar sobre una lista procesando cada elemento concurrentemente. Esto es ideal cuando llegan 10 archivos a la vez: cada uno se procesa en paralelo por su propia Lambda.

1# Estado Map: procesar cada archivo en paralelo
2map_state = {
3    "Type": "Map",
4    "ItemsPath": "$.archivos",
5    "MaxConcurrency": 5,  # Máximo 5 en paralelo
6    "Iterator": {
7        "StartAt": "ProcesarArchivo",
8        "States": {
9            "ProcesarArchivo": {
10                "Type": "Task",
11                "Resource": "arn:aws:lambda:...:function:procesar-archivo",
12                "End": True
13            }
14        }
15    },
16    "Next": "ConsolidarResultados"
17}
18print("Map state: procesa N archivos en paralelo (máx 5 concurrentes)")

El estado Map itera sobre una lista procesando items en paralelo

Consejo de senior: Step Functions Express Workflows ($0.000001 por transición) son 10x más baratos que Standard Workflows ($0.000025 por transición) para pipelines de alta frecuencia. Usa Express para pipelines que se ejecutan > 100 veces/día y duran < 5 minutos. Standard para pipelines diarios largos.

Cuidado con los límites de concurrencia de Lambda: por defecto 1000 ejecuciones simultáneas por región. Si tu Step Functions Map lanza 500 Lambdas en paralelo y otros servicios también usan Lambda, puedes alcanzar el límite y provocar throttling. Usa MaxConcurrency en Map para controlarlo.

Cada Lambda es independiente y testeable — Step Functions las conecta

## ejercicios

[01]

Crear pipeline ETL serverless

Define y crea una máquina de estados que orqueste un pipeline de 3 pasos: ingestar, transformar y publicar, con reintentos en cada paso.

Cargando editor...

[02]

Ejecutar pipeline y monitorizar progreso

Escribe una función que ejecute el pipeline y haga polling del estado hasta que termine, mostrando cada transición de estado.

Cargando editor...

[03]

Pipeline con procesamiento paralelo (Map)

Modifica el pipeline para que reciba una lista de archivos y los procese en paralelo usando el estado Map.

Cargando editor...

[04]

Comparar coste Step Functions vs Airflow

Calcula el coste mensual de ejecutar un pipeline con Step Functions (Standard y Express) vs mantener un servidor Airflow.

💡 Resultado esperado

sf_standard_usd: 0.38
sf_express_usd: 1.5
airflow_mwaa_usd: 352.8
airflow_self_hosted_usd: 45

Cargando editor...

Regístrate para guardar tu progreso.

## comentarios

Reporta erratas, ayuda a otros o comparte tu opinión. Sé constructivo.

Inicia sesión para comentar y responder.

cargando comentarios...