%pip install -q "jupyter-notebook-slide @ git+https://github.com/dsevilla/jupyter-notebook-slide.git"
%load_ext notebook_slide

import notebook_slide as jnbs

# Colores de 26-27/teoria/tcdm.css, adaptados al tema de las sesiones.
jnbs.configure(
    background="#eaf2f8",
    foreground="#1f2933",
    border="#c9d6e1",
    heading="#0c304d",
    subheading="#0c304d",
    link="#174f7a",
    code_background="#eaf2f8",
    code_foreground="#0c304d",
    quote_background="#ffffff",
    font_family="Atkinson Hyperlegible, Inter, Aptos, Segoe UI, Helvetica, Arial, sans-serif",
    font_url="https://fonts.googleapis.com/css2?family=Atkinson+Hyperlegible:ital,wght@0,400;0,700;1,400;1,700&display=swap",
)
jnbs.register_slide_type("avance", "A continuación", "#174f7a")
jnbs.register_slide_type("resumen", "Recapitulación", "#a54467")
jnbs.register_slide_type("pregunta", "Pregunta guía", "#0c304d")
jnbs.register_slide_type("evaluacion", "Evaluación de la sesión", "#b3701a")
jnbs.register_slide_type(
    "titulo",
    "Sección",
    "#174f7a",
    layout="title",
    background="linear-gradient(135deg, #0c304d, #174f7a 62%, #a54467)",
    foreground="#ffffff",
    border="transparent",
    heading="#ffffff",
    subheading="#dceaf4",
)
jnbs.register_alias("diapositiva")

hdfs:///datalake/raw/tpcds/date_dim/part-...

cd ~/tcdm-public
git pull                       # trae s3/ y entorno/compose-s3.yml
make -C entorno warehouse-up   # si el clúster de S2 no sigue arrancado
make -C entorno s3-up

!hdfs dfs -ls -d /datalake/raw/tpcds

from pathlib import Path
from urllib.request import Request, urlopen

SESSION_FILES: tuple[str, ...] = (
    "check_s3_api.py",
    "read_s3_pyarrow.py",
    "read_s3_polars.py",
    "read_s3_duckdb.py",
)
base_url: str = "https://raw.githubusercontent.com/dsevilla/tcdm-public/26-27/s3"

for filename in SESSION_FILES:
    request: Request = Request(f"{base_url}/{filename}", headers={"User-Agent": "Mozilla/5.0"})
    with urlopen(request, timeout=30) as response:
        Path(filename).write_bytes(response.read())

print(f"Descargados {len(SESSION_FILES)} ficheros de la sesión al directorio de trabajo del kernel")

%%writefile requirements.txt
# Dependencias de los ejemplos de S3.
# Se utilizan únicamente APIs S3 estándar, de modo que el endpoint local se
# puede sustituir por un bucket real de Amazon S3 sin cambiar las lecturas.
# s3fs se usa en las lecturas de PyArrow y Polars; DuckDB usa httpfs.
# boto3 y aiobotocore se fijan en versiones que comparten botocore 1.43.56.
boto3==1.43.56
aiobotocore==3.9.0
duckdb
fsspec>=2026.2.0
polars
pyarrow
# Esta versión de s3fs proporciona la consulta de metadatos `modified` que
# utilizan los lectores de la sesión.
s3fs>=2026.2.0

%pip install -q -r requirements.txt

%env AWS_ENDPOINT_URL=http://rustfs:9000
%env AWS_ACCESS_KEY_ID=tcdm_student
%env AWS_SECRET_ACCESS_KEY=tcdm_student_secret
%env AWS_DEFAULT_REGION=us-east-1
%env AWS_S3_ADDRESSING_STYLE=path

import os

import boto3

buckets: list[str] = [
    bucket["Name"]
    for bucket in boto3.client("s3", endpoint_url=os.environ["AWS_ENDPOINT_URL"]).list_buckets()[
        "Buckets"
    ]
]
print(f"Buckets en RustFS: {buckets}")
assert (
    "tcdm-datalake" in buckets
), "No existe el bucket tcdm-datalake: revisa `make -C entorno s3-up`"

hdfs dfs -cat fichero.parquet
        │ flujo binario
        ▼
boto3.upload_fileobj(...) -> s3://tcdm-datalake/raw/tpcds/tabla/fichero

import os
from subprocess import PIPE, Popen, run
from typing import Any

import boto3

