Pipeline de Datos Serverless

import boto3
import json
import uuid
from datetime import datetime, timezone
from typing import Any

kinesis = boto3.client("kinesis", region_name="us-east-1")
STREAM_NAME = "ecommerce-events"


def publish_event(event_type: str, payload: dict[str, Any], user_id: str) -> None:
    """
    Publica un evento al stream de Kinesis.

    Args:
        event_type: Tipo de evento ('page_view', 'add_to_cart', 'purchase')
        payload: Datos del evento
        user_id: ID del usuario (usado como partition key para preservar orden)
    """
    record = {
        "eventId": str(uuid.uuid4()),
        "eventType": event_type,
        "userId": user_id,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "payload": payload,
        "metadata": {
            "source": "backend-api",
            "version": "1.0",
        },
    }

    kinesis.put_record(
        StreamName=STREAM_NAME,
        Data=json.dumps(record),
        PartitionKey=user_id,  # mismo shard para el mismo usuario → orden garantizado
    )


def publish_batch(events: list[dict]) -> dict:
    """Publica hasta 500 eventos en un solo request."""
    records = [
        {
            "Data": json.dumps(event),
            "PartitionKey": event.get("userId", str(uuid.uuid4())),
        }
        for event in events
    ]

    response = kinesis.put_records(StreamName=STREAM_NAME, Records=records)

    failed = response.get("FailedRecordCount", 0)
    if failed > 0:
        print(f"WARNING: {failed} records failed to publish")

    return response

import base64
import json
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
import io
from datetime import datetime, timezone

s3 = boto3.client("s3")
OUTPUT_BUCKET = "my-data-lake"


def handler(event: dict, context) -> dict:
    """
    Consume records de Kinesis, transforma a Parquet y escribe en S3.
    BisectOnFunctionError: si falla, Kinesis divide el batch y reintenta.
    """
    records = []
    failed_items = []

    for i, record in enumerate(event["Records"]):
        try:
            # Kinesis codifica el data en base64
            raw = base64.b64decode(record["kinesis"]["data"])
            event_data = json.loads(raw)
            records.append(transform(event_data))
        except Exception as e:
            print(f"Failed to process record {i}: {e}")
            failed_items.append(
                {"itemIdentifier": record["kinesis"]["sequenceNumber"]}
            )

    if records:
        write_parquet_to_s3(records)

    return {"batchItemFailures": failed_items}


def transform(raw: dict) -> dict:
    """Normaliza y enriquece el evento."""
    ts = datetime.fromisoformat(raw["timestamp"])
    return {
        "event_id": raw["eventId"],
        "event_type": raw["eventType"],
        "user_id": raw["userId"],
        "ts": raw["timestamp"],
        "year": ts.year,
        "month": ts.month,
        "day": ts.day,
        "hour": ts.hour,
        # Flatten payload
        **raw.get("payload", {}),
    }


def write_parquet_to_s3(records: list[dict]) -> None:
    now = datetime.now(timezone.utc)
    # Particionado Hive-compatible para partition pruning en Athena
    key = (
        f"events/year={now.year}/month={now.month:02d}/"
        f"day={now.day:02d}/hour={now.hour:02d}/"
        f"part-{now.strftime('%H%M%S%f')}.parquet"
    )

    table = pa.Table.from_pylist(records)
    buffer = io.BytesIO()
    pq.write_table(table, buffer, compression="snappy")
    buffer.seek(0)

    s3.put_object(Bucket=OUTPUT_BUCKET, Key=key, Body=buffer)
    print(f"Wrote {len(records)} records to s3://{OUTPUT_BUCKET}/{key}")