Máquinas de estado: Step Functions desde cero

### ¿Qué es una máquina de estado?

Antes de meternos en la sintaxis de AWS, necesitas el concepto mental. Una máquina de estado es simplemente un diagrama de flujo ejecutable. Piensa en un semáforo: tiene tres estados (verde, ámbar, rojo) y transiciones claras entre ellos (verde → ámbar → rojo → verde). En cada momento está en exactamente UN estado, y las reglas de transición determinan a qué estado pasa después. (Hay una excepción: Parallel y Map abren varias ramas a la vez — pero incluso ahí, cada rama está en un solo estado.)

Tu pipeline es exactamente eso: un conjunto de estados (tareas) con reglas de transición. "Si ingesta termina bien → ir a transformación. Si falla → ir a manejo de error". En lugar de dibujarlo en una pizarra y luego programar la lógica en Python con try/except anidados, lo defines en un JSON declarativo y Step Functions se encarga de ejecutarlo.

La ventaja del enfoque declarativo: separas el QUÉ (la lógica del flujo) del CÓMO (el código de cada tarea). El flujo se define en JSON y es visual. El código de cada tarea vive en funciones Lambda separadas. Si necesitas cambiar el orden o añadir un paso, modificas el JSON sin tocar el código.

### Amazon States Language (ASL)

Step Functions usa un lenguaje llamado Amazon States Language (ASL) para definir máquinas de estado. Es un JSON con una estructura muy específica. No te asustes — es más simple de lo que parece. Un documento ASL tiene esta estructura:

1{
2  "Comment": "Descripción de lo que hace esta máquina",
3  "StartAt": "NombreDelPrimerEstado",
4  "States": {
5    "NombreDelPrimerEstado": {
6      "Type": "...",
7      "Next": "NombreDelSegundoEstado"
8    },
9    "NombreDelSegundoEstado": {
10      "Type": "...",
11      "End": true
12    }
13  }
14}

• ▹Comment: descripción opcional para humanos
• ▹StartAt: nombre del estado por el que empieza la ejecución
• ▹States: diccionario con todos los estados de la máquina
• ▹Cada estado tiene un Type y o bien Next (siguiente estado) o End: true (fin)

### Los 8 tipos de estado

ASL define 8 tipos de estado. No necesitas memorizarlos todos ahora — los iremos usando uno a uno. Pero aquí va el panorama completo para que entiendas la potencia del sistema:

1. 01.Task: ejecuta algo (una Lambda, un servicio AWS, una API). Es el tipo más importante — donde vive tu código real.
2. 02.Pass: no hace nada útil excepto pasar datos al siguiente estado. Perfecto para prototipar flujos antes de escribir código.
3. 03.Wait: espera un tiempo (10 segundos, 5 minutos, hasta una fecha concreta). Útil para rate limiting o esperar a que un proceso externo termine.
4. 04.Choice: bifurcación condicional. "Si el campo status es error, ve a ManejodeError. Si es ok, ve a Siguiente". Es el if/else de las máquinas de estado.
5. 05.Parallel: ejecuta varias ramas en paralelo y espera a que todas terminen. Como un Promise.all() pero para flujos de datos.
6. 06.Map: para cada elemento de un array, ejecuta un sub-flujo. Es un bucle for paralelo.
7. 07.Succeed: termina la ejecución con éxito. No lleva Next ni End — él mismo es el final feliz.
8. 08.Fail: termina la ejecución con error. Lleva Error y Cause para decir qué falló y por qué.

### Tu primera máquina de estado real: Task + Pass

Vamos a crear algo más interesante que un Hello World. Imagina que tu pipeline tiene que: 1) Verificar si hay datos nuevos, 2) Si los hay, procesarlos. Empezaremos con un estado Pass simulando la verificación y un Task real invocando una Lambda para el procesamiento.

Pero antes de meter Lambdas reales, vamos a usar el patrón que todo ingeniero senior conoce: prototipar con Pass. Defines el flujo completo usando estados Pass que simulan los datos que devolvería cada paso real. Una vez que el flujo funciona, reemplazas los Pass por Tasks. Es como hacer un wireframe antes de diseñar la UI.

1{
2  "Comment": "Pipeline de procesamiento: verificar datos → procesar → notificar",
3  "StartAt": "VerificarDatosNuevos",
4  "States": {
5    "VerificarDatosNuevos": {
6      "Type": "Pass",
7      "Result": {
8        "hay_datos": true,
9        "num_archivos": 3,
10        "tamano_total_mb": 150
11      },
12      "Next": "ProcesarDatos"
13    },
14    "ProcesarDatos": {
15      "Type": "Pass",
16      "Result": {
17        "registros_procesados": 45000,
18        "errores": 0
19      },
20      "Next": "NotificarExito"
21    },
22    "NotificarExito": {
23      "Type": "Pass",
24      "Result": {
25        "notificacion": "Pipeline completado: 45000 registros procesados"
26      },
27      "End": true
28    }
29  }
30}