HDFS_ROOT: str = "/datalake/raw/tpcds"
S3_BUCKET: str = "tcdm-datalake"
S3_PREFIX: str = "raw/tpcds"

listing: str = run(
    ["hdfs", "dfs", "-ls", "-R", HDFS_ROOT],
    check=True,
    capture_output=True,
    text=True,
).stdout
hdfs_files: list[str] = sorted(
    fields[-1]
    for line in listing.splitlines()
    if (fields := line.split())
    if fields[0].startswith("-") and not fields[-1].endswith("/_SUCCESS")
)
table_names: set[str] = {path.removeprefix(f"{HDFS_ROOT}/").split("/", 1)[0] for path in hdfs_files}
assert len(table_names) == 24, sorted(table_names)
assert hdfs_files, "S2 no dejó ficheros Parquet en HDFS"

s3: Any = boto3.client("s3", endpoint_url=os.environ["AWS_ENDPOINT_URL"])
old_keys: list[str] = [
    item["Key"]
    for page in s3.get_paginator("list_objects_v2").paginate(
        Bucket=S3_BUCKET, Prefix=f"{S3_PREFIX}/"
    )
    for item in page.get("Contents", [])
]
for start in range(0, len(old_keys), 1000):
    batch: list[str] = old_keys[start : start + 1000]
    s3.delete_objects(
        Bucket=S3_BUCKET,
        Delete={"Objects": [{"Key": key} for key in batch], "Quiet": True},
    )

for position, hdfs_path in enumerate(hdfs_files, start=1):
    relative_path: str = hdfs_path.removeprefix(f"{HDFS_ROOT}/")
    s3_key: str = f"{S3_PREFIX}/{relative_path}"
    process: Popen[bytes] = Popen(["hdfs", "dfs", "-cat", hdfs_path], stdout=PIPE)
    assert process.stdout is not None
    try:
        s3.upload_fileobj(process.stdout, S3_BUCKET, s3_key)
    except Exception:
        process.kill()
        process.wait()
        raise
    finally:
        process.stdout.close()
    return_code: int = process.wait()
    if return_code != 0:
        raise RuntimeError(f"hdfs dfs -cat falló para {hdfs_path}: {return_code}")
    print(f"[{position}/{len(hdfs_files)}] {s3_key}")

uploaded_keys: list[str] = [
    item["Key"]
    for page in s3.get_paginator("list_objects_v2").paginate(
        Bucket=S3_BUCKET, Prefix=f"{S3_PREFIX}/"
    )
    for item in page.get("Contents", [])
]
assert len(uploaded_keys) == len(hdfs_files)
print(f"Copiadas {len(hdfs_files)} piezas Parquet de {len(table_names)} tablas.")

docker exec tcdm-s3-client aws s3 ls s3://tcdm-datalake/raw/tpcds/

!python check_s3_api.py tcdm-datalake raw/tpcds/date_dim --minimum-objects 1

PyArrow Parquet
      │
      ▼
FSSpecHandler
      │
      ▼
s3fs → botocore → API S3

!python read_s3_pyarrow.py tcdm-datalake raw/tpcds/date_dim \
  --key-column d_date_sk --expected-rows 73049

!python read_s3_polars.py tcdm-datalake raw/tpcds/date_dim \
  --key-column d_date_sk --expected-rows 73049

!python read_s3_duckdb.py tcdm-datalake raw/tpcds/date_dim \
  --key-column d_date_sk --expected-rows 73049

DESCRIBE SELECT * FROM read_parquet([...]);
SELECT * FROM read_parquet([...]) LIMIT 5;
SELECT count(*), min(d_date_sk), max(d_date_sk), sum(d_date_sk)
FROM read_parquet([...]);

import os

import polars as pl
import s3fs

filesystem: s3fs.S3FileSystem = s3fs.S3FileSystem(
    client_kwargs={"endpoint_url": os.environ["AWS_ENDPOINT_URL"]}
)
date_fragments: list[str] = sorted(filesystem.glob("tcdm-datalake/raw/tpcds/date_dim/*"))
assert date_fragments, "No hay fragmentos de date_dim en S3"
with filesystem.open(date_fragments[0], "rb") as stream:
    sample: pl.DataFrame = pl.read_parquet(stream, columns=["d_date", "d_year"])
print(sample.head())

Servicio	Función	¿Permanece ejecutándose?
`rustfs-s3`	API S3 y consola web	Sí
`bucket-setup-s3`	Crea el bucket si todavía no existe	No, termina correctamente
`s3-client`	Contenedor con AWS CLI, útil desde una terminal	Sí

