Proyecto: pipeline end-to-end en AWS

1import boto3
2import json
3
4# Helper para crear clientes
5def client(service):
6    return boto3.client(service, endpoint_url='http://localhost:4566',
7                        aws_access_key_id='test', aws_secret_access_key='test',
8                        region_name='eu-west-1')
9
10s3 = client('s3')
11iam = client('iam')
12glue = client('glue')
13sfn = client('stepfunctions')
14lambda_client = client('lambda')
15
16# Crear buckets
17for bucket in ['fashionstore-raw', 'fashionstore-processed', 'fashionstore-athena-results']:
18    s3.create_bucket(Bucket=bucket,
19                    CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'})
20    print(f"[OK] Bucket: {bucket}")
21
22# Crear database en Glue Catalog
23glue.create_database(DatabaseInput={'Name': 'fashionstore_prod'})
24print("[OK] Database: fashionstore_prod")

Paso 1: infraestructura base — buckets + database del catálogo

1# Role para Lambdas del pipeline
2trust = json.dumps({"Version": "2012-10-17", "Statement": [
3    {"Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole"}
4]})
5
6iam.create_role(RoleName='pipeline-lambda-role', AssumeRolePolicyDocument=trust)
7
8# Policy: las Lambdas pueden leer raw, escribir processed, y actualizar catálogo
9policy_doc = json.dumps({"Version": "2012-10-17", "Statement": [
10    {"Effect": "Allow", "Action": ["s3:GetObject"], "Resource": ["arn:aws:s3:::fashionstore-raw/*"]},
11    {"Effect": "Allow", "Action": ["s3:PutObject"], "Resource": ["arn:aws:s3:::fashionstore-processed/*"]},
12    {"Effect": "Allow", "Action": ["glue:*"], "Resource": ["*"]},
13    {"Effect": "Allow", "Action": ["logs:*"], "Resource": ["*"]},
14]})
15
16policy = iam.create_policy(PolicyName='pipeline-lambda-policy', PolicyDocument=policy_doc)
17iam.attach_role_policy(RoleName='pipeline-lambda-role', PolicyArn=policy['Policy']['Arn'])
18print("[OK] IAM configurado con mínimo privilegio")

Paso 2: roles con permisos mínimos — cada Lambda solo accede a lo que necesita

1import zipfile, io
2
3# El endpoint que se ve DESDE DENTRO de una Lambda no es el tuyo.
4# Tu hablas con LocalStack por localhost:4566 porque el puerto esta publicado
5# en tu maquina. Pero LocalStack ejecuta cada Lambda en su propio contenedor,
6# y ahi localhost es ese contenedor: la llamada muere con connection refused.
7# LocalStack levanta un servidor DNS y resuelve localhost.localstack.cloud a
8# su propio contenedor, y lo configura automaticamente en los contenedores que
9# crea (Lambda entre ellos). Ese es el nombre que funciona dentro.
10ENDPOINT_EN_LAMBDA = 'http://localhost.localstack.cloud:4566'
11
12def crear_lambda(nombre, codigo):
13    """Helper para crear una Lambda con su código."""
14    buf = io.BytesIO()
15    with zipfile.ZipFile(buf, 'w') as zf:
16        zf.writestr('handler.py', codigo)
17    buf.seek(0)
18    lambda_client.create_function(
19        FunctionName=nombre, Runtime='python3.11',
20        Role='arn:aws:iam::000000000000:role/pipeline-lambda-role',
21        Handler='handler.handler', Code={'ZipFile': buf.read()},
22        Timeout=300, MemorySize=256,
23        # El codigo de la Lambda lee el endpoint de aqui. En AWS real no
24        # defines esta variable y boto3 va a los endpoints de verdad: el
25        # mismo codigo sirve en los dos sitios.
26        Environment={'Variables': {'AWS_ENDPOINT_URL': ENDPOINT_EN_LAMBDA}}
27    )
28    print(f"  [OK] Lambda: {nombre}")
29
30# Lambda 1: Validar archivo
31crear_lambda('fs-validar', """
32import json
33def handler(event, context):
34    key = event['key']
35    size = event.get('size', 0)
36    errores = []
37    if size == 0: errores.append('Archivo vacío')
38    if not key.endswith('.csv'): errores.append('No es CSV')
39    return {'key': key, 'valido': len(errores) == 0, 'errores': errores}
40""")
41
42# Lambda 2: Transformar CSV → JSON (simula Parquet)
43crear_lambda('fs-transformar', """
44import json, csv, io, os, boto3
45def handler(event, context):
46    # os.environ.get, no 'http://localhost:4566': dentro de la Lambda
47    # localhost es este contenedor. Si la variable no existe (AWS real),
48    # endpoint_url=None y boto3 usa los endpoints de AWS.
49    s3 = boto3.client('s3', endpoint_url=os.environ.get('AWS_ENDPOINT_URL'))
50    key = event['key']
51    obj = s3.get_object(Bucket='fashionstore-raw', Key=key)
52    content = obj['Body'].read().decode()
53    reader = csv.DictReader(io.StringIO(content))
54    records = list(reader)
55    out_key = key.replace('raw/', '').replace('.csv', '.json')
56    s3.put_object(Bucket='fashionstore-processed', Key=out_key,
57                  Body=json.dumps(records).encode())
58    return {'output_key': out_key, 'records': len(records)}
59""")
60
61# Lambda 3: Registrar en catálogo
62crear_lambda('fs-catalogar', """
63import json, boto3
64def handler(event, context):
65    return {'catalogado': True, 'tabla': 'ventas', 'registros': event.get('records', 0)}
66""")
67
68# Lambda 4: Notificar resultado
69crear_lambda('fs-notificar', """
70import json
71def handler(event, context):
72    print(f"NOTIFICACION: Pipeline completado - {json.dumps(event)}")
73    return {'notificado': True}
74""")
75print("\n[OK] Todas las Lambdas creadas")

Paso 3: 4 Lambdas independientes — cada una hace una cosa bien

1# Step Functions: orquesta las 4 Lambdas
2pipeline = {
3    "StartAt": "Validar",
4    "States": {
5        "Validar": {
6            "Type": "Task",
7            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:fs-validar",
8            "Next": "CheckValidez",
9            "Retry": [{"ErrorEquals": ["States.ALL"], "MaxAttempts": 2}]
10        },
11        "CheckValidez": {
12            "Type": "Choice",
13            "Choices": [{"Variable": "$.valido", "BooleanEquals": True, "Next": "Transformar"}],
14            "Default": "NotificarError"
15        },
16        "Transformar": {
17            "Type": "Task",
18            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:fs-transformar",
19            "Next": "Catalogar"
20        },
21        "Catalogar": {
22            "Type": "Task",
23            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:fs-catalogar",
24            "Next": "Notificar"
25        },
26        "Notificar": {
27            "Type": "Task",
28            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:fs-notificar",
29            "End": True
30        },
31        "NotificarError": {
32            "Type": "Task",
33            "Resource": "arn:aws:lambda:eu-west-1:000000000000:function:fs-notificar",
34            "End": True
35        }
36    }
37}
38
39sfn.create_state_machine(
40    name='fashionstore-etl-pipeline',
41    definition=json.dumps(pipeline),
42    roleArn='arn:aws:iam::000000000000:role/pipeline-lambda-role'
43)
44print("[OK] Pipeline Step Functions creado")

Paso 4: Step Functions conecta las Lambdas con lógica de flujo y reintentos

1import time
2
3# Simular llegada de un archivo CSV de ventas
4csv_data = """order_id,customer_id,product,quantity,amount,date
5ORD001,CUST42,Camiseta Premium,2,51.98,2024-01-15
6ORD002,CUST17,Zapatillas Runner,1,89.99,2024-01-15
7ORD003,CUST42,Pantalón Slim,1,49.99,2024-01-15
8ORD004,CUST08,Chaqueta Invierno,1,129.99,2024-01-15
9"""
10
11s3.put_object(Bucket='fashionstore-raw', Key='ventas/2024-01-15/ventas_hora_14.csv',
12              Body=csv_data.encode())
13print("Archivo subido a S3 raw")
14
15# Ejecutar el pipeline
16execution = sfn.start_execution(
17    stateMachineArn='arn:aws:states:eu-west-1:000000000000:stateMachine:fashionstore-etl-pipeline',
18    input=json.dumps({
19        "key": "ventas/2024-01-15/ventas_hora_14.csv",
20        "size": len(csv_data),
21        "bucket": "fashionstore-raw"
22    })
23)
24print(f"Pipeline ejecutándose: {execution['executionArn'].split(':')[-1]}")
25
26# Esperar resultado
27while True:
28    desc = sfn.describe_execution(executionArn=execution['executionArn'])
29    if desc['status'] != 'RUNNING':
30        break
31    time.sleep(1)
32
33print(f"\n{'[OK]' if desc['status'] == 'SUCCEEDED' else '[ERROR]'} Pipeline: {desc['status']}")
34if 'output' in desc:
35    print(f"   Output: {desc['output']}")

Paso 5: simular un archivo y ver el pipeline en acción

1# Verificar que los datos llegaron a processed
2processed = s3.list_objects_v2(Bucket='fashionstore-processed')
3print("\nArchivos en fashionstore-processed:")
4for obj in processed.get('Contents', []):
5    print(f"   {obj['Key']} ({obj['Size']} bytes)")
6
7# Leer el archivo transformado
8if processed.get('Contents'):
9    key = processed['Contents'][0]['Key']
10    data = s3.get_object(Bucket='fashionstore-processed', Key=key)
11    contenido = json.loads(data['Body'].read().decode())
12    print(f"\nDatos transformados ({len(contenido)} registros):")
13    for registro in contenido[:3]:
14        print(f"   {registro}")

Paso 6: verificar que todo el pipeline funcionó correctamente

1# Lambda dispatcher: recibe evento S3 y arranca Step Functions
2dispatcher_code = """
3import json, os, boto3
4def handler(event, context):
5    # Mismo motivo que en fs-transformar: el endpoint viene del entorno,
6    # porque localhost aqui dentro es el contenedor de la propia Lambda.
7    sfn = boto3.client('stepfunctions',
8                       endpoint_url=os.environ.get('AWS_ENDPOINT_URL'))
9    for record in event.get('Records', []):
10        key = record['s3']['object']['key']
11        size = record['s3']['object']['size']
12        sfn.start_execution(
13            stateMachineArn='arn:aws:states:eu-west-1:000000000000:stateMachine:fashionstore-etl-pipeline',
14            input=json.dumps({'key': key, 'size': size, 'bucket': 'fashionstore-raw'})
15        )
16    return {'dispatched': len(event.get('Records', []))}
17"""
18
19crear_lambda('fs-dispatcher', dispatcher_code)
20
21# Configurar S3 para triggear el dispatcher
22s3.put_bucket_notification_configuration(
23    Bucket='fashionstore-raw',
24    NotificationConfiguration={
25        'LambdaFunctionConfigurations': [{
26            'LambdaFunctionArn': 'arn:aws:lambda:eu-west-1:000000000000:function:fs-dispatcher',
27            'Events': ['s3:ObjectCreated:*'],
28            'Filter': {'Key': {'FilterRules': [{'Name': 'suffix', 'Value': '.csv'}]}}
29        }]
30    }
31)
32print("[OK] Automatización completa: archivo nuevo → pipeline se ejecuta solo")

El pipeline es ahora 100% automático — zero intervención humana