--- title: "6.6.5 Nube, lakehouse y GPU" description: "S3 y GCS medidos en peticiones, el problema de los ficheros pequeños, un registro de transacciones tipo Delta o Iceberg y la GPU con RAPIDS y cuDF." url: "https://pythonparatrading.com/curso-python-trading/nivel-6/nube-lakehouse-y-gpu/" lang: "es" image: "https://pythonparatrading.com/og/curso-python-trading/nivel-6/nube-lakehouse-y-gpu.png" --- # 6.6.5 · Almacenamiento en la nube, lakehouse (Delta, Iceberg) y GPU (RAPIDS, cuDF) > En la nube, cada lectura es una petición que se paga y tarda. Esta lección guarda dos millones de ticks de tres formas en un cubo simulado y cuenta las peticiones y su coste, construye un registro de transacciones con JSON y Parquet para entender qué añaden Delta Lake e Iceberg (commits atómicos, viaje en el tiempo, compactación, vacuum) y sitúa la GPU con RAPIDS, cuDF y Polars como cultura, con sus requisitos. - Curso: [Nivel 6: IA generativa, LLMs, agentes, big data y datos alternativos](https://pythonparatrading.com/curso-python-trading/nivel-6/) - Módulo: 6.6 · Big data para trading - Duración: 60 min - Requisitos: [6.6.2 · Parquet, Arrow, Polars en streaming y DuckDB fuera de memoria](https://pythonparatrading.com/curso-python-trading/nivel-6/parquet-arrow-polars-streaming-y-duckdb/), [6.6.4 · Bases de series temporales: TimescaleDB, QuestDB, ClickHouse, ArcticDB y kdb+](https://pythonparatrading.com/curso-python-trading/nivel-6/bases-de-series-temporales-y-arcticdb/) ## Al terminar esta lección - Explicar en qué se diferencia el almacenamiento de objetos (S3, GCS) de un disco: claves planas, objetos enteros, lecturas por rango, consistencia y coste por petición. - Medir cuántas peticiones y bytes cuesta una consulta según el diseño del lago y reconocer el problema de los ficheros pequeños. - Construir un registro de transacciones mínimo con commits atómicos, viaje en el tiempo, compactación, concurrencia optimista y vacuum, y relacionarlo con Delta Lake e Iceberg. - Escribir el código de Delta Lake (delta-rs) y PyIceberg para las mismas operaciones a partir de su documentación. - Decidir cuándo compensa una GPU (RAPIDS, cuDF, Polars en GPU) para datos tabulares y qué comprobar antes. ## 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 tales[^1]. Listar es pedir las claves con un prefijo: `ListObjectsV2` devuelve como mucho 1.000 por petición[^1]. - **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[^1]. - **Se lee por rangos.** Un `GET` con la cabecera `Range` descarga un intervalo de bytes[^1]; 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 escribir[^2]. - **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)[^3]. El almacenamiento y la transferencia de datos se cobran aparte. ```mermaid 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. ```python 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__) ``` ```output 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í: ```python skip title="lago_en_s3.py" 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. ```python 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 ``` ```output 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. ```python 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)) ``` ```output 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. ```mermaid 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. ```python 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") ``` ```output 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. ```python # 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 ``` ```output 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 ``` *Fig. — Ficheros de datos en disco y ficheros vivos en cada versión del registro de transacciones de la lección.* Barras anchas con los ficheros en disco y estrechas con los vivos. Versión 0: 15 y 15. Versión 1: 20 y 20. Versión 2: 21 en disco y 20 vivos (la corrección retira uno). Versión 3: 22 en disco y 1 vivo tras compactar (destacado). Tras vacuum: 1 y 1. 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**): ```python raises 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"}]) ``` ```output 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 existe[^4], e Iceberg delega el cambio de versión en un catálogo que intercambia de forma atómica el puntero al fichero de metadatos vigente[^5]. 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: ```python 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") ``` ```output 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 principio[^6]. **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[^5]. 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: ```python skip title="delta_ticks.py" 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: ```python skip title="iceberg_ticks.py" 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 CPU[^7]. ```python skip title="cudf_pandas.py" # 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)`[^8]. 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[^7][^8]. Sin nada de eso, Polars ni siquiera intenta la consulta: ```python 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") ``` ```output 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 | Necesidad | Opción | Lo que cuesta | |---|---|---| | Lago compartido por un equipo, sin servidor | Parquet en S3 o GCS, particionado por fecha | Peticiones y almacenamiento; vigilar los ficheros pequeños | | Escrituras concurrentes, correcciones, versiones | Delta Lake o Iceberg sobre el lago | Un registro y, con Iceberg, un catálogo; compactación y vacuum periódicos | | Series por símbolo con versiones desde Python | ArcticDB sobre S3 (lección 6.6.4) | Licencia BSL; un modelo propio de datos | | Agregaciones masivas repetidas | GPU con cuDF o Polars en GPU | Hardware de NVIDIA, memoria de la tarjeta, soporte parcial de operaciones | | Investigar solo, datos de un disco | Parquet local, Polars y DuckDB (lección 6.6.2) | Nada nuevo | [^1]: AWS, «What is Amazon S3?», «Organizing objects using prefixes», «Copying objects» (renombrar), `GetObject` (Range) y `ListObjectsV2`: y , consultados el 23 de octubre de 2026. [^2]: AWS, anuncio del 1 de diciembre de 2020 sobre consistencia fuerte en S3: ; Google Cloud, «Consistency»: . Consultados el 23 de octubre de 2026. [^3]: AWS, «Amazon S3 pricing»: , consultado el 23 de octubre de 2026. Los precios cambian y dependen de la región y la clase de almacenamiento. [^4]: AWS, «Conditional writes»: , consultado el 23 de octubre de 2026. [^5]: Apache Iceberg, especificación del formato (metadatos, snapshots, manifiestos y concurrencia optimista): , consultado el 23 de octubre de 2026. [^6]: Delta Lake, protocolo de transacciones: ; delta-rs: . Consultados el 23 de octubre de 2026. [^7]: NVIDIA RAPIDS, `cudf.pandas` y guía de instalación: y , consultados el 23 de octubre de 2026. [^8]: Polars, «GPU Support»: , consultado el 23 de octubre de 2026. ## 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? **Solución.** ```python 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)}) ``` ```output 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 ``` Céntimos: entre 0,014 y 0,023 USD por backtest y entre 3,5 y 5,9 USD al año lanzándolo cada sesión. Para consultas puntuales, las peticiones de S3 son baratas con cualquier diseño razonable; los `LIST` pesan más que los `GET`, y por eso el diseño de 2.000 objetos, que necesita dos por consulta, es el más caro aunque descargue menos. El coste de verdad aparece en las lecturas completas con objetos pequeños (la segunda consulta de la lección) y, fuera de esta cuenta, en la transferencia de datos y en el tiempo: 2.000 consultas en serie con decenas de milisegundos de latencia cada petición son minutos. Un backtest que lee lo mismo cada noche debería además guardar una copia local o leer por lotes. 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. **Solución.** ```python 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() ``` ```output 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 ``` Cinco versiones: la 4 es la nota del escritor A, que no añade ni retira nada. El historial sale entero del registro, sin abrir ningún Parquet; los formatos reales guardan en cada acción mucho más (tamaño, estadísticas de columnas, número de filas) para que un lector pueda descartar ficheros sin abrirlos, como hace el pie de Parquet dentro de un fichero. La última celda borra la tabla y el cubo simulado. ```python import shutil shutil.rmtree(RAIZ) s3.rm(CUBO, recursive=True) print(sorted(p.name for p in Path(".").glob("tabla_*")), s3.exists(CUBO)) ``` ```output [] 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? **Solución.** Primero, dónde se van las dos horas: medir cuánto es lectura, cuánto conversión y cuánto cálculo, y el pico de memoria de cada paso (lección 6.6.1). Si la mayor parte es leer y convertir datos (CSV, columnas `object`, tipos mal elegidos), la GPU no ayuda: lo que ayuda es Parquet bien particionado y leer solo las columnas necesarias (lección 6.6.2). Segundo, si el cálculo es vectorizable: bucles de Python por fila o funciones propias con `apply` no van a la GPU, y en `cudf.pandas` volverían a la CPU copiando datos. Tercero, si cabe: 20 GB en pandas pueden ser bastante menos en Arrow, pero hay que comprobar que caben en la memoria de la tarjeta con sus estructuras intermedias. Cuarto, la alternativa barata: el mismo cálculo en Polars lazy o DuckDB en la CPU suele acelerar mucho un proceso de pandas sin hardware nuevo. Solo si después de todo eso el cuello de botella es cálculo sobre datos que caben en la GPU, compensa probarla, por ejemplo alquilando una por horas en la nube antes de comprarla, y medir con los datos reales. ## Autoevaluación 1. Leer la columna `precio` entera cuesta 64 `GET` con un único objeto y 4.061 con 2.000 objetos de 14 KB, que además descargan 36,9 MB de un lago de 27,9 MB. ¿Por qué? - A. Porque S3 comprime peor los objetos pequeños - B. Porque cada objeto exige al menos una lectura del pie y otra de la columna, y con objetos de 14 KB la lectura del pie (64 KB) ya trae el objeto entero: muchas peticiones y bytes repetidos - C. Porque los objetos pequeños no tienen estadísticas - D. Porque `LIST` devuelve solo 1.000 claves *Respuesta B: Porque cada objeto exige al menos una lectura del pie y otra de la columna, y con objetos de 14 KB la lectura del pie (64 KB) ya trae el objeto entero: muchas peticiones y bytes repetidos.* Un millón de lecturas así costaría unos 1.634 USD en peticiones frente a 31 USD con un objeto. Es el problema de los ficheros pequeños, y la solución es compactar en objetos de cientos de megabytes. 2. ¿Qué garantiza que dos escritores que leen la versión 3 de la tabla no publiquen los dos una versión 4 distinta? - A. Que cada uno escribe sus ficheros de datos en carpetas distintas - B. Un bloqueo del sistema operativo durante toda la escritura - C. Que el lector elige la versión más reciente - D. Que el commit se crea solo si no existe (`open(..., "x")` en local, escritura condicional en S3, intercambio atómico del puntero en el catálogo de Iceberg): el segundo recibe un error, relee y reintenta *Respuesta D: Que el commit se crea solo si no existe (`open(..., "x")` en local, escritura condicional en S3, intercambio atómico del puntero en el catálogo de Iceberg): el segundo recibe un error, relee y reintenta.* Es la concurrencia optimista: no se bloquea nada mientras se escriben los datos; solo el paso final, publicar el commit, es atómico. El segundo escritor obtiene `FileExistsError` y debe comprobar si su cambio sigue siendo válido. 3. Tras la compactación (versión 3) hay 22 ficheros en disco y 1 vivo. Después de `vacuum(desde_version=3)`, ¿qué se pierde? - A. Nada: el registro conserva las versiones - B. La última versión - C. La posibilidad de leer las versiones 0 a 2: el registro las describe, pero sus ficheros de datos se han borrado - D. La compactación, que se deshace *Respuesta C: La posibilidad de leer las versiones 0 a 2: el registro las describe, pero sus ficheros de datos se han borrado.* Vacuum borra los ficheros que no usa ninguna versión retenida. Por eso los formatos reales imponen un periodo de retención: el viaje en el tiempo solo llega hasta donde llegan los datos. 4. Una consulta con `collect(engine="gpu")` en Polars incluye una operación que el motor de GPU no soporta. ¿Qué ocurre por defecto? - A. La consulta entera se ejecuta en la CPU con el motor normal, salvo que se pida `GPUEngine(raise_on_fail=True)` - B. Falla siempre - C. Se ejecuta en la GPU de forma aproximada - D. Polars instala cuDF automáticamente *Respuesta A: La consulta entera se ejecuta en la CPU con el motor normal, salvo que se pida `GPUEngine(raise_on_fail=True)`.* Según la documentación de Polars, el motor de GPU (en beta abierta) vuelve de forma transparente a la CPU. Sin el paquete `cudf_polars` instalado, en cambio, la petición falla con `ModuleNotFoundError`, como en la lección. 5. ¿Qué cuenta la tabla de coste de la lección y qué deja fuera? - A. Todo el coste de S3 - B. Solo las peticiones `GET` y `LIST` a los precios de S3 Standard; el almacenamiento, la transferencia de datos y la latencia quedan fuera - C. Solo el almacenamiento - D. El coste de la GPU *Respuesta B: Solo las peticiones `GET` y `LIST` a los precios de S3 Standard; el almacenamiento, la transferencia de datos y la latencia quedan fuera.* La página de precios de S3 cobra por separado almacenamiento, peticiones y transferencia. La lección mide peticiones porque son lo que cambia con el diseño del lago; la latencia de cada petición convierte además miles de peticiones pequeñas en minutos. ## Ideas clave - El almacenamiento de objetos es plano, se escribe por objetos enteros, se lee por rangos, es fuertemente consistente (S3 desde diciembre de 2020) y cobra cada petición: 0,0004 USD por 1.000 GET y 0,005 USD por 1.000 PUT o LIST en S3 Standard. - Con los mismos dos millones de ticks, leer una columna entera cuesta 64 GET con un objeto, 102 con uno por día y 4.061 con 2.000 objetos pequeños, que además descargan más bytes que todo el lago: particionar por fecha con objetos grandes y compactar. - Un lakehouse añade un registro de transacciones sobre los Parquet: commits atómicos que añaden y retiran ficheros, copy-on-write para corregir, viaje en el tiempo, concurrencia optimista y vacuum, que borra los ficheros y con ellos las versiones antiguas. - Delta Lake guarda JSON por commit y checkpoints en `_delta_log/`; Iceberg, metadatos, snapshots y manifiestos con un catálogo que intercambia de forma atómica el puntero a la versión vigente. - La GPU (RAPIDS, `cudf.pandas`, Polars con `engine="gpu"`) acelera agregaciones masivas que caben en su memoria; exige NVIDIA Volta o posterior, CUDA 12 o 13 y Linux o WSL2, y antes de comprarla hay que medir si el cuello de botella es el cálculo. ## Recursos ### Documentación - [Amazon S3: guía de usuario (consistencia, prefijos, rangos)](https://docs.aws.amazon.com/AmazonS3/latest/userguide/Welcome.html) · docs.aws.amazon.com - [Amazon S3: precios](https://aws.amazon.com/s3/pricing/) · aws.amazon.com - [Delta Lake: protocolo del registro de transacciones](https://github.com/delta-io/delta/blob/master/PROTOCOL.md) · github.com - [Apache Iceberg: especificación de la tabla](https://iceberg.apache.org/spec/) · iceberg.apache.org - [PyIceberg: API de Python](https://py.iceberg.apache.org/api/) · py.iceberg.apache.org - [NVIDIA RAPIDS: cudf.pandas](https://docs.rapids.ai/api/cudf/stable/cudf_pandas/) · docs.rapids.ai - [Polars: soporte de GPU](https://docs.pola.rs/user-guide/gpu-support/) · docs.pola.rs Lección anterior: [6.6.4 · Bases de series temporales: TimescaleDB, QuestDB, ClickHouse, ArcticDB y kdb+](https://pythonparatrading.com/curso-python-trading/nivel-6/bases-de-series-temporales-y-arcticdb/) Lección siguiente: [6.7.1 · Panorama y tamaño del mercado de datos alternativos](https://pythonparatrading.com/curso-python-trading/nivel-6/panorama-del-mercado-de-datos-alternativos/)