Servicio	Dirección (desde el host)
API S3	http://localhost:9000
Consola RustFS	http://localhost:9001

Variable	RustFS local	Amazon S3
`AWS_ENDPOINT_URL`	`http://rustfs:9000` dentro de Docker	normalmente no se define
`AWS_ACCESS_KEY_ID`	`tcdm_student`	credencial IAM del laboratorio o rol
`AWS_SECRET_ACCESS_KEY`	secreto local del Compose	secreto temporal o rol IAM
`AWS_SESSION_TOKEN`	no se define	token temporal del laboratorio, si existe
`AWS_DEFAULT_REGION`	`us-east-1`	región del bucket
`AWS_S3_ADDRESSING_STYLE`	`path` dentro de Docker	preferentemente `virtual`

Situación	Orden	Consecuencia
Detener RustFS conservando sus objetos	`docker compose -f entorno/compose-s3.yml down`	Los objetos siguen en el volumen Docker; se recupera con `make -C entorno s3-up`.
Eliminar el laboratorio S3 por completo (destructivo)	`docker compose -f entorno/compose-s3.yml down -v --remove-orphans`	Borra también el volumen de RustFS y sus objetos. No elimina los bloques HDFS del clúster Hadoop. Después hay que ejecutar `make -C entorno s3-up` y volver a ejecutar el notebook.

Sesión 3 — Un data lake sobre almacenamiento S3-compatible¶

Objetivos¶

Diapositivas de la sesión¶

Sesión 3: un data lake sobre almacenamiento S3-compatible

Cómo vamos a recorrer la sesión

Cómo se evalúa esta sesión

De rutas HDFS a claves de objeto

Del data lake HDFS al data lake S3¶

El servicio local RustFS¶

Arrancar el clúster, RustFS y Jupyter

Arrancar el clúster y RustFS¶

Copiar `raw/tpcds` de HDFS al bucket

Traer los programas de la sesión al kernel¶

Instalar las dependencias de la sesión en este kernel¶

Crear el data lake S3 a partir de HDFS¶

Copiar los Parquet desde HDFS con la API S3¶

El mismo `raw`, ahora como objetos

Comprobar la API de AWS con boto3¶

Tres lectores para los mismos objetos

Explorar con PyArrow¶

Explorar con Polars¶

Explorar con DuckDB¶

Comparar los resultados¶

Tres caminos, la misma firma

Del laboratorio a Amazon S3 real

Configuración local y configuración AWS¶

Sesión 3

Detener, regenerar y limpiar¶

Preguntas para interpretar la experiencia¶

Evidencias para la siguiente sesión¶

Qué mostrar en el ordenador durante la reunión¶

Memoria escrita (una o dos páginas)¶

Qué es importante de cara al examen final¶

Continuación del curso¶

Sesión 3 — Un data lake sobre almacenamiento S3-compatible¶

Objetivos¶

Diapositivas de la sesión¶

Sesión 3: un data lake sobre almacenamiento S3-compatible

Cómo vamos a recorrer la sesión

Cómo se evalúa esta sesión

De rutas HDFS a claves de objeto

Del data lake HDFS al data lake S3¶

El servicio local RustFS¶

Arrancar el clúster, RustFS y Jupyter

Arrancar el clúster y RustFS¶

Copiar raw/tpcds de HDFS al bucket

Traer los programas de la sesión al kernel¶

Instalar las dependencias de la sesión en este kernel¶

Crear el data lake S3 a partir de HDFS¶

Copiar los Parquet desde HDFS con la API S3¶

El mismo raw, ahora como objetos

Comprobar la API de AWS con boto3¶

Tres lectores para los mismos objetos

Explorar con PyArrow¶

Explorar con Polars¶

Explorar con DuckDB¶

Comparar los resultados¶

Tres caminos, la misma firma

Del laboratorio a Amazon S3 real

Configuración local y configuración AWS¶

Sesión 3

Detener, regenerar y limpiar¶

Preguntas para interpretar la experiencia¶

Evidencias para la siguiente sesión¶

Qué mostrar en el ordenador durante la reunión¶

Memoria escrita (una o dos páginas)¶

Qué es importante de cara al examen final¶

Continuación del curso¶

Copiar `raw/tpcds` de HDFS al bucket

El mismo `raw`, ahora como objetos