Parquet, Arrow, Polars en streaming y DuckDB fuera 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:

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/).

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")

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 grupo11Nota 1 · Apache Parquet, «Concepts», «File Format» y «Configurations» (que recomienda grupos de 512 MB a 1 GB): https://parquet.apache.org/docs/file-format/configurations/, consultado el 23 de octubre de 2026..

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:

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)

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.

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

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.

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

grupos leídos         simbolo = BAGO  5 minutos
ordenado por símbolo               1          6
ordenado por hora                  6          1

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)22Nota 2 · Apache Arrow, «Arrow Columnar Format»: https://arrow.apache.org/docs/format/Columnar.html, consultado el 23 de octubre de 2026.. 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.

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'
""")

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:

con_nulos = pa.array([101.5, None, 101.7])
con_nulos.to_numpy(zero_copy_only=True)

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:

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())

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:

semana.collect()

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ón33Nota 3 · Polars, guía de usuario, «Streaming»: https://docs.pola.rs/user-guide/concepts/streaming/, consultado el 23 de octubre de 2026.. 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:

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)

(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:

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 ventana44Nota 4 · DuckDB 1.5, «Tuning Workloads»: https://duckdb.org/docs/1.5/guides/performance/how_to_tune_workloads, consultado el 23 de octubre de 2026.. 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.

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)

('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):

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())

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⁴.

Qué usar y cuándo

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.

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?

pd.DataFrame([medir("franja sin fecha", ["ts", "precio"], franja),
              medir("franja con fecha", ["ts", "precio"],
                    franja & f14)])

           consulta  filas  MB_leidos  pct_lago
0  franja sin fecha   4481       3.08       1.4
1  franja con fecha   4481       3.08       1.4

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.

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)

(12000, 9) {'apertura': True, 'maximo': True, 'minimo': True, 'cierre': True, 'volumen': True, 'vwap': True, 'ticks': True}

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*")))

[]

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.