Ejecuta este prototipo y mira el resultado final. Sale esto: {"notificacion": "Pipeline completado: 45000 registros procesados"}. Y nada mas. Ni el input, ni los 3 archivos de VerificarDatosNuevos, ni los 45.000 registros de ProcesarDatos: cada Pass con Result REEMPLAZA todo lo que recibia. El flujo es correcto — los tres estados se ejecutan en orden — pero los datos no sobreviven al viaje. Arreglar eso es lo que vamos a ver ahora con ResultPath.

### Pasar datos entre estados: Input/Output Processing

Cada estado recibe un input JSON y produce un output JSON. Por defecto, el output de un estado se convierte en el input del siguiente. Pero puedes controlar esto con campos especiales:

• ▹InputPath: filtra qué parte del input recibe el estado (default: $ = todo)
• ▹ResultPath: dónde colocar el resultado dentro del input original (default: $ = reemplaza todo)
• ▹OutputPath: filtra qué parte del resultado pasa al siguiente estado (default: $ = todo)

1{
2  "Comment": "Ejemplo de ResultPath: añade resultado sin perder el input original",
3  "StartAt": "ObtenerConfig",
4  "States": {
5    "ObtenerConfig": {
6      "Type": "Pass",
7      "Result": {"bucket": "datos-raw", "prefix": "ventas/"},
8      "ResultPath": "$.config",
9      "Next": "Procesar"
10    },
11    "Procesar": {
12      "Type": "Pass",
13      "Comment": "Este estado recibe: {config: {bucket, prefix}, ...input_original}",
14      "End": true
15    }
16  }
17}

Y el error opuesto: acumular todo sin recortar. El JSON que viaja entre estados tiene un tope duro de 256 KiB (verificado en las cuotas de AWS, no se puede subir). Si tu paso Extraer devuelve 10.000 registros enteros en vez de un contador, con una fila realista de 76 bytes son 752 KiB — el triple del tope. La regla: por los estados viajan referencias y contadores (rutas de S3, ids, numero de filas), nunca los datos en si.

Y si el JSON acumulado estorba para el paso siguiente, OutputPath lo recorta antes de pasar. "OutputPath": "$.carga" deja pasar solo esa rama al siguiente estado. Se usa al final de un tramo, cuando el contexto acumulado ya no le sirve a nadie.

### Ejecutar y observar en LocalStack

Vamos a crear nuestra máquina de tres pasos y ejecutarla. El flujo completo es: crear → ejecutar → describir (ver resultado).

1# 1. Guardar la definición (usa el JSON del pipeline de 3 pasos)
2cat > pipeline-tres-pasos.json << 'EOF'
3{
4  "Comment": "Pipeline: verificar → procesar → notificar",
5  "StartAt": "VerificarDatosNuevos",
6  "States": {
7    "VerificarDatosNuevos": {
8      "Type": "Pass",
9      "Result": {"hay_datos": true, "num_archivos": 3},
10      "ResultPath": "$.verificacion",
11      "Next": "ProcesarDatos"
12    },
13    "ProcesarDatos": {
14      "Type": "Pass",
15      "Result": {"registros_procesados": 45000, "errores": 0},
16      "ResultPath": "$.procesamiento",
17      "Next": "NotificarExito"
18    },
19    "NotificarExito": {
20      "Type": "Pass",
21      "Result": {"enviado": true},
22      "ResultPath": "$.notificacion",
23      "End": true
24    }
25  }
26}
27EOF
28
29# 2. Crear la máquina
30aws stepfunctions create-state-machine \
31  --name "PipelineTresPasos" \
32  --definition file://pipeline-tres-pasos.json \
33  --role-arn "arn:aws:iam::000000000000:role/DummyRole" \
34  --endpoint-url http://localhost:4566
35
36# 3. Ejecutar con input
37aws stepfunctions start-execution \
38  --state-machine-arn "arn:aws:states:eu-west-1:000000000000:stateMachine:PipelineTresPasos" \
39  --input '{"pipeline": "ventas", "fecha": "2024-01-15"}' \
40  --endpoint-url http://localhost:4566

El output final de esta ejecución contendrá todo: el input original más los resultados de cada paso. Así cada estado tiene contexto completo de lo que pasó antes.

1# 4. Ver el resultado (usa el executionArn que devolvio el paso 3)
2EXEC_ARN="<pega aqui el executionArn del paso anterior>"
3
4aws stepfunctions describe-execution \
5  --execution-arn "$EXEC_ARN" \
6  --endpoint-url http://localhost:4566 \
7  --query 'output' --output text
8
9# Resultado esperado (todo acumulado gracias a ResultPath):
10# {"pipeline":"ventas","fecha":"2024-01-15","verificacion":{"hay_datos":true,"num_archivos":3},
11#  "procesamiento":{"registros_procesados":45000,"errores":0},"notificacion":{"enviado":true}}

1# 5. Ver el historial paso a paso (que hizo cada estado)
2aws stepfunctions get-execution-history \
3  --execution-arn "$EXEC_ARN" \
4  --endpoint-url http://localhost:4566 \
5  --query 'events[?type==`PassStateExited`].[stateExitedEventDetails.name, stateExitedEventDetails.output]' \
6  --output table
7
8# Muestra la acumulacion: cada estado ve lo que tenia el anterior + lo suyo