Almacenamiento en la nube, lakehouse (Delta, Iceberg) y GPU (RAPIDS, cuDF).

Almacenamiento de objetos: S3 y GCS

Un lago de datos en la nube no vive en un disco, sino en un almacenamiento de objetos: Amazon S3, Google Cloud Storage (GCS), Azure Blob Storage o sus equivalentes compatibles con S3. Desde Python parecen carpetas (Polars, DuckDB y PyArrow leen s3://cubo/lago/ como leen lago/), pero funcionan de otra forma, y esas diferencias explican la mitad de las decisiones de diseño de un lago:

• No hay carpetas. Un cubo (bucket) de S3 es plano: cada objeto tiene una clave (lago/fecha=2025-03-14/parte-0.parquet) y las «carpetas» son prefijos compartidos que la consola presenta como tales11Nota 1 · AWS, «What is Amazon S3?», «Organizing objects using prefixes», «Copying objects» (renombrar), GetObject (Range) y ListObjectsV2: https://docs.aws.amazon.com/AmazonS3/latest/userguide/Welcome.html y https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html, consultados el 23 de octubre de 2026.. Listar es pedir las claves con un prefijo: ListObjectsV2 devuelve como mucho 1.000 por petición¹.
• Se escribe el objeto entero. No se puede modificar un trozo ni añadir al final: se sube (PUT) un objeto nuevo completo. Renombrar es copiar y borrar el original¹.
• Se lee por rangos. Un GET con la cabecera Range descarga un intervalo de bytes¹; así lee un lector de Parquet el pie y después solo los trozos de columna que necesita (lección 6.6.2).
• Consistencia fuerte. Desde diciembre de 2020, S3 garantiza que una lectura después de una escritura o un borrado ve el resultado en todas las regiones; GCS también ofrece consistencia fuerte global en la lectura tras escribir22Nota 2 · AWS, anuncio del 1 de diciembre de 2020 sobre consistencia fuerte en S3: https://aws.amazon.com/about-aws/whats-new/2020/12/amazon-s3-now-delivers-strong-read-after-write-consistency-automatically-for-all-applications/; Google Cloud, «Consistency»: https://cloud.google.com/storage/docs/consistency. Consultados el 23 de octubre de 2026..
• Cada petición se paga. En la página de precios de S3, en S3 Standard, los GET cuestan 0,0004 USD por cada 1.000 y los PUT, COPY, POST y LIST, 0,005 USD por cada 1.000 (Norte de Virginia, consultada el 23 de octubre de 2026)33Nota 3 · AWS, «Amazon S3 pricing»: https://aws.amazon.com/s3/pricing/, consultado el 23 de octubre de 2026. Los precios cambian y dependen de la región y la clase de almacenamiento.. El almacenamiento y la transferencia de datos se cobran aparte.

flowchart LR
  C[Cliente Python] -- LIST prefijo --> B[(Cubo S3)]
  C -- GET Range: pie --> B
  C -- GET Range: columnas --> B
  C -- PUT objeto entero --> B

En el entorno del curso no hay s3fs ni ninguna librería que imite S3, y la lección no usa la red. Para medir de todos modos, la primera celda define un «cubo» que es una carpeta local vista a través de fsspec (la misma capa que usa s3fs para S3) y que cuenta cada objeto escrito como un PUT y cada lectura como un GET por rango. No reproduce la latencia ni los precios de S3, pero sí el número de peticiones que haría un lector, que es lo que cuesta dinero y tiempo en la nube.

import math

import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.dataset as ds
import pyarrow.parquet as pq
import os

from fsspec.implementations.local import LocalFileSystem
from pyarrow.fs import FSSpecHandler, PyFileSystem

class AlmacenObjetos(LocalFileSystem):
    """Imita un cubo de S3 en una carpeta: cuenta PUT y GET."""
    def reiniciar(self):
        self.put = self.get = self.bytes_get = 0

    def _open(self, path, mode="rb", **kw):
        f = super()._open(path, mode, **kw)
        if "w" in mode:
            self.put += 1                  # un objeto entero por PUT
            return f
        leer = f.read
        def read(n=-1):
            datos = leer(n)
            self.get += 1                  # cada lectura, un GET
            self.bytes_get += len(datos)
            return datos
        f.read = read
        return f

CUBO = os.path.abspath("cubo")         # el «cubo»: una carpeta
s3 = AlmacenObjetos()
s3.reiniciar()
cubo = PyFileSystem(FSSpecHandler(s3))
print(type(cubo).__name__, "sobre", type(s3).__name__)

PyFileSystem sobre AlmacenObjetos

El código real contra S3 solo cambia la ruta y las credenciales (que no se escriben en el código: las toma del entorno o del perfil de AWS). Con s3fs instalado, las tres herramientas de la lección 6.6.2 leen así:

import duckdb
import polars as pl
import pyarrow.dataset as ds

ticks = pl.scan_parquet("s3://mi-cubo/lago/ticks/**/*.parquet",
                        hive_partitioning=True)  # credenciales: entorno
lago = ds.dataset("s3://mi-cubo/lago/ticks/", format="parquet",
                  partitioning="hive")
con = duckdb.connect()
con.sql("INSTALL httpfs; LOAD httpfs;")
con.sql("CREATE SECRET (TYPE s3, PROVIDER credential_chain)")
con.sql("SELECT count(*) FROM read_parquet("
        "'s3://mi-cubo/lago/ticks/**/*.parquet')")

Los datos de la prueba: dos millones de ticks sintéticos de 100 acciones en las 20 sesiones de marzo de 2025, guardados de tres formas: un único objeto, un objeto por día y un objeto por día y símbolo.

import exchange_calendars as xcals

tickers = sorted(pd.read_csv("panel5_activos.csv")["ticker"])[:100]
cal = xcals.get_calendar("XNYS")
sesiones = cal.sessions_in_range("2025-03-03", "2025-03-28")
rng = np.random.default_rng(665)
partes = []
for s in sesiones:
    n = rng.poisson(1000, len(tickers))
    abre = cal.session_open(s).value // 1000
    partes.append(pd.DataFrame({
        "fecha": s.date(),
        "simbolo": np.repeat(tickers, n),
        "ts": pd.to_datetime(
            rng.integers(abre, abre + 23_400_000_000, n.sum()),
            unit="us", utc=True),
        "precio": rng.lognormal(4, 0.5, n.sum()).round(2),
        "tamano": rng.geometric(0.02, n.sum()).astype("int32")}))
ticks = pd.concat(partes).sort_values(["fecha", "simbolo", "ts"])
tabla = pa.Table.from_pandas(ticks, preserve_index=False)

formato = ds.ParquetFileFormat()
opciones = formato.make_write_options(compression="zstd")
def escribir(ruta, particiones):
    s3.reiniciar()
    ds.write_dataset(tabla, ruta, filesystem=cubo, format=formato,
                     file_options=opciones, partitioning=particiones,
                     partitioning_flavor="hive" if particiones
                     else None,
                     basename_template="parte-{i}.parquet",
                     max_rows_per_group=131_072)
    claves = s3.find(ruta)
    return {"objetos": len(claves), "PUT": s3.put,
            "MB": round(sum(s3.size(k) for k in claves) / 1e6, 1)}

disenos = {"un fichero": (f"{CUBO}/uno", None),
           "por día": (f"{CUBO}/dia", ["fecha"]),
           "por día y símbolo": (f"{CUBO}/dia_simbolo",
                                 ["fecha", "simbolo"])}
escritura = pd.DataFrame([escribir(*v) for v in disenos.values()],
                         index=list(disenos))
print(f"{len(ticks):,} ticks, {len(sesiones)} sesiones,"
      f" {len(tickers)} símbolos")
escritura

1,999,891 ticks, 20 sesiones, 100 símbolos
                   objetos   PUT    MB
un fichero               1     1  22.1
por día                 20    20  22.3
por día y símbolo     2000  2000  27.9

Escribir cuesta un PUT por objeto: 1, 20 o 2.000. Y los 2.000 objetos pequeños ocupan un 26 % más (27,9 MB frente a 22,1): cada uno lleva su propio pie y comprime peor, como se vio en la lección 1.4.7. Ahora, la misma pareja de consultas sobre los tres diseños: los ticks de BAGO de un día y la columna precio entera.

GET_1000, PUT_LIST_1000 = 0.0004, 0.005   # USD, S3 Standard

def consultar(ruta, particiones, filtro, columnas):
    s3.reiniciar()
    esquema = pa.schema([tabla.schema.field(c)
                         for c in particiones or []])
    lago = ds.dataset(ruta, filesystem=cubo, format="parquet",
                      partitioning=ds.partitioning(
                          esquema, flavor="hive") if particiones
                      else None)
    claves = len(s3.find(ruta))
    for frag in lago.get_fragments(filtro):  # pies que hacen falta
        frag.ensure_complete_metadata()
    t = lago.to_table(                       # lecturas en orden fijo
        columns=columnas, filter=filtro, use_threads=False,
        fragment_scan_options=ds.ParquetFragmentScanOptions(
            pre_buffer=False))
    listados = math.ceil(claves / 1000)   # 1.000 claves por LIST
    return {"filas": t.num_rows, "LIST": listados, "GET": s3.get,
            "MB": round(s3.bytes_get / 1e6, 2),
            "USD_por_millón": round((s3.get * GET_1000 + listados
                                     * PUT_LIST_1000) * 1000, 2)}

un_dia = ((pc.field("fecha") == sesiones[9].date())
          & (pc.field("simbolo") == "BAGO"))
filas = {}
for k, (ruta, part) in disenos.items():
    filas[(k, "BAGO, un día")] = consultar(ruta, part, un_dia,
                                           ["ts", "precio"])
    filas[(k, "todo, una columna")] = consultar(ruta, part, None,
                                                ["precio"])
pd.DataFrame(list(filas.values()),
             index=pd.MultiIndex.from_tuples(filas))

                                       filas  LIST   GET     MB  USD_por_millón
un fichero        BAGO, un día          1016     1    10   0.79             9.0
                  todo, una columna  1999891     1    64   5.75            30.6
por día           BAGO, un día          1016     1     5   0.31             7.0
                  todo, una columna  1999891     1   102   7.22            45.8
por día y símbolo BAGO, un día          1016     2     4   0.04            11.6
                  todo, una columna  1999891     2  4061  36.90          1634.4

La figura de la cabecera dibuja la columna de coste. Lo que dicen las cifras:

• Para la consulta puntual, los tres diseños son baratos: entre 4 y 10 peticiones. El fichero único necesita más GET (10) porque tiene que leer el pie y los trozos de varias columnas en los grupos de filas candidatos; el diseño por día y símbolo lee solo 0,04 MB, pero necesita 2 LIST (2.000 claves, 1.000 por petición), que en S3 cuestan más que los GET.
• Para la lectura completa de una columna, los objetos pequeños son un desastre: 4.061 GET y 36,90 MB, más que todo el lago (27,9 MB). Con objetos de 14 KB, la primera lectura del pie (64 KB) ya trae el objeto entero, y después el lector vuelve a pedir la columna. Un millón de lecturas así costaría unos 1.634 USD solo en peticiones, frente a 31 USD con un único objeto.
• El diseño por día es el compromiso razonable aquí: barato en las dos consultas. Con datos reales de cientos de gigabytes, el equivalente es particionar por fecha con objetos de cientos de megabytes, no de kilobytes.

El tiempo sigue la misma lógica, con más fuerza: cada petición a S3 tiene una latencia de decenas de milisegundos, mucho mayor que la de un disco local, así que miles de peticiones pequeñas en serie pueden tardar minutos aunque descarguen pocos megabytes. Los lectores lo compensan lanzando peticiones en paralelo, pero el número de peticiones sigue siendo lo que hay que vigilar.

Lakehouse: un registro de transacciones sobre el lago

Un lago de ficheros tiene tres carencias que una base de datos no tiene: no hay transacciones (si un proceso reescribe diez ficheros y falla en el sexto, el lago queda a medias), no hay versiones (corregir un día borra lo que había) y dos escritores a la vez pueden pisarse. Los formatos de lakehouse (Delta Lake, Apache Iceberg, Apache Hudi) añaden encima de los Parquet un registro de transacciones: una lista ordenada de commits, cada uno con los ficheros que añade y los que retira. La tabla, en una versión, es el conjunto de ficheros vivos tras aplicar los commits hasta ella.

flowchart LR
  R[_registro/] --> V0[000000.json: add x15]
  R --> V1[000001.json: add x5]
  R --> V2[000002.json: remove, add]
  R --> V3[000003.json: compactar]
  V3 --> D[datos/*.parquet]

Los ficheros de datos nunca se modifican: corregir es escribir un fichero nuevo y retirar el viejo en el mismo commit (copy-on-write). Y el commit es atómico: el fichero 000004.json existe entero o no existe. Para entender la idea sin depender de ninguna librería, las celdas siguientes construyen un registro mínimo con ficheros JSON y Parquet. No es Delta ni Iceberg, pero tiene sus piezas: commits numerados, escritura con open(..., "x") (crear solo si no existe: el equivalente local de una escritura condicional), lectura de cualquier versión y limpieza de ficheros viejos.

import json
import os
from pathlib import Path

RAIZ = Path("tabla_ticks")
(RAIZ / "_registro").mkdir(parents=True)
(RAIZ / "datos").mkdir()

def confirmar(version, acciones):
    """Escribe el commit `version` solo si no existe (put-if-absent)."""
    ruta = RAIZ / "_registro" / f"{version:06d}.json"
    with open(ruta, "x") as f:           # "x": falla si ya existe
        json.dump(acciones, f)

def ficheros_vivos(version):
    vivos = set()
    for v in range(version + 1):
        ruta = RAIZ / "_registro" / f"{v:06d}.json"
        for a in json.loads(ruta.read_text()):
            if a["op"] == "add":
                vivos.add(a["f"])
            elif a["op"] == "remove":
                vivos.discard(a["f"])
    return sorted(vivos)

def ultima_version():
    return len(list((RAIZ / "_registro").glob("*.json"))) - 1

def leer(version=None):
    v = ultima_version() if version is None else version
    return pa.concat_tables(
        pq.read_table(RAIZ / "datos" / f) for f in ficheros_vivos(v))

def escribir_fichero(df, nombre):
    pq.write_table(pa.Table.from_pandas(df, preserve_index=False),
                   RAIZ / "datos" / nombre, compression="zstd")
    return {"op": "add", "f": nombre, "filas": len(df)}

dias = [d for d, _ in ticks.groupby("fecha")]
# versión 0: las 15 primeras sesiones, un fichero por día
confirmar(0, [escribir_fichero(ticks[ticks["fecha"] == d],
                               f"{d}-a.parquet") for d in dias[:15]])
# versión 1: llegan las 5 sesiones siguientes
confirmar(1, [escribir_fichero(ticks[ticks["fecha"] == d],
                               f"{d}-a.parquet") for d in dias[15:]])
print("versión", ultima_version(), "|",
      len(ficheros_vivos(1)), "ficheros vivos |",
      f"{leer().num_rows:,} filas")

versión 1 | 20 ficheros vivos | 1,999,891 filas

Dos commits: el primero añade las 15 primeras sesiones (un fichero por día) y el segundo, las 5 siguientes. La tabla en la versión 1 tiene 20 ficheros vivos y los 1.999.891 ticks. Ahora, dos operaciones que en un lago de ficheros sueltos serían arriesgadas: el proveedor corrige el 14 de marzo (anula las impresiones de una sola acción, que en estos datos son 2.058 filas) y después se compactan los 20 ficheros diarios en uno.

# versión 2: el proveedor corrige el 14 de marzo (copiar al escribir)
d14 = ticks[ticks["fecha"] == dias[9]].copy()
d14 = d14[d14["tamano"] > 1]        # anula las impresiones de 1 acción
confirmar(2, [{"op": "remove", "f": f"{dias[9]}-a.parquet"},
              escribir_fichero(d14, f"{dias[9]}-b.parquet")])
# versión 3: compactar los 20 ficheros diarios en uno
todo = leer().to_pandas()
confirmar(3, [{"op": "remove", "f": f} for f in ficheros_vivos(2)]
          + [escribir_fichero(todo, "compacto-0.parquet")])

historia = pd.DataFrame([
    {"versión": v, "ficheros_vivos": len(ficheros_vivos(v)),
     "filas": leer(v).num_rows,
     "filas_14_marzo": pc.sum(pc.equal(
         leer(v)["fecha"], pa.scalar(dias[9]))).as_py() or 0}
    for v in range(4)])
en_disco = len(list((RAIZ / "datos").iterdir()))
print(en_disco, "ficheros en disco")
historia

22 ficheros en disco
   versión  ficheros_vivos    filas  filas_14_marzo
0        0              15  1499553           99833
1        1              20  1999891           99833
2        2              20  1997833           97775
3        3               1  1997833           97775

Cada versión sigue legible (time travel): la 0 tiene 15 ficheros y 1.499.553 filas; la 1, 20 ficheros y 1.999.891; la 2 sustituye el fichero del 14 de marzo y baja a 97.775 filas ese día; la 3 tiene un solo fichero con las mismas 1.997.833 filas que la 2. Un lector que empezó a leer la versión 2 mientras se compactaba no ve nada raro: sus 20 ficheros siguen en disco, porque la compactación solo añade uno nuevo y anota en el registro que los viejos ya no cuentan. En disco hay 22 ficheros (los 20 originales, la corrección y el compacto) para 1 vivo.

La atomicidad también resuelve los escritores concurrentes. Si dos procesos leen la versión 3 y los dos intentan escribir la 4, solo uno puede crear 000004.json; el segundo recibe un error y debe releer el registro, comprobar si su cambio sigue siendo válido y volver a intentarlo con la 5 (concurrencia optimista):

version_leida = ultima_version()            # dos escritores leen la 3
confirmar(version_leida + 1,
          [{"op": "nota", "texto": "escritor A"}])
confirmar(version_leida + 1,                # el segundo llega tarde
          [{"op": "nota", "texto": "escritor B"}])

FileExistsError: [Errno 17] File exists: 'tabla_ticks/_registro/000004.json'

En un disco local, open(..., "x") es atómico. En el almacenamiento de objetos hace falta su equivalente: S3 admite escrituras condicionales con la cabecera If-None-Match, que impiden sobrescribir un objeto que ya existe44Nota 4 · AWS, «Conditional writes»: https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-writes.html, consultado el 23 de octubre de 2026., e Iceberg delega el cambio de versión en un catálogo que intercambia de forma atómica el puntero al fichero de metadatos vigente55Nota 5 · Apache Iceberg, especificación del formato (metadatos, snapshots, manifiestos y concurrencia optimista): https://iceberg.apache.org/spec/, consultado el 23 de octubre de 2026..

Lo que no limpia nadie son los ficheros que ya no usa ninguna versión. Vacuum los borra, y con ellos la posibilidad de viajar a las versiones que los usaban:

def vacuum(desde_version):
    """Borra los datos que no usa ninguna versión >= desde_version."""
    usados = set()
    for v in range(desde_version, ultima_version() + 1):
        usados |= set(ficheros_vivos(v))
    borrados = 0
    for f in (RAIZ / "datos").iterdir():
        if f.name not in usados:
            f.unlink()
            borrados += 1
    return borrados

print("borrados:", vacuum(desde_version=3),
      "| quedan:", len(list((RAIZ / "datos").iterdir())))
try:
    leer(1)
except FileNotFoundError:
    print("la versión 1 ya no se puede leer")
print(f"{leer().num_rows:,} filas en la última versión")

borrados: 21 | quedan: 1
la versión 1 ya no se puede leer
1,997,833 filas en la última versión

Borra 21 ficheros y deja el compacto. Desde ese momento, la versión 1 ya no se puede leer: el registro la describe, pero sus datos no existen. Por eso los formatos reales imponen un periodo de retención (por ejemplo, no borrar nada que haya dejado de usarse hace menos de una semana) y por eso conviene decidir antes cuánto historial hay que poder reproducir, como con las versiones de ArcticDB en la lección 6.6.4.

Delta Lake e Iceberg

Delta Lake guarda el registro en _delta_log/: un fichero JSON por commit (00000000000000000042.json) con acciones como las de la celda, más checkpoints periódicos en Parquet que resumen el estado para no tener que leer todos los JSON desde el principio66Nota 6 · Delta Lake, protocolo de transacciones: https://github.com/delta-io/delta/blob/master/PROTOCOL.md; delta-rs: https://delta-io.github.io/delta-rs/latest/. Consultados el 23 de octubre de 2026.. Apache Iceberg usa una jerarquía de metadatos: un fichero de metadatos de la tabla con sus snapshots; cada snapshot, una lista de manifiestos (desde la versión 4 del formato, un manifiesto raíz) y cada manifiesto, una lista de ficheros de datos con sus estadísticas; un catálogo guarda el puntero al fichero de metadatos vigente y cada commit lo intercambia de forma atómica⁵.

Ninguno de los dos está instalado en el entorno del curso (deltalake y pyiceberg no se añadieron), así que el código no se ejecuta. Con deltalake (delta-rs, la implementación en Rust con API de Python), las operaciones de la sección anterior son:

from deltalake import DeltaTable, write_deltalake

write_deltalake("s3://mi-cubo/delta/ticks", ticks_v0,
                partition_by=["fecha"])                 # versión 0
write_deltalake("s3://mi-cubo/delta/ticks", ticks_nuevos,
                mode="append")                           # versión 1
dt = DeltaTable("s3://mi-cubo/delta/ticks")
dt.load_as_version(0)                    # viaje en el tiempo
v0 = dt.to_pyarrow_table()
dt = DeltaTable("s3://mi-cubo/delta/ticks")
dt.optimize.compact()                    # compactar ficheros pequeños
dt.vacuum(retention_hours=168, dry_run=False)

Y con PyIceberg, que necesita un catálogo (REST, Glue, SQL…) configurado:

from pyiceberg.catalog import load_catalog

catalogo = load_catalog("produccion")   # en .pyiceberg.yaml
tabla = catalogo.load_table("mercado.ticks")
bago = tabla.scan(row_filter="simbolo == 'BAGO'",
                  selected_fields=("ts", "precio")).to_arrow()
for s in tabla.snapshots():
    print(s.snapshot_id, s.timestamp_ms)
antes = tabla.scan(snapshot_id=tabla.snapshots()[0].snapshot_id)

La elección entre los dos suele venir dada por la plataforma y los motores que ya se usan; la diferencia de diseño más visible es que Iceberg hace explícito el catálogo, mientras que Delta guarda todo el estado junto a los datos. Para un investigador solo con un lago de Parquet, ninguno es imprescindible; empiezan a compensar cuando varios procesos escriben en el mismo lago o hace falta reproducir el estado de los datos de una fecha.

GPU: RAPIDS, cuDF y Polars

Las GPU, pensadas para gráficos y hoy para entrenar redes, también sirven para tablas: miles de núcleos aplican la misma operación a millones de filas a la vez. RAPIDS es el conjunto de librerías de NVIDIA para eso, y cuDF su DataFrame en GPU. Su modo cudf.pandas acelera código de pandas sin cambiarlo: las operaciones que cuDF sabe hacer van a la GPU y las demás vuelven automáticamente a pandas en la CPU77Nota 7 · NVIDIA RAPIDS, cudf.pandas y guía de instalación: https://docs.rapids.ai/api/cudf/stable/cudf_pandas/ y https://docs.rapids.ai/install/, consultados el 23 de octubre de 2026..

# En Jupyter: %load_ext cudf.pandas antes de importar pandas
# Como script: python -m cudf.pandas barras_diarias.py
import pandas as pd

ticks = pd.read_parquet("ticks/")             # en la GPU si se puede
diarias = ticks.groupby(["simbolo", "fecha"]).agg(
    volumen=("tamano", "sum"), maximo=("precio", "max"))

Polars tiene un motor de GPU (en beta abierta, sobre cuDF) que se pide con collect(engine="gpu"); si una operación no está soportada, la consulta se ejecuta entera en la CPU, salvo que se pida lo contrario con pl.GPUEngine(raise_on_fail=True)88Nota 8 · Polars, «GPU Support»: https://docs.pola.rs/user-guide/gpu-support/, consultado el 23 de octubre de 2026.. Los requisitos, según la documentación de RAPIDS y de Polars: una GPU de NVIDIA de arquitectura Volta o posterior (capacidad de cómputo 7.0 o superior), CUDA 12 o 13 y Linux o WSL2⁷⁸. Sin nada de eso, Polars ni siquiera intenta la consulta: :::cell{title="Polars con engine="gpu" sin GPU"}

import polars as pl

consulta = pl.LazyFrame(ticks).group_by("simbolo").agg(
    pl.col("tamano").sum())
try:
    consulta.collect(engine="gpu")
except ModuleNotFoundError as e:
    print(str(e).splitlines()[0])
print(consulta.collect().height, "grupos en CPU")

GPU engine requested, but required package 'cudf_polars' could not be imported.
100 grupos en CPU

:::

Sin el paquete cudf_polars la petición falla con un mensaje claro; la misma consulta en CPU devuelve los 100 grupos. En una máquina con GPU y pip install polars[gpu], el código no cambiaría.

Cuándo compensa una GPU para datos tabulares: agregaciones, uniones y ordenaciones sobre decenas o cientos de millones de filas que caben en la memoria de la tarjeta, repetidas muchas veces (un barrido de features, un backtest vectorizado sobre muchos activos). Cuándo no: datos que no caben en la memoria de la GPU (mucho menor que la RAM de un servidor), consultas que leen poco y calculan menos (el cuello de botella es el disco o la red, no el cálculo) y código con muchas operaciones que la GPU no soporta, que van y vuelven a la CPU copiando datos por el bus. Como con los clústeres de la lección 6.6.3, primero hay que medir dónde se va el tiempo.

Costes y decisiones

Ejercicios

1. Un backtest pide los ticks de cada uno de los 100 símbolos en cada una de las 20 sesiones (2.000 consultas puntuales como «BAGO, un día»). Con la tabla de peticiones de la lección, ¿cuánto cuesta en peticiones de S3 con cada diseño? ¿Y si se lanzara cada noche durante un año?

tabla_p = pd.DataFrame(list(filas.values()),
                       index=pd.MultiIndex.from_tuples(filas))
puntual = tabla_p.xs("BAGO, un día", level=1)
coste = (puntual["GET"] * GET_1000 + puntual["LIST"] * PUT_LIST_1000) \
    / 1000 * 2000
pd.DataFrame({"USD_backtest": coste.round(4),
              "USD_año": (coste * 252).round(2)})

                   USD_backtest  USD_año
un fichero               0.0180     4.54
por día                  0.0140     3.53
por día y símbolo        0.0232     5.85

2. Escribe una función que, leyendo solo el registro (sin abrir ningún Parquet), devuelva para cada versión cuántos ficheros añade y retira y cuántas filas añade.

def historial():
    filas_h = []
    for v in range(ultima_version() + 1):
        ruta = RAIZ / "_registro" / f"{v:06d}.json"
        acciones = json.loads(ruta.read_text())
        filas_h.append({
            "versión": v,
            "añade": sum(a["op"] == "add" for a in acciones),
            "retira": sum(a["op"] == "remove" for a in acciones),
            "filas_añadidas": sum(a.get("filas", 0) for a in acciones
                                  if a["op"] == "add")})
    return pd.DataFrame(filas_h)

historial()

   versión  añade  retira  filas_añadidas
0        0     15       0         1499553
1        1      5       0          500338
2        2      1       1           97775
3        3      1      20         1997833
4        4      0       0               0

import shutil

shutil.rmtree(RAIZ)
s3.rm(CUBO, recursive=True)
print(sorted(p.name for p in Path(".").glob("tabla_*")),
      s3.exists(CUBO))

[] False

3. Un equipo procesa cada noche en pandas 300 millones de filas de barras de 1 minuto (unos 20 GB en memoria) para calcular 40 features. Tarda dos horas. Alguien propone comprar una GPU. ¿Qué comprobarías antes?