--- title: "6.6.2 Parquet, Arrow, Polars en streaming y DuckDB" description: "21 millones de ticks en un portátil: predicate pushdown medido en bytes, Arrow sin copias, Polars lazy en streaming y DuckDB con límite de memoria." url: "https://pythonparatrading.com/curso-python-trading/nivel-6/parquet-arrow-polars-streaming-y-duckdb/" lang: "es" image: "https://pythonparatrading.com/og/curso-python-trading/nivel-6/parquet-arrow-polars-streaming-y-duckdb.png" --- # 6.6.2 · Parquet, Arrow, Polars en streaming y DuckDB fuera de memoria > Antes de pensar en un clúster conviene exprimir una sola máquina. Esta lección genera 21 millones de ticks en un lago de Parquet particionado, mide cuántos bytes lee cada consulta según la partición, el orden y las estadísticas, comprueba qué conversiones de Arrow copian datos y cuáles no, ejecuta Polars en streaming y fuerza a DuckDB a trabajar con 300 y con 100 MB de memoria. - 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: 70 min - Requisitos: [0.5.4 · Polars: API lazy frente a eager, expresiones, rendimiento y cuándo elegirlo frente a pandas](https://pythonparatrading.com/curso-python-trading/nivel-0/polars-lazy-y-expresiones/), [0.5.5 · DuckDB y SQL esencial: SELECT, JOIN, GROUP BY y funciones de ventana sobre Parquet](https://pythonparatrading.com/curso-python-trading/nivel-0/duckdb-y-sql-sobre-parquet/), [1.4.7 · Almacenamiento: CSV, Parquet, particionado, DuckDB, TimescaleDB y ArcticDB](https://pythonparatrading.com/curso-python-trading/nivel-1/almacenamiento-parquet-duckdb-y-bases-temporales/), [6.6.1 · Cuándo los datos son «big»: ticks, libros, opciones y universos](https://pythonparatrading.com/curso-python-trading/nivel-6/cuando-los-datos-son-big/) ## Al terminar esta lección - Leer el pie de un Parquet (grupos de filas, mínimos y máximos) y medir cuántos bytes ahorran la proyección, la poda de particiones y las estadísticas. - Elegir el orden de escritura y la partición según las consultas habituales, sabiendo que las estadísticas solo sirven con datos ordenados. - Pasar datos entre Arrow, Polars, pandas, NumPy y DuckDB sin copias, y reconocer cuándo la copia es inevitable (nulos, varios trozos, textos). - Leer el plan optimizado de Polars sobre un lago particionado y ejecutar consultas con el motor de streaming, sabiendo qué memoria acota y cuál no. - Ejecutar en DuckDB consultas mayores que la memoria con `memory_limit`, `temp_directory` e hilos, e interpretar un error de falta de memoria. ## Cuando la tabla no cabe en la memoria La lección 6.6.1 puso cifras a la palabra «big»: los ticks de un universo de acciones, los libros L2 y L3 o las cadenas de opciones llegan pronto a cientos de gigabytes. La reacción habitual es pensar en un clúster. Casi siempre es prematura: un portátil con un buen formato en disco y un motor que no carga lo que no necesita llega mucho más lejos de lo que parece. Esta lección recorre esa pila de una sola máquina, de abajo arriba: ```mermaid flowchart TB P[Parquet en disco] --> A[Arrow en memoria] A --> PL[Polars lazy] A --> D[DuckDB] A --> PD[pandas y NumPy] PL --> R[Resultado pequeño] D --> R ``` - **Parquet** decide cuánto hay que leer: columnas, particiones y grupos de filas con estadísticas (la lección 1.4.7 presentó el formato; aquí se mide qué se ahorra con cada pieza). - **Arrow** es el formato en memoria que comparten todas las herramientas de la figura: pasar datos de una a otra no tiene por qué costar una copia. - **Polars lazy y su motor de streaming** procesan la consulta por bloques, sin cargar la tabla entera (la API lazy y `explain()` están en la lección 0.5.4). - **DuckDB** ejecuta SQL con un límite de memoria y vuelca a disco lo que no cabe (el SQL básico, en la lección 0.5.5). Para medirlo hace falta un conjunto grande. La primera celda genera ticks sintéticos de 200 acciones del panel del nivel 5 (los tickers de `panel5_activos.csv`; los precios no tienen relación con los del panel) durante las 60 sesiones de la NYSE del primer trimestre de 2025. Cada acción tiene su propia actividad media, de unas pocas centenas a varios miles de ticks por sesión, y sus precios siguen un paseo aleatorio con una volatilidad diaria del 2 %. Se escribe un fichero Parquet por sesión, con las filas ordenadas por símbolo y hora, en carpetas con el formato Hive (`fecha=2025-03-14/`). ```python from pathlib import Path import exchange_calendars as xcals import numpy as np import pandas as pd import pyarrow as pa import pyarrow.parquet as pq tickers = np.array( sorted(pd.read_csv("panel5_activos.csv")["ticker"])[:200]) cal = xcals.get_calendar("XNYS") sesiones = cal.sessions_in_range("2025-01-02", "2025-03-31") rng = np.random.default_rng(662) actividad = np.exp(rng.normal(np.log(1250), 0.8, 200)) ultimo = np.exp(rng.uniform(np.log(10), np.log(400), 200)) CENTROS = pa.array(["XNYS", "ARCX", "BATS", "EDGX", "OSCURO"]) def ticks_de_una_sesion(sesion): global ultimo abre = cal.session_open(sesion).value // 1000 # µs, UTC cierra = cal.session_close(sesion).value // 1000 n = rng.poisson(actividad) # ticks de cada símbolo sim = np.repeat(np.arange(200), n) N = len(sim) inicio = np.repeat(np.cumsum(n) - n, n) ts = rng.integers(abre, cierra - 100_000, N) orden = np.lexsort((ts, sim)) # símbolo y hora ts = ts[orden] + (np.arange(N) - inicio) # sin empates paso = rng.normal(0, 1, N) * np.repeat( 0.02 / np.sqrt(np.maximum(n, 1)), n) c = np.concatenate([[0.0], np.cumsum(paso)]) camino = c[1:] - c[np.cumsum(n) - n].repeat(n) precio = np.round(np.repeat(ultimo, n) * np.exp(camino), 2) ultimo = np.where(n > 0, precio[np.cumsum(n) - 1], ultimo) return pa.table({ "ts": pa.array(ts, pa.timestamp("us", tz="UTC")), "simbolo": pa.array(tickers).take(sim), "precio": precio, "tamano": rng.geometric(0.02, N).astype(np.int32), "lado": rng.choice(np.array([-1, 1], np.int8), N), "centro": CENTROS.take( rng.choice(5, N, p=[.3, .2, .15, .15, .2])), }) filas = memoria = 0 for s in sesiones: t = ticks_de_una_sesion(s) ruta = Path(f"ticks/fecha={s.date()}/parte-0.parquet") ruta.parent.mkdir(parents=True) pq.write_table(t, ruta, compression="zstd", row_group_size=65_536) filas += t.num_rows memoria += t.nbytes ficheros = sorted(Path("ticks").rglob("*.parquet")) disco = sum(p.stat().st_size for p in ficheros) print(f"{filas:,} ticks de {len(tickers)} símbolos" f" en {len(sesiones)} sesiones") print(f"en Arrow (memoria): {memoria / 1e6:,.0f} MB") print(f"en Parquet (disco): {disco / 1e6:,.0f} MB") ``` ```output 20,783,757 ticks de 200 símbolos en 60 sesiones en Arrow (memoria): 783 MB en Parquet (disco): 220 MB ``` Casi 21 millones de ticks: 783 MB en memoria con el formato de Arrow y 220 MB en disco con Parquet y zstd, 3,6 veces menos. Es un tamaño pensado para que la lección se ejecute en un minuto en un portátil (la celda tarda unos diez segundos en la máquina en la que se escribió); los mecanismos son los mismos con cien veces más datos, y las proporciones que se miden abajo no dependen del tamaño. ## Dentro del fichero: grupos de filas y estadísticas Un Parquet no es un bloque opaco. Se divide en **grupos de filas** (*row groups*); dentro de cada grupo, cada columna va por separado (*column chunk*) en páginas comprimidas, y al final del fichero va el **pie** (*footer*) con el esquema, la posición de cada trozo y sus **estadísticas**: el mínimo, el máximo y el número de nulos de cada columna en cada grupo[^1]. ```mermaid flowchart TB F[Fichero Parquet] --> G0[Grupo de filas 0] F --> G1[Grupo de filas 1] F --> PIE[Pie: esquema y min/max] G0 --> C1[Columna ts] G0 --> C2[Columna simbolo] C2 --> PG[Páginas zstd] ``` El pie es lo primero que lee cualquier lector: con él decide qué trozos le hacen falta sin abrir el resto. El de la sesión del 14 de marzo: ```python pf = pq.ParquetFile("ticks/fecha=2025-03-14/parte-0.parquet") md = pf.metadata print(md.num_rows, "filas,", md.num_row_groups, "grupos de filas,", f"pie de {md.serialized_size / 1e3:.1f} KB") filas_rg = [] for i in range(md.num_row_groups): rg = md.row_group(i) s_sim, s_ts = rg.column(1).statistics, rg.column(0).statistics filas_rg.append({ "grupo": i, "filas": rg.num_rows, "simbolo_min": s_sim.min, "simbolo_max": s_sim.max, "ts_min": s_ts.min.strftime("%H:%M:%S"), "ts_max": s_ts.max.strftime("%H:%M:%S"), "MB": round(rg.total_byte_size / 1e6, 1)}) pd.DataFrame(filas_rg) ``` ```output 346529 filas, 6 grupos de filas, pie de 4.5 KB grupo filas simbolo_min simbolo_max ts_min ts_max MB 0 0 65536 BAGO CELE 13:30:00 19:59:59 0.9 1 1 65536 CELE DORU 13:30:00 19:59:59 1.0 2 2 65536 DORU GUBO 13:30:00 19:59:58 0.9 3 3 65536 GUBO LAVI 13:30:00 19:59:59 0.9 4 4 65536 LAVI MENO 13:30:00 19:59:59 0.9 5 5 18849 MENO MONA 13:30:01 19:59:58 0.3 ``` Seis grupos de 65.536 filas (el último, con el resto) y un pie de 4,5 KB que los describe. Como el fichero está ordenado por símbolo, cada grupo cubre un tramo alfabético estrecho (`BAGO` a `CELE`, `CELE` a `DORU`…) y la sesión entera en la columna `ts`: de las 13:30 a las 20:00 UTC, la sesión de Nueva York en horario de verano, que empezó el 9 de marzo. Esas estadísticas son las que permiten el *predicate pushdown*: un filtro `simbolo = 'BAGO'` solo puede cumplirse en el grupo 0, porque en los demás el mínimo ya es posterior en el alfabeto. La columna `MB` es el tamaño sin comprimir de cada grupo según el pie. ## Predicate pushdown, medido en bytes Que un lector «puede» saltarse datos se lee en todas partes. Aquí se mide. La celda siguiente envuelve el disco local en un sistema de ficheros de `fsspec` que suma los bytes que se leen de verdad y lo pasa a `pyarrow.dataset`, el lector de conjuntos de ficheros de PyArrow. Antes de medir, lee los 60 pies una vez (un lector los guarda en memoria tras la primera consulta): así cada consulta cuenta solo los datos. ```python import datetime as dt import pyarrow.compute as pc import pyarrow.dataset as ds from fsspec.implementations.local import LocalFileSystem from pyarrow.fs import FSSpecHandler, PyFileSystem class Contador(LocalFileSystem): """Disco local que anota cuántos bytes se leen.""" bytes_leidos = 0 lecturas = 0 def _open(self, path, mode="rb", **kw): f = super()._open(path, mode, **kw) leer = f.read def read(n=-1): datos = leer(n) self.bytes_leidos += len(datos) self.lecturas += 1 return datos f.read = read return f disco_contado = Contador() lago = ds.dataset( str(Path("ticks").resolve()), format="parquet", filesystem=PyFileSystem(FSSpecHandler(disco_contado)), partitioning=ds.partitioning( pa.schema([("fecha", pa.date32())]), flavor="hive")) for frag in lago.get_fragments(): # leer los 60 pies una vez frag.ensure_complete_metadata() print(f"pies: {disco_contado.lecturas} lecturas," f" {disco_contado.bytes_leidos / 1e6:.1f} MB") def medir(nombre, columnas=None, filtro=None): disco_contado.bytes_leidos = disco_contado.lecturas = 0 t = lago.to_table(columns=columnas, filter=filtro) return {"consulta": nombre, "filas": t.num_rows, "MB_leidos": round(disco_contado.bytes_leidos / 1e6, 2), "pct_lago": round(100 * disco_contado.bytes_leidos / disco, 2)} f14 = pc.field("fecha") == dt.date(2025, 3, 14) bago = pc.field("simbolo") == "BAGO" consultas = pd.DataFrame([ medir("todo"), medir("solo precio", ["precio"]), medir("una sesión", None, f14), medir("BAGO en una sesión", ["ts", "precio"], f14 & bago), medir("BAGO en todo el lago", ["ts", "precio"], bago), medir("precio > 1000", ["precio"], pc.field("precio") > 1000), ]) consultas ``` ```output pies: 61 lecturas, 4.0 MB consulta filas MB_leidos pct_lago 0 todo 20783757 219.23 99.88 1 solo precio 20783757 35.20 16.04 2 una sesión 346529 3.62 1.65 3 BAGO en una sesión 1568 0.59 0.27 4 BAGO en todo el lago 90225 35.57 16.21 5 precio > 1000 0 0.00 0.00 ``` La figura de la cabecera dibuja la última columna. Los pies se leen con 61 lecturas (una por fichero y otra más, la del primer fichero, que `ds.dataset` abre para deducir el esquema) y suman 4,0 MB. Después, cada consulta: - **Todo**: 219,23 MB, el 99,88 % del lago. Lo que falta hasta el 100 % son los pies, ya leídos. - **Solo precio**: un 16 %. Es la **proyección**: las demás columnas ni se tocan. El precio pesa más que su parte proporcional porque es la columna menos repetitiva; los símbolos y los centros, con pocos valores distintos, se comprimen con diccionario y casi no ocupan. - **Una sesión**: un 1,65 %, un fichero de 60. Es la **poda de particiones**: el valor de `fecha` está en la ruta, y el lector descarta 59 ficheros sin abrirlos. - **BAGO en una sesión**: 0,59 MB, un 0,27 % del lago. Partición más **estadísticas**: dentro del fichero, solo el grupo 0 puede contener `BAGO`, y de ese grupo solo se leen tres columnas (`ts` y `precio`, que se piden, y `simbolo`, que hace falta para filtrar fila a fila dentro del grupo). - **BAGO en todo el lago**: un 16,2 %. Sin filtro de fecha se abren los 60 ficheros, pero en cada uno solo el primer grupo. Lee casi lo mismo que «solo precio» para devolver 230 veces menos filas. - **Precio > 1000**: cero bytes. El máximo de `precio` en el pie de cada grupo es menor que 1.000, así que el lector sabe que ninguna fila cumple el filtro sin leer ni una página. Las estadísticas solo sirven si los datos están **ordenados** por la columna del filtro. Lo comprueba la celda siguiente: reescribe la sesión del 14 de marzo ordenada por hora y cuenta, con `split_by_row_group`, cuántos grupos sobreviven a dos filtros distintos en cada versión. ```python un_dia = pq.read_table("ticks/fecha=2025-03-14/parte-0.parquet") pq.write_table(un_dia.sort_by("ts"), "por_hora.parquet", compression="zstd", row_group_size=65_536) def grupos_leidos(ruta, filtro): frag = next(ds.dataset(ruta).get_fragments()) vivos = frag.split_by_row_group(filtro) return sum(len(f.row_groups) for f in vivos) franja = ((pc.field("ts") >= pd.Timestamp("2025-03-14 15:00Z")) & (pc.field("ts") < pd.Timestamp("2025-03-14 15:05Z"))) orden = pd.DataFrame({ "simbolo = BAGO": [ grupos_leidos("ticks/fecha=2025-03-14", bago), grupos_leidos("por_hora.parquet", bago)], "5 minutos": [ grupos_leidos("ticks/fecha=2025-03-14", franja), grupos_leidos("por_hora.parquet", franja)], }, index=["ordenado por símbolo", "ordenado por hora"]) orden.columns.name = "grupos leídos" orden ``` ```output grupos leídos simbolo = BAGO 5 minutos ordenado por símbolo 1 6 ordenado por hora 6 1 ``` *Fig. — Grupos de filas que lee cada filtro en la sesión del 14 de marzo (346 529 ticks, 6 grupos) según el orden con el que se escribió el fichero.* Cuatro filas de seis casillas, una por grupo de filas. Ordenado por símbolo, el filtro por símbolo lee 1 de 6 (destacado) y la franja de 5 minutos, 6 de 6. Ordenado por hora, el filtro por símbolo lee 6 de 6 y la franja de 5 minutos, 1 de 6. Simétrico: ordenado por símbolo, el filtro por símbolo lee 1 grupo de 6 y una franja de 5 minutos los lee todos (cada grupo tiene ticks de toda la sesión); ordenado por hora, al revés. No hay un orden que sirva para todo, y la elección depende de las consultas habituales. Para un backtest que recorre un activo cada vez, símbolo y hora; para reconstruir el mercado entero en un instante (un libro consolidado, un corte transversal), hora. Algunos lagos guardan las dos copias. Otra opción es particionar por una columna y ordenar por la otra, como hace esta lección: partición por fecha y orden por símbolo, de modo que un filtro por fecha y símbolo lee un solo grupo de un solo fichero. ## Arrow: los mismos bytes para todos Leer poco del disco no sirve de nada si después cada herramienta copia los datos a su propio formato. **Apache Arrow** define cómo se colocan las columnas en memoria (búferes contiguos de valores, un mapa de bits de validez para los nulos y desplazamientos para los textos) de forma independiente del lenguaje (*language-agnostic*)[^2]. Polars, DuckDB, pandas con `ArrowDtype` y PyArrow entienden ese formato y pueden trabajar sobre los mismos búferes sin copiarlos (*zero-copy*). La celda lo comprueba con la sesión del 14 de marzo: mide cuánta memoria nueva reserva el asignador de Arrow en cada conversión y si el búfer de precios es el mismo. ```python import duckdb import polars as pl t = un_dia.combine_chunks() # un trozo por columna direccion = t.column("precio").chunk(0).buffers()[1].address def mb_nuevos(funcion): antes = pa.total_allocated_bytes() resultado = funcion() return resultado, (pa.total_allocated_bytes() - antes) / 1e6 df_pl, mb_pl = mb_nuevos(lambda: pl.from_arrow(t)) df_ar, mb_ar = mb_nuevos( lambda: t.to_pandas(types_mapper=pd.ArrowDtype)) df_np, mb_np = mb_nuevos(lambda: t.to_pandas()) v = t.column("precio").chunk(0).to_numpy(zero_copy_only=True) print(f"Polars: {mb_pl:5.1f} MB nuevos,", "mismo búfer:", df_pl["precio"].to_arrow().buffers()[1].address == direccion) print(f"pandas ArrowDtype: {mb_ar:5.1f} MB nuevos") print(f"pandas NumPy: {mb_np:5.1f} MB nuevos") print("NumPy sin copia:", v.ctypes.data == direccion, "| escribible:", v.flags.writeable) duckdb.sql(""" SELECT count(*) AS ticks, round(sum(precio * tamano) / sum(tamano), 4) AS vwap FROM t WHERE simbolo = 'BAGO' """) ``` ```output Polars: 0.0 MB nuevos, mismo búfer: True pandas ArrowDtype: 0.0 MB nuevos pandas NumPy: 12.8 MB nuevos NumPy sin copia: True | escribible: False ┌───────┬─────────┐ │ ticks │ vwap │ │ int64 │ double │ ├───────┼─────────┤ │ 1568 │ 23.9505 │ └───────┴─────────┘ ``` Lo que dice cada línea: - **Polars** no reserva nada: su `DataFrame` apunta al mismo búfer de precios que la tabla de Arrow (la comparación de direcciones da `True`; las direcciones no se imprimen porque cambian en cada ejecución). - **pandas con `ArrowDtype`** tampoco: cada columna es una envoltura sobre el array de Arrow. - **pandas con NumPy** (lo que devuelve `to_pandas()` por defecto) reserva 12,8 MB en el asignador de Arrow: la copia de las columnas al formato que espera NumPy. - **NumPy** puede ver el búfer de precios sin copiarlo, pero **de solo lectura** (`escribible: False`): escribir en él cambiaría datos que las demás herramientas creen inmutables. - **DuckDB** consulta la variable `t` de Python directamente (el *replacement scan* de la lección 0.5.5) y lee sus búferes en el sitio: 1.568 ticks de BAGO, los mismos que leyó `medir`. El *zero-copy* tiene condiciones. La más común en datos de mercado son los **nulos**. Arrow los marca en un mapa de bits aparte y deja un valor cualquiera en el hueco; NumPy no tiene nulos para los `float64`, así que para dárselos hay que copiar y escribir `NaN` en esos huecos. Si se exige que no haya copia, PyArrow se niega: ```python raises con_nulos = pa.array([101.5, None, 101.7]) con_nulos.to_numpy(zero_copy_only=True) ``` ```output pyarrow.lib.ArrowInvalid: Needed to copy 1 chunks with 1 nulls, but zero_copy_only was True ``` Las otras condiciones: que la columna esté en un solo trozo (por eso la celda usa `combine_chunks()`; una tabla leída de varios ficheros tiene varios) y que el tipo tenga la misma representación en las dos herramientas. Los textos, por ejemplo, son un búfer de bytes con desplazamientos en Arrow y objetos de Python en una columna `object` de pandas: pasar de uno a otro siempre copia. ## Polars lazy: el plan antes que los datos La lección 0.5.4 presentó `scan_parquet` y `explain()` con un fichero. Con un lago particionado, el plan enseña además qué ficheros se abren. La consulta pide el número de ticks y el volumen de BAGO por centro de negociación en la semana del 3 al 7 de marzo. Primero el plan tal como se escribió (`optimized=False`) y después el que ejecuta Polars: ```python from datetime import date ticks = pl.scan_parquet("ticks/**/*.parquet", hive_partitioning=True) semana = ( ticks .filter(pl.col("fecha").is_between(date(2025, 3, 3), date(2025, 3, 7)) & (pl.col("simbolo") == "BAGO")) .group_by("centro") .agg(ticks=pl.len(), volumen=pl.col("tamano").cast(pl.Int64).sum()) .sort("centro") ) print(semana.explain(optimized=False)) print() print(semana.explain()) ``` ```output SORT BY [col("centro")] AGGREGATE[maintain_order: false] [len().alias("ticks"), col("tamano").cast(Int64).sum().alias("volumen")] BY [col("centro")] FROM FILTER col("simbolo") == "BAGO" FROM FILTER col("fecha").is_between([2025-03-03, 2025-03-07]) FROM Parquet SCAN [ticks/fecha=2025-01-02/parte-0.parquet, ... 59 other sources] PROJECT */7 COLUMNS ESTIMATED ROWS: 20780871 SORT BY [col("centro")] AGGREGATE[maintain_order: false] [len().alias("ticks"), col("tamano").cast(Int64).sum().alias("volumen")] BY [col("centro")] FROM simple π 2/2 ["centro", "tamano"] Parquet SCAN [ticks/fecha=2025-03-03/parte-0.parquet, ... 4 other sources] PROJECT 4/7 COLUMNS SELECTION: (col("fecha").is_between([2025-03-03, 2025-03-07]) & (col("simbolo") == "BAGO")) ESTIMATED ROWS: 1731740 ``` Los planes se leen de abajo arriba. Sin optimizar, Polars leería los 60 ficheros (`Parquet SCAN [… 59 other sources]`) con todas las columnas (`PROJECT */7 COLUMNS`: las seis del fichero más `fecha`, que sale de la ruta) y filtraría después, en dos pasos. El plan optimizado cambia tres cosas: - **Poda de particiones**: `Parquet SCAN [ticks/fecha=2025-03-03/…, … 4 other sources]`. El filtro de fechas se resuelve con los nombres de las carpetas: cinco ficheros de 60. - **Proyección**: `PROJECT 4/7 COLUMNS`. Lee `simbolo` y `fecha` para filtrar, y `centro` y `tamano` para agregar. - **Filtro empujado al lector**: `SELECTION: …` dentro del `Parquet SCAN`. El lector aplica el filtro mientras lee, y puede usar las estadísticas de los grupos para saltarse los que no contienen `BAGO`. `ESTIMATED ROWS` es una estimación a partir del tamaño de los ficheros, no un recuento. El resultado: ```python semana.collect() ``` ```output shape: (5, 3) ┌────────┬───────┬─────────┐ │ centro ┆ ticks ┆ volumen │ │ --- ┆ --- ┆ --- │ │ str ┆ u32 ┆ i64 │ ╞════════╪═══════╪═════════╡ │ ARCX ┆ 1562 ┆ 77312 │ │ BATS ┆ 1106 ┆ 56094 │ │ EDGX ┆ 1147 ┆ 57924 │ │ OSCURO ┆ 1560 ┆ 79308 │ │ XNYS ┆ 2224 ┆ 114732 │ └────────┴───────┴─────────┘ ``` BAGO tuvo 7.599 ticks esa semana, repartidos entre los cinco centros en proporciones parecidas a las del generador (un 30 % en `XNYS`, un 20 % en `ARCX` y en el centro oscuro, un 15 % en `BATS` y en `EDGX`): la consulta lee lo que tiene que leer y nada más. ### El motor de streaming `collect()` ejecuta el plan con el motor por defecto. Con `engine="streaming"`, Polars ejecuta el mismo plan por bloques de filas: lee un trozo de cada fichero, lo filtra y lo agrega, y libera la memoria antes de leer el siguiente. La tabla completa no llega a estar nunca en memoria, solo el estado de la agregación[^3]. La celda calcula barras diarias de los 200 símbolos (apertura, máximo, mínimo, cierre, volumen, VWAP y número de ticks) a partir de los 21 millones de ticks: ```python diarias = ( ticks .with_columns(importe=pl.col("precio") * pl.col("tamano")) .group_by("simbolo", "fecha") .agg(apertura=pl.col("precio").first(), maximo=pl.col("precio").max(), minimo=pl.col("precio").min(), cierre=pl.col("precio").last(), volumen=pl.col("tamano").cast(pl.Int64).sum(), importe=pl.col("importe").sum(), ticks=pl.len()) .with_columns(vwap=(pl.col("importe") / pl.col("volumen")) .round(4)) .drop("importe") .sort("simbolo", "fecha") .collect(engine="streaming") ) print(diarias.shape, diarias["ticks"].sum()) diarias.filter(pl.col("simbolo") == "BAGO").select( "fecha", "apertura", "cierre", "volumen", "ticks", "vwap" ).head(3) ``` ```output (12000, 9) 20783757 shape: (3, 6) ┌────────────┬──────────┬────────┬─────────┬───────┬─────────┐ │ fecha ┆ apertura ┆ cierre ┆ volumen ┆ ticks ┆ vwap │ │ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- │ │ date ┆ f64 ┆ f64 ┆ i64 ┆ u32 ┆ f64 │ ╞════════════╪══════════╪════════╪═════════╪═══════╪═════════╡ │ 2025-01-02 ┆ 27.23 ┆ 27.54 ┆ 80546 ┆ 1546 ┆ 27.5508 │ │ 2025-01-03 ┆ 27.53 ┆ 27.39 ┆ 72071 ┆ 1481 ┆ 27.254 │ │ 2025-01-06 ┆ 27.38 ┆ 26.85 ┆ 75824 ┆ 1497 ┆ 27.0718 │ └────────────┴──────────┴────────┴─────────┴───────┴─────────┘ ``` `first()` y `last()` respetan el orden de lectura, que aquí es el de los ficheros (por símbolo y hora); en un lago sin ese orden habría que usar `sort_by("ts")` dentro de la agregación. Las 12.000 barras (200 símbolos × 60 sesiones) suman los 20.783.757 ticks: no se ha perdido ninguno por el camino. Lo importante no se ve en la salida: la memoria. Medido en la máquina en la que se escribió la lección (16 núcleos, cada variante en un proceso aparte, pico de memoria residente del proceso), las mismas barras diarias: | Variante | Pico de memoria | |---|---| | pandas: `read_parquet` del lago y `groupby` | ≈ 3,0 GB | | Polars: `read_parquet` del lago y `group_by` | ≈ 2,5 GB | | Polars lazy con el motor en memoria | ≈ 2,0 GB | | Polars lazy con `engine="streaming"` | ≈ 0,7 GB | La lectura completa paga los 783 MB de Arrow más las copias y las estructuras intermedias. La consulta lazy en memoria ya ahorra (lee solo las columnas que hacen falta), y el streaming baja a una cuarta parte de la lectura completa. Con un lago diez veces mayor, las tres primeras crecerían de forma parecida; la de streaming, mucho menos, porque solo guarda 12.000 grupos. ## DuckDB fuera de memoria DuckDB aborda el mismo problema desde el otro lado: un gestor de memoria con un **límite** (`memory_limit`) y una carpeta temporal (`temp_directory`) en la que los operadores que lo admiten vuelcan lo que no cabe (*spilling*): las agregaciones, las uniones, las ordenaciones y las funciones de ventana[^4]. La prueba: un rendimiento logarítmico tick a tick con `lag()` sobre los 21 millones de filas, que obliga a ordenar cada símbolo por hora, con un límite de 300 MB. ```python con = duckdb.connect() con.sql("SET memory_limit = '300MB'") con.sql("SET temp_directory = 'duckdb_tmp'") con.sql("SET threads = 4") print(con.sql("SELECT current_setting('memory_limit')").fetchone()) RV = """ WITH r AS ( SELECT simbolo, ln(precio / lag(precio) OVER ( PARTITION BY simbolo ORDER BY ts)) AS ret FROM read_parquet('ticks/**/*.parquet', hive_partitioning = true) ) SELECT count(*) AS ticks, count(ret) AS rendimientos, round(sum(ret * ret), 3) AS suma_r2 FROM r """ con.sql(RV) ``` ```output ('286.1 MiB',) ┌──────────┬──────────────┬─────────┐ │ ticks │ rendimientos │ suma_r2 │ │ int64 │ int64 │ double │ ├──────────┼──────────────┼─────────┤ │ 20783757 │ 20783557 │ 5.427 │ └──────────┴──────────────┴─────────┘ ``` `300MB` se guarda como `286.1 MiB` (DuckDB muestra mebibytes, de 1.048.576 bytes). La consulta termina: 20.783.757 ticks y 20.783.557 rendimientos, 200 menos, porque el primer tick de cada símbolo no tiene anterior y `lag()` da nulo. `suma_r2` es la suma de los rendimientos al cuadrado de los 200 símbolos en las 60 sesiones, incluidos los saltos entre el cierre de un día y la apertura del siguiente: 5,427, unas 0,027 por símbolo. Repartido entre 60 sesiones, √(0,027 / 60) ≈ 2,1 % diario, la volatilidad con la que se generaron los precios (2 %) más el ruido del redondeo al céntimo, que en las acciones baratas añade varianza tick a tick. Es la respuesta conocida, y sirve para comprobar que la ventana ha emparejado cada tick con el anterior de su símbolo. Lo que no se ve: medido en la máquina de la lección, DuckDB escribió unos 500 MB en `duckdb_tmp` durante la consulta y el proceso no pasó de unos 530 MB de memoria residente. Sin límite, la misma consulta usó ≈ 1,3 GB y no escribió nada en disco. Los ficheros temporales se borran al terminar la consulta. Con menos memoria, el número de hilos importa: cada hilo necesita sus propios búferes. Con 100 MB y cuatro hilos, la consulta falla; con un hilo, termina (más despacio): ```python con.sql("SET memory_limit = '100MB'") try: con.sql(RV).fetchall() except duckdb.OutOfMemoryException as e: lineas = str(e).splitlines() print(lineas[0].split(":")[0]) print("\n".join(l for l in lineas if l.startswith("*"))) con.sql("SET threads = 1") print(con.sql(RV).fetchall()) ``` ```output Out of Memory Error * Reducing the number of threads (SET threads=X) * Disabling insertion-order preservation (SET preserve_insertion_order=false) * Increasing the memory limit (SET memory_limit='...GB') [(20783757, 20783557, 5.427)] ``` El mensaje de error trae tres remedios, en este orden: menos hilos, `preserve_insertion_order = false` (DuckDB deja de garantizar que las filas salgan en el orden en que entraron cuando la consulta no tiene `ORDER BY`, lo que ahorra memoria en importaciones y exportaciones grandes) y más memoria. Con un hilo, el resultado es idéntico al de antes; medido en la máquina de la lección, tardó unos ocho segundos, frente a unos tres con cuatro hilos y 300 MB. La documentación de DuckDB precisa los límites del volcado: todos los operadores bloqueantes (agrupar, unir, ordenar y ventanas) pueden trabajar con más datos que memoria, pero algunas funciones de agregación, como `string_agg()`, no vuelcan a disco, y una consulta con varios operadores bloqueantes puede quedarse sin memoria aunque cada uno por separado no lo haría[^4]. ## Qué usar y cuándo | Necesidad | Herramienta | Por qué | |---|---|---| | Leer pocas columnas o pocas fechas | Parquet particionado | Proyección, poda de particiones y estadísticas | | Filtrar por un campo dentro de cada fichero | Orden de escritura por ese campo | Las estadísticas solo sirven con datos ordenados | | Pasar datos entre librerías | Arrow | Mismos búferes, sin copia si no hay nulos ni varios trozos | | Transformaciones en Python con resultado pequeño | Polars lazy, `engine="streaming"` | Plan optimizado y lectura por bloques | | SQL, uniones, ventanas grandes | DuckDB con `memory_limit` | Vuelca a disco lo que no cabe | | Varias máquinas, datos de terabytes | Dask, Spark (lección 6.6.3) | Cuando una máquina no basta, no antes | La regla que resume la lección: antes de comprar memoria o montar un clúster, mirar el plan y contar los bytes. En el lago de la lección, la diferencia entre la peor y la mejor forma de pedir los ticks de BAGO de un día es de 370 veces (219 MB frente a 0,59 MB), sin cambiar nada del hardware. [^1]: Apache Parquet, «Concepts», «File Format» y «Configurations» (que recomienda grupos de 512 MB a 1 GB): , consultado el 23 de octubre de 2026. [^2]: Apache Arrow, «Arrow Columnar Format»: , consultado el 23 de octubre de 2026. [^3]: Polars, guía de usuario, «Streaming»: , consultado el 23 de octubre de 2026. [^4]: DuckDB 1.5, «Tuning Workloads»: , consultado el 23 de octubre de 2026. [^5]: DuckDB 1.5, «Configuration»: , consultado el 23 de octubre de 2026. ## Ejercicios 1. Sin filtrar por `fecha`, pide al lago (`medir`) las columnas `ts` y `precio` de la franja de 15:00 a 15:05 UTC del 14 de marzo (`franja`). ¿Cuántos bytes lee y por qué no son los 220 MB del lago? **Solución.** ```python pd.DataFrame([medir("franja sin fecha", ["ts", "precio"], franja), medir("franja con fecha", ["ts", "precio"], franja & f14)]) ``` ```output consulta filas MB_leidos pct_lago 0 franja sin fecha 4481 3.08 1.4 1 franja con fecha 4481 3.08 1.4 ``` Lee 3,08 MB, un 1,4 % del lago, con o sin el filtro de fecha. Los otros 59 ficheros se descartan igualmente, pero por otra vía: el mínimo y el máximo de `ts` de cada grupo de filas de otra sesión quedan fuera de la franja, así que las estadísticas los eliminan sin necesidad de la partición. Dentro del 14 de marzo, en cambio, se leen los seis grupos (ordenados por símbolo, todos cubren la sesión entera) y solo las columnas `ts` y `precio`. Si las consultas por franja horaria fueran las habituales, compensaría ordenar por hora. 2. Calcula las mismas barras diarias con DuckDB (con `arg_min` y `arg_max` para la apertura y el cierre) y comprueba que coinciden con las de Polars. **Solución.** ```python con.sql("SET threads = 4") con.sql("SET memory_limit = '300MB'") d_duck = con.sql(""" SELECT simbolo, fecha, arg_min(precio, ts) AS apertura, max(precio) AS maximo, min(precio) AS minimo, arg_max(precio, ts) AS cierre, sum(tamano) AS volumen, round(sum(precio * tamano) / sum(tamano), 4) AS vwap, count(*) AS ticks FROM read_parquet('ticks/**/*.parquet', hive_partitioning = true) GROUP BY simbolo, fecha ORDER BY simbolo, fecha """).pl() cols = ["apertura", "maximo", "minimo", "cierre", "volumen", "vwap", "ticks"] iguales = {c: bool((d_duck[c] == diarias[c].cast(d_duck[c].dtype)) .all()) for c in cols} print(d_duck.shape, iguales) ``` ```output (12000, 9) {'apertura': True, 'maximo': True, 'minimo': True, 'cierre': True, 'volumen': True, 'vwap': True, 'ticks': True} ``` Las 12.000 barras coinciden en los siete campos. `arg_min(precio, ts)` devuelve el precio del tick con la `ts` más baja: es la apertura sin depender del orden de lectura, a diferencia de `first()` en Polars, que confía en que los ficheros estén ordenados. Comparar dos motores independientes sobre los mismos datos es una forma barata de validar un cálculo antes de usarlo en un backtest. La última celda borra los datos temporales: la carpeta de trabajo queda como estaba. ```python import shutil con.close() shutil.rmtree("ticks") shutil.rmtree("duckdb_tmp", ignore_errors=True) Path("por_hora.parquet").unlink() print(sorted(p.name for p in Path(".").glob("ticks*"))) ``` ```output [] ``` 3. Un compañero quiere guardar las actualizaciones del libro L2 (lección 1.1.3) de 500 acciones, unos 2.000 millones de filas al mes, para dos usos: backtests de un activo cada vez y reconstrucciones del mercado entero en momentos concretos. Propón la partición, el orden y el tamaño de los grupos de filas. **Solución.** Partición por fecha (un día por carpeta) y, dentro de cada día, varios ficheros por grupos de símbolos (por ejemplo, por la inicial o por un hash del símbolo en 8-16 cubos), para que cada fichero quede en cientos de megabytes y no en decenas de gigabytes. Dentro de cada fichero, orden por símbolo y hora: un backtest de un activo lee un fichero por día y, en él, los grupos de filas de ese símbolo. Para las reconstrucciones del mercado en un instante, ese orden obliga a leer todos los ficheros del día, aunque las estadísticas de `ts` no ayudan dentro de cada uno (todos los grupos cubren la sesión). Si esas consultas son frecuentes, se mantiene una segunda copia ordenada por hora (es lo que hacen muchos lagos: el almacenamiento es barato frente al tiempo de lectura) o se escriben grupos de filas por símbolo y bloque horario. Grupos de filas de decenas a cientos de megabytes: grandes para leer en secuencia, suficientes para que las estadísticas descarten. Y antes de decidir, medir con las consultas reales como en esta lección. ## Autoevaluación 1. En el lago de la lección, `simbolo = 'BAGO'` en una sesión lee 1 grupo de filas de 6. La misma sesión escrita ordenada por hora, ¿cuántos grupos lee para ese filtro? - A. 1, porque las estadísticas no dependen del orden - B. Ninguno, porque el filtro se resuelve en el pie - C. Los 6: ordenado por hora, cada grupo contiene ticks de casi todos los símbolos y su mínimo y máximo de `simbolo` abarcan todo el alfabeto - D. 3, la mitad *Respuesta C: Los 6: ordenado por hora, cada grupo contiene ticks de casi todos los símbolos y su mínimo y máximo de `simbolo` abarcan todo el alfabeto.* Las estadísticas son el mínimo y el máximo de cada columna en cada grupo. Solo descartan grupos si el filtro cae fuera de ese intervalo, y eso exige que los datos estén ordenados (o agrupados) por la columna del filtro. Con el orden por hora pasa lo contrario: una franja de 5 minutos lee 1 grupo de 6. 2. La consulta `precio > 1000` sobre los 21 millones de ticks lee 0 bytes de datos. ¿Por qué? - A. Porque el máximo de `precio` en el pie de cada grupo de filas es menor que 1.000: el lector sabe que no hay filas que cumplan el filtro sin abrir ninguna página - B. Porque PyArrow tiene la consulta en caché - C. Porque la columna `precio` está comprimida con diccionario - D. Porque es un error del contador de bytes *Respuesta A: Porque el máximo de `precio` en el pie de cada grupo de filas es menor que 1.000: el lector sabe que no hay filas que cumplan el filtro sin abrir ninguna página.* Los pies se leyeron una vez antes de medir (4,0 MB en 61 lecturas). Con sus estadísticas, el lector descarta los 360 grupos de filas. Es el caso extremo del predicate pushdown. 3. `pa.array([101.5, None, 101.7]).to_numpy(zero_copy_only=True)` lanza `ArrowInvalid`. ¿Qué lo impide? - A. Que NumPy no admite números con decimales - B. Que el array es demasiado pequeño - C. Que Arrow guarda los arrays de forma no contigua - D. El nulo: Arrow lo marca en un mapa de bits aparte y NumPy no tiene nulos en `float64`, así que hay que copiar y escribir `NaN` en el hueco *Respuesta D: El nulo: Arrow lo marca en un mapa de bits aparte y NumPy no tiene nulos en `float64`, así que hay que copiar y escribir `NaN` en el hueco.* Sin nulos y en un solo trozo, NumPy puede ver el búfer de Arrow tal cual (de solo lectura). Con nulos, varios trozos o textos, la conversión copia. Polars y pandas con `ArrowDtype` conservan el mapa de validez y no copian. 4. Barras diarias de 21 millones de ticks: ≈ 3,0 GB de pico con pandas y ≈ 0,7 GB con Polars en streaming. Con barras de 1 minuto, el streaming sube a ≈ 2,0 GB. ¿Por qué? - A. Porque el motor de streaming no funciona con marcas de tiempo - B. Porque el streaming acota la memoria de la entrada, no la del estado: con 4,2 millones de grupos abiertos, la tabla de agregación ocupa mucho - C. Porque `truncate("1m")` obliga a leer el lago dos veces - D. Porque Polars cambia al motor de pandas *Respuesta B: Porque el streaming acota la memoria de la entrada, no la del estado: con 4,2 millones de grupos abiertos, la tabla de agregación ocupa mucho.* Los datos se leen por bloques, pero cada grupo (`simbolo`, `minuto`) necesita su estado hasta el final. Con 12.000 grupos (barras diarias) el estado es pequeño. Para resultados grandes, se escribe con `sink_parquet` o se parte el cálculo, por ejemplo un día cada vez. 5. Con `memory_limit = '100MB'` y cuatro hilos, la ventana de DuckDB sobre los 21 millones de ticks falla con `Out of Memory Error`; con un hilo, termina. ¿Qué explica la diferencia? - A. Que con un hilo DuckDB no usa `temp_directory` - B. Que con un hilo la consulta se calcula de forma aproximada - C. Que cada hilo necesita sus propios búferes en memoria: con menos hilos cabe el trabajo en curso y lo demás se vuelca a disco - D. Que con cuatro hilos el límite real es de 25 MB *Respuesta C: Que cada hilo necesita sus propios búferes en memoria: con menos hilos cabe el trabajo en curso y lo demás se vuelca a disco.* El resultado con un hilo es idéntico (20.783.557 rendimientos, suma de cuadrados 5,427), solo más lento. El propio mensaje de error propone reducir hilos, desactivar `preserve_insertion_order` o subir el límite. ## Ideas clave - Un Parquet tiene grupos de filas y un pie con el mínimo y el máximo de cada columna en cada grupo: el lector decide con él qué leer. En el lago de la lección, los ticks de BAGO en un día cuestan 0,59 MB de 220 (un 0,27 %) y `precio > 1000`, cero bytes. - Proyección, poda de particiones y estadísticas se suman, pero las estadísticas solo sirven si los datos están ordenados por la columna del filtro: ordenado por símbolo, un símbolo lee 1 grupo de 6 y una franja de 5 minutos los lee todos; ordenado por hora, al revés. - Arrow es el formato en memoria común: Polars, pandas con `ArrowDtype`, NumPy (de solo lectura) y DuckDB usan los mismos búferes sin copiar; los nulos, los varios trozos y los textos hacia `object` obligan a copiar. - El plan optimizado de Polars enseña la poda de particiones (5 de 60 ficheros), las columnas leídas y el filtro empujado al lector; `engine="streaming"` bajó el pico de memoria de ≈ 3,0 GB (pandas) a ≈ 0,7 GB, pero no acota el estado de una agregación con millones de grupos. - DuckDB vuelca a `temp_directory` lo que no cabe en `memory_limit` en agregaciones, uniones, ordenaciones y ventanas: una ventana sobre 21 millones de filas termina con 300 MB, y con 100 MB solo con un hilo. ## Recursos ### Research - [Big Data Management. 1 - Introduccion al Big Data para trading con ArticDB](https://pythonparatrading.com/big-data-para-trading/) · Research-0019 · 16 min ### Documentación - [Apache Parquet: conceptos, formato de fichero y configuración (tamaño de los grupos de filas)](https://parquet.apache.org/docs/file-format/configurations/) · parquet.apache.org - [Apache Arrow: formato columnar en memoria](https://arrow.apache.org/docs/format/Columnar.html) · arrow.apache.org - [PyArrow: lectura de conjuntos de datos con filtros (pyarrow.dataset)](https://arrow.apache.org/docs/python/dataset.html) · arrow.apache.org - [Polars: el motor de streaming](https://docs.pola.rs/user-guide/concepts/streaming/) · docs.pola.rs - [DuckDB 1.5: ajuste de cargas de trabajo y procesamiento fuera de memoria](https://duckdb.org/docs/1.5/guides/performance/how_to_tune_workloads) · duckdb.org ### Datos - [Datos del nivel 5: tickers del panel de acciones (CSV)](https://pythonparatrading.com/curso/datos/nivel-5/panel5_activos.csv) · Descarga · panel5_activos.csv Lección anterior: [6.6.1 · Cuándo los datos son «big»: ticks, libros, opciones y universos](https://pythonparatrading.com/curso-python-trading/nivel-6/cuando-los-datos-son-big/) Lección siguiente: [6.6.3 · Dask, Spark y Ray: de una máquina a un clúster](https://pythonparatrading.com/curso-python-trading/nivel-6/dask-spark-y-ray/)