diff --git a/.env.example b/.env.example index d122f6b..ed55b56 100644 --- a/.env.example +++ b/.env.example @@ -6,5 +6,5 @@ AWS_REGION=us-east-1 # S3 S3_BUCKET_NAME=nombre-de-tu-bucket -S3_DATA_KEY=landing/datos.parquet +S3_DATA_KEY=curated/citibike_trips/citibike_curated.snappy.parquet S3_MODEL_PREFIX=models/ diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e3e39e5..227e593 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -42,7 +42,7 @@ jobs: pip install -r requirements-dev.txt - name: Lint (ruff) - run: ruff check ml tests + run: ruff check ml tests scripts - name: Tests (pytest + cobertura) run: pytest tests/ -v --cov=ml --cov-report=xml --cov-report=term diff --git a/README.md b/README.md index 7974604..a4b183c 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,56 @@ cp .env.example .env | `S3_DATA_KEY` | Ruta al Parquet curado en S3 | | `S3_MODEL_PREFIX` | Prefix donde subir artefactos (default: `models/`) | +### Recrear S3 con un nuevo Learner Lab + +Si el laboratorio anterior expiró y el bucket ya no existe, estos scripts +reconstruyen todo el demo a partir del Parquet raw local. + +```bash +# 1. Credenciales temporales del Learner Lab +export AWS_ACCESS_KEY_ID=... +export AWS_SECRET_ACCESS_KEY=... +export AWS_SESSION_TOKEN=... +export AWS_REGION=us-east-1 + +# 2. Generar dataset curado + sample liviano para el dashboard +python -m scripts.build_curated_dataset \ + --input /ruta/a/citibike_trips.parquet \ + --output-dir data/curated + +# 3. Entrenar y subir artefactos al bucket +S3_BUCKET_NAME=citibike-datalake-seb- \ +S3_MODEL_PREFIX=models/ \ +python -m ml.train --input data/curated/citibike_curated.snappy.parquet + +# 4. Crear el bucket público y subir datos + artefactos +python -m scripts.seed_s3 \ + --bucket citibike-datalake-seb- \ + --region us-east-1 \ + --landing /ruta/a/citibike_trips.parquet \ + --public-read +``` + +Layout resultante en S3: + +``` +landing/citibike_trips.parquet +curated/citibike_trips/citibike_curated.snappy.parquet +curated/citibike_trips/citibike_dashboard_sample.parquet +models/modelo.pkl +models/transformers.pkl +models/metricas.json +``` + +- `train.yml` usa `S3_DATA_KEY` (dataset curado completo) y sube artefactos a + `S3_MODEL_PREFIX`. +- El dashboard usa el sample `citibike_dashboard_sample.parquet` para responder + rápido y sin descargar las 473k filas completas. +- El bucket es público de solo lectura; el dashboard no necesita credenciales. +- Las credenciales de Learner Lab expiran. El dashboard sigue funcionando por el + bucket público, pero hay que renovar los secrets del repo para volver a + ejecutar `train.yml`. + --- ## Uso local diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..b0f167b --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Scripts de soporte para recrear el dataset y el bucket S3 del demo.""" diff --git a/scripts/build_curated_dataset.py b/scripts/build_curated_dataset.py new file mode 100644 index 0000000..ef43e33 --- /dev/null +++ b/scripts/build_curated_dataset.py @@ -0,0 +1,133 @@ +"""Genera el dataset curado y el sample del dashboard desde el Parquet raw. + +Replica el ETL de AWS Glue y normaliza las columnas que exige el contrato de +``ml/data_validation.py``. Se usa para recrear el demo sin depender de Glue. + +Uso: + python -m scripts.build_curated_dataset \ + --input /ruta/citibike_trips.parquet \ + --output-dir data/curated +""" +from __future__ import annotations + +import argparse +from pathlib import Path + +import pandas as pd + +from ml.data_validation import validate_dataframe + +GENDER_MAP = {"male": "Male", "female": "Female", "unknown": "Unknown"} + +CURATED_COLUMNS = [ + "tripduration", + "starttime", + "start_station_id", + "usertype", + "gender", + "birth_year", + "year", + "hour", + "month", + "dayofweek", + "is_weekend", + "age", +] + +REQUIRED_RAW_COLUMNS = { + "tripduration", + "starttime", + "start_station_id", + "usertype", + "gender", + "birth_year", +} + + +def build_curated(df: pd.DataFrame) -> pd.DataFrame: + """Aplica el ETL y devuelve un dataframe listo para entrenar.""" + missing = REQUIRED_RAW_COLUMNS - set(df.columns) + if missing: + raise ValueError(f"Faltan columnas requeridas en el raw: {sorted(missing)}") + + curated = df.copy() + curated["starttime"] = pd.to_datetime(curated["starttime"], errors="coerce") + curated["gender"] = ( + curated["gender"] + .astype(str) + .str.strip() + .str.lower() + .map(GENDER_MAP) + .fillna("Unknown") + ) + + curated["year"] = curated["starttime"].dt.year + curated["hour"] = curated["starttime"].dt.hour + curated["month"] = curated["starttime"].dt.month + curated["dayofweek"] = curated["starttime"].dt.dayofweek + curated["is_weekend"] = curated["dayofweek"].isin([5, 6]).astype(int) + curated["age"] = curated["year"] - curated["birth_year"] + + curated = curated[ + (curated["tripduration"] > 60) + & (curated["tripduration"] < 3600) + & (curated["age"] >= 0) + & (curated["age"] <= 90) + & (curated["birth_year"] >= 1900) + & (curated["birth_year"] <= 2025) + & curated["starttime"].notna() + ].copy() + + curated = curated.drop(columns=["end_station_id"], errors="ignore") + curated = curated[CURATED_COLUMNS] + + result = validate_dataframe(curated) + result.raise_if_invalid() + return curated + + +def main() -> None: + parser = argparse.ArgumentParser(description="Genera el dataset curado de Citi Bike") + parser.add_argument("--input", required=True, help="Parquet raw de entrada") + parser.add_argument( + "--output-dir", default="data/curated", help="Directorio donde guardar los Parquet" + ) + parser.add_argument( + "--sample-size", type=int, default=50_000, help="Filas del sample para el dashboard" + ) + parser.add_argument( + "--sample-name", + default="citibike_dashboard_sample.parquet", + help="Nombre del Parquet del sample", + ) + parser.add_argument("--seed", type=int, default=42, help="Semilla para el sample") + args = parser.parse_args() + + input_path = Path(args.input) + if not input_path.exists(): + raise FileNotFoundError(f"No existe el archivo de entrada: {input_path}") + + output_dir = Path(args.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + print(f"[1/3] Leyendo {input_path}") + raw = pd.read_parquet(input_path) + print(f" Shape raw: {raw.shape}") + + print("[2/3] Construyendo dataset curado") + curated = build_curated(raw) + print(f" Shape curado: {curated.shape}") + + curated_path = output_dir / "citibike_curated.snappy.parquet" + curated.to_parquet(curated_path, index=False, compression="snappy") + print(f" Guardado: {curated_path}") + + sample_size = min(args.sample_size, len(curated)) + sample = curated.sample(n=sample_size, random_state=args.seed).reset_index(drop=True) + sample_path = output_dir / args.sample_name + sample.to_parquet(sample_path, index=False, compression="snappy") + print(f"[3/3] Sample dashboard: {sample.shape} -> {sample_path}") + + +if __name__ == "__main__": + main() diff --git a/scripts/seed_s3.py b/scripts/seed_s3.py new file mode 100644 index 0000000..88d13d8 --- /dev/null +++ b/scripts/seed_s3.py @@ -0,0 +1,164 @@ +"""Crea y puebla el bucket S3 del demo con datos y artefactos. + +Uso: + python -m scripts.seed_s3 \ + --bucket citibike-datalake-seb- \ + --region us-east-1 \ + --landing /ruta/citibike_trips.parquet \ + --curated-dir data/curated \ + --artifacts-dir ml/artefactos \ + --public-read + +Requiere credenciales AWS con permisos para crear el bucket, configurar su +policy y subir objetos. No contiene credenciales: usa la cadena por defecto de +boto3 (variables de entorno, perfil, rol, etc.). +""" +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +import boto3 +from botocore.exceptions import ClientError + +S3_DATA_PREFIX = "curated/citibike_trips" +S3_MODEL_PREFIX = "models" +S3_LANDING_PREFIX = "landing" + + +def _public_policy(bucket: str) -> str: + return json.dumps( + { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "PublicReadGetObject", + "Effect": "Allow", + "Principal": "*", + "Action": "s3:GetObject", + "Resource": f"arn:aws:s3:::{bucket}/*", + } + ], + } + ) + + +def ensure_bucket(s3, bucket: str, region: str) -> None: + try: + s3.head_bucket(Bucket=bucket) + print(f"Bucket existente: {bucket}") + return + except ClientError as exc: + code = exc.response.get("Error", {}).get("Code", "") + if code not in {"404", "NoSuchBucket"}: + raise + + kwargs = {"Bucket": bucket} + if region != "us-east-1": + kwargs["CreateBucketConfiguration"] = {"LocationConstraint": region} + s3.create_bucket(**kwargs) + print(f"Bucket creado: {bucket}") + + +def make_public(s3, bucket: str) -> None: + s3.put_public_access_block( + Bucket=bucket, + PublicAccessBlockConfiguration={ + "BlockPublicAcls": False, + "IgnorePublicAcls": False, + "BlockPublicPolicy": False, + "RestrictPublicBuckets": False, + }, + ) + s3.put_bucket_policy(Bucket=bucket, Policy=_public_policy(bucket)) + print("Bucket configurado como publico de solo lectura") + + +def upload(s3, bucket: str, local: Path, key: str, content_type: str) -> None: + if not local.exists(): + print(f" [WARN] No existe {local}, se omite") + return + s3.upload_file( + str(local), + bucket, + key, + ExtraArgs={"ContentType": content_type}, + ) + print(f" {local} -> s3://{bucket}/{key}") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Crea y puebla el bucket S3 del demo") + parser.add_argument("--bucket", required=True, help="Nombre del bucket") + parser.add_argument("--region", default="us-east-1", help="Region AWS") + parser.add_argument("--landing", default=None, help="Parquet raw para la zona landing") + parser.add_argument("--curated-dir", default="data/curated", help="Directorio del curado") + parser.add_argument("--artifacts-dir", default="ml/artefactos", help="Directorio de artefactos") + parser.add_argument( + "--public-read", + action="store_true", + help="Configura el bucket como publico de solo lectura", + ) + args = parser.parse_args() + + s3 = boto3.client("s3", region_name=args.region) + ensure_bucket(s3, args.bucket, args.region) + + if args.public_read: + try: + make_public(s3, args.bucket) + except ClientError as exc: + print(f"[WARN] No se pudo configurar el bucket publico: {exc}") + print(" El dashboard necesitara credenciales o un bucket publico.") + + curated_dir = Path(args.curated_dir) + artifacts_dir = Path(args.artifacts_dir) + + print("\nSubiendo datos:") + if args.landing: + upload( + s3, + args.bucket, + Path(args.landing), + f"{S3_LANDING_PREFIX}/citibike_trips.parquet", + "application/octet-stream", + ) + upload( + s3, + args.bucket, + curated_dir / "citibike_curated.snappy.parquet", + f"{S3_DATA_PREFIX}/citibike_curated.snappy.parquet", + "application/octet-stream", + ) + upload( + s3, + args.bucket, + curated_dir / "citibike_dashboard_sample.parquet", + f"{S3_DATA_PREFIX}/citibike_dashboard_sample.parquet", + "application/octet-stream", + ) + + print("\nSubiendo artefactos:") + for name, content_type in [ + ("modelo.pkl", "application/octet-stream"), + ("transformers.pkl", "application/octet-stream"), + ("metricas.json", "application/json"), + ]: + upload( + s3, + args.bucket, + artifacts_dir / name, + f"{S3_MODEL_PREFIX}/{name}", + content_type, + ) + + base_url = f"https://{args.bucket}.s3.{args.region}.amazonaws.com" + print("\nListo. URLs de ejemplo:") + print(f" {base_url}/{S3_DATA_PREFIX}/citibike_dashboard_sample.parquet") + print(f" {base_url}/{S3_DATA_PREFIX}/citibike_curated.snappy.parquet") + print(f" {base_url}/{S3_MODEL_PREFIX}/metricas.json") + + +if __name__ == "__main__": + main()