--- title: "6.6.2 Parquet, Arrow, Polars streaming and DuckDB" description: "21 million ticks on a laptop: predicate pushdown measured in bytes, zero-copy Arrow, lazy Polars in streaming and DuckDB with a memory limit." url: "https://pythonparatrading.com/en/python-trading-course/level-6/parquet-arrow-polars-streaming-duckdb/" lang: "en" image: "https://pythonparatrading.com/og/en/python-trading-course/level-6/parquet-arrow-polars-streaming-duckdb.png" --- # 6.6.2 · Parquet, Arrow, Polars streaming and out-of-core DuckDB > Before thinking about a cluster, it pays to squeeze a single machine. This lesson generates 21 million ticks in a partitioned Parquet lake, measures how many bytes each query reads depending on partitioning, sort order and statistics, checks which Arrow conversions copy data and which do not, runs Polars in streaming mode and forces DuckDB to work with 300 and with 100 MB of memory. - Course: [Level 6: Generative AI, LLMs, agents, big data and alternative data](https://pythonparatrading.com/en/python-trading-course/level-6/) - Module: 6.6 · Big data for trading - Duration: 70 min - Prerequisites: [0.5.4 · Polars: lazy versus eager API, expressions, performance and when to pick it over pandas](https://pythonparatrading.com/en/python-trading-course/level-0/polars-lazy-api-and-expressions/), [0.5.5 · DuckDB and essential SQL: SELECT, JOIN, GROUP BY and window functions on Parquet](https://pythonparatrading.com/en/python-trading-course/level-0/duckdb-sql-on-parquet/), [1.4.7 · Storage: CSV, Parquet, partitioning, DuckDB, TimescaleDB and ArcticDB](https://pythonparatrading.com/en/python-trading-course/level-1/storing-market-data-parquet-duckdb-timeseries/), [6.6.1 · When data becomes “big”: ticks, order books, options and universes](https://pythonparatrading.com/en/python-trading-course/level-6/when-financial-data-becomes-big-data/) ## By the end of this lesson - Read a Parquet footer (row groups, minimums and maximums) and measure how many bytes projection, partition pruning and statistics save. - Choose the write order and the partitioning according to the usual queries, knowing that statistics only help with sorted data. - Pass data between Arrow, Polars, pandas, NumPy and DuckDB without copies, and recognize when a copy is unavoidable (nulls, multiple chunks, strings). - Read the optimized Polars plan over a partitioned lake and run queries with the streaming engine, knowing which memory it bounds and which it does not. - Run larger-than-memory queries in DuckDB with `memory_limit`, `temp_directory` and threads, and interpret an out-of-memory error. ## When the table does not fit in memory Lesson 6.6.1 put numbers on the word “big”: the ticks of a stock universe, L2 and L3 books or option chains quickly reach hundreds of gigabytes. The usual reaction is to think about a cluster. It is almost always premature: a laptop with a good on-disk format and an engine that does not load what it does not need goes much further than it seems. This lesson walks through that single-machine stack, from the bottom up: ```mermaid flowchart TB P[Parquet on disk] --> A[Arrow in memory] A --> PL[Polars lazy] A --> D[DuckDB] A --> PD[pandas and NumPy] PL --> R[Small result] D --> R ``` - **Parquet** decides how much has to be read: columns, partitions and row groups with statistics (lesson 1.4.7 introduced the format; here we measure what each piece saves). - **Arrow** is the in-memory format shared by all the tools in the figure: passing data from one to another does not have to cost a copy. - **Polars lazy and its streaming engine** process the query in batches, without loading the whole table (the lazy API and `explain()` are in lesson 0.5.4). - **DuckDB** runs SQL with a memory limit and spills to disk whatever does not fit (basic SQL is in lesson 0.5.5). Measuring this needs a large dataset. The first cell generates synthetic ticks for 200 stocks from the level 5 panel (the tickers in `panel5_activos.csv`; the prices have no relation to those of the panel) over the 60 NYSE sessions of the first quarter of 2025. Each stock has its own average activity, from a few hundred to several thousand ticks per session, and its prices follow a random walk with a daily volatility of 2%. One Parquet file is written per session, with the rows sorted by symbol and time, in folders with the Hive layout (`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 per symbol 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)) # symbol and time ts = ts[orden] + (np.arange(N) - inicio) # no ties 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 for {len(tickers)} symbols" f" in {len(sesiones)} sessions") print(f"in Arrow (memory): {memoria / 1e6:,.0f} MB") print(f"in Parquet (disk): {disco / 1e6:,.0f} MB") ``` ```output 20,783,757 ticks for 200 symbols in 60 sessions in Arrow (memory): 783 MB in Parquet (disk): 220 MB ``` Almost 21 million ticks: 783 MB in memory in the Arrow format and 220 MB on disk with Parquet and zstd, 3.6 times less. It is a size chosen so that the lesson runs in a minute on a laptop (the cell takes about ten seconds on the machine it was written on); the mechanisms are the same with a hundred times more data, and the proportions measured below do not depend on the size. ## Inside the file: row groups and statistics A Parquet file is not an opaque block. It is divided into **row groups**; inside each group, each column is stored separately (*column chunk*) in compressed pages, and at the end of the file comes the **footer** with the schema, the position of each chunk and its **statistics**: the minimum, the maximum and the null count of each column in each group[^1]. ```mermaid flowchart TB F[Parquet file] --> G0[Row group 0] F --> G1[Row group 1] F --> PIE[Footer: schema and min/max] G0 --> C1[Column ts] G0 --> C2[Column simbolo] C2 --> PG[zstd pages] ``` The footer is the first thing any reader reads: with it, it decides which chunks it needs without opening the rest. The one for the March 14 session: ```python pf = pq.ParquetFile("ticks/fecha=2025-03-14/parte-0.parquet") md = pf.metadata print(md.num_rows, "rows,", md.num_row_groups, "row groups,", f"footer of {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 rows, 6 row groups, footer of 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 ``` Six groups of 65,536 rows (the last one with the remainder) and a 4.5 KB footer that describes them. Since the file is sorted by symbol, each group covers a narrow alphabetical range (`BAGO` to `CELE`, `CELE` to `DORU`…) and the whole session in the `ts` column: from 13:30 to 20:00 UTC, the New York session on daylight saving time, which began on March 9. Those statistics are what enable *predicate pushdown*: a filter `simbolo = 'BAGO'` can only be satisfied in group 0, because in the others the minimum is already later in the alphabet. The `MB` column is the uncompressed size of each group according to the footer. ## Predicate pushdown, measured in bytes That a reader “can” skip data is something you read everywhere. Here it is measured. The next cell wraps the local disk in an `fsspec` file system that adds up the bytes actually read and passes it to `pyarrow.dataset`, PyArrow's reader for sets of files. Before measuring, it reads the 60 footers once (a reader keeps them in memory after the first query): that way each query counts only the data. ```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): """Local disk that records how many bytes are read.""" 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(): # read the 60 footers once frag.ensure_complete_metadata() print(f"footers: {disco_contado.lecturas} reads," 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("everything"), medir("price only", ["precio"]), medir("one session", None, f14), medir("BAGO in one session", ["ts", "precio"], f14 & bago), medir("BAGO in the whole lake", ["ts", "precio"], bago), medir("precio > 1000", ["precio"], pc.field("precio") > 1000), ]) consultas ``` ```output footers: 61 reads, 4.0 MB consulta filas MB_leidos pct_lago 0 everything 20783757 219.23 99.88 1 price only 20783757 35.20 16.04 2 one session 346529 3.62 1.65 3 BAGO in one session 1568 0.59 0.27 4 BAGO in the whole lake 90225 35.57 16.21 5 precio > 1000 0 0.00 0.00 ``` The figure in the header plots the last column. The footers are read with 61 reads (one per file plus one more, for the first file, which `ds.dataset` opens to infer the schema) and add up to 4.0 MB. After that, each query: - **Everything**: 219.23 MB, 99.88% of the lake. What is missing up to 100% is the footers, already read. - **Price only**: 16%. This is **projection**: the other columns are not even touched. The price weighs more than its proportional share because it is the least repetitive column; symbols and venues, with few distinct values, are dictionary-encoded and take up almost nothing. - **One session**: 1.65%, one file out of 60. This is **partition pruning**: the value of `fecha` is in the path, and the reader discards 59 files without opening them. - **BAGO in one session**: 0.59 MB, 0.27% of the lake. Partition plus **statistics**: inside the file, only group 0 can contain `BAGO`, and from that group only three columns are read (`ts` and `precio`, which are requested, and `simbolo`, which is needed to filter row by row inside the group). - **BAGO in the whole lake**: 16.2%. Without a date filter all 60 files are opened, but in each one only the first group. It reads almost as much as “price only” to return 230 times fewer rows. - **precio > 1000**: zero bytes. The maximum of `precio` in the footer of each group is below 1,000, so the reader knows that no row satisfies the filter without reading a single page. Statistics only help if the data is **sorted** by the filter column. The next cell checks this: it rewrites the March 14 session sorted by time and counts, with `split_by_row_group`, how many groups survive two different filters in each version. ```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 minutes": [ grupos_leidos("ticks/fecha=2025-03-14", franja), grupos_leidos("por_hora.parquet", franja)], }, index=["sorted by symbol", "sorted by time"]) orden.columns.name = "row groups read" orden ``` ```output row groups read simbolo = BAGO 5 minutes sorted by symbol 1 6 sorted by time 6 1 ``` *Fig. — Row groups each filter reads in the 14 March session (346,529 ticks, 6 groups) depending on the order in which the file was written.* Four rows of six cells, one per row group. Sorted by symbol, the symbol filter reads 1 of 6 (highlighted) and the 5-minute window 6 of 6. Sorted by time, the symbol filter reads 6 of 6 and the 5-minute window 1 of 6. Symmetric: sorted by symbol, the symbol filter reads 1 group out of 6 and a 5-minute window reads all of them (each group has ticks from the whole session); sorted by time, the reverse. No single order serves everything, and the choice depends on the usual queries. For a backtest that goes through one asset at a time, symbol and time; to rebuild the whole market at one instant (a consolidated book, a cross-section), time. Some lakes keep both copies. Another option is to partition by one column and sort by the other, as this lesson does: partition by date and sort by symbol, so that a filter by date and symbol reads a single group from a single file. ## Arrow: the same bytes for everyone Reading little from disk is useless if afterwards each tool copies the data into its own format. **Apache Arrow** defines how columns are laid out in memory (contiguous value buffers, a validity bitmap for nulls and offsets for strings) in a language-agnostic way[^2]. Polars, DuckDB, pandas with `ArrowDtype` and PyArrow understand that format and can work on the same buffers without copying them (*zero-copy*). The cell checks this with the March 14 session: it measures how much new memory the Arrow allocator reserves in each conversion and whether the price buffer is the same. ```python import duckdb import polars as pl t = un_dia.combine_chunks() # one chunk per column 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 new,", "same buffer:", df_pl["precio"].to_arrow().buffers()[1].address == direccion) print(f"pandas ArrowDtype: {mb_ar:5.1f} MB new") print(f"pandas NumPy: {mb_np:5.1f} MB new") print("NumPy zero-copy:", v.ctypes.data == direccion, "| writable:", 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 new, same buffer: True pandas ArrowDtype: 0.0 MB new pandas NumPy: 12.8 MB new NumPy zero-copy: True | writable: False ┌───────┬─────────┐ │ ticks │ vwap │ │ int64 │ double │ ├───────┼─────────┤ │ 1568 │ 23.9505 │ └───────┴─────────┘ ``` What each line says: - **Polars** reserves nothing: its `DataFrame` points to the same price buffer as the Arrow table (the address comparison gives `True`; the addresses are not printed because they change on every run). - **pandas with `ArrowDtype`** does not either: each column is a wrapper around the Arrow array. - **pandas with NumPy** (what `to_pandas()` returns by default) reserves 12.8 MB in the Arrow allocator: the copy of the columns into the format NumPy expects. - **NumPy** can view the price buffer without copying it, but **read-only** (`writable: False`): writing to it would change data that the other tools believe is immutable. - **DuckDB** queries the Python variable `t` directly (the *replacement scan* from lesson 0.5.5) and reads its buffers in place: 1,568 BAGO ticks, the same ones `medir` read. *Zero-copy* has conditions. The most common one in market data is **nulls**. Arrow marks them in a separate bitmap and leaves an arbitrary value in the slot; NumPy has no nulls for `float64`, so to hand them over it has to copy and write `NaN` in those slots. If no copy is demanded, PyArrow refuses: ```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 ``` The other conditions: that the column is in a single chunk (that is why the cell uses `combine_chunks()`; a table read from several files has several) and that the type has the same representation in both tools. Strings, for example, are a byte buffer with offsets in Arrow and Python objects in a pandas `object` column: going from one to the other always copies. ## Polars lazy: the plan before the data Lesson 0.5.4 introduced `scan_parquet` and `explain()` with one file. With a partitioned lake, the plan also shows which files are opened. The query asks for BAGO's number of ticks and volume per trading venue in the week of March 3 to 7. First the plan as written (`optimized=False`) and then the one Polars runs: ```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 ``` Plans are read from the bottom up. Unoptimized, Polars would read all 60 files (`Parquet SCAN [… 59 other sources]`) with all the columns (`PROJECT */7 COLUMNS`: the six in the file plus `fecha`, which comes from the path) and filter afterwards, in two steps. The optimized plan changes three things: - **Partition pruning**: `Parquet SCAN [ticks/fecha=2025-03-03/…, … 4 other sources]`. The date filter is resolved with the folder names: five files out of 60. - **Projection**: `PROJECT 4/7 COLUMNS`. It reads `simbolo` and `fecha` to filter, and `centro` and `tamano` to aggregate. - **Filter pushed down to the reader**: `SELECTION: …` inside the `Parquet SCAN`. The reader applies the filter while reading, and can use the group statistics to skip those that do not contain `BAGO`. `ESTIMATED ROWS` is an estimate from the file sizes, not a count. The result: ```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 had 7,599 ticks that week, spread across the five venues in proportions similar to those of the generator (30% on `XNYS`, 20% on `ARCX` and on the dark pool, `OSCURO`, 15% on `BATS` and on `EDGX`): the query reads what it has to read and nothing more. ### The streaming engine `collect()` runs the plan with the default engine. With `engine="streaming"`, Polars runs the same plan in batches of rows: it reads a chunk of each file, filters it and aggregates it, and frees the memory before reading the next one. The full table is never in memory, only the state of the aggregation[^3]. The cell computes daily bars for the 200 symbols (open, high, low, close, volume, VWAP and number of ticks) from the 21 million 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()` and `last()` respect the read order, which here is that of the files (by symbol and time); in a lake without that order you would have to use `sort_by("ts")` inside the aggregation. The 12,000 bars (200 symbols × 60 sessions) add up to the 20,783,757 ticks: none has been lost along the way. What matters does not show in the output: memory. Measured on the machine the lesson was written on (16 cores, each variant in a separate process, peak resident memory of the process), the same daily bars: | Variant | Peak memory | |---|---| | pandas: `read_parquet` of the lake and `groupby` | ≈ 3.0 GB | | Polars: `read_parquet` of the lake and `group_by` | ≈ 2.5 GB | | Polars lazy with the in-memory engine | ≈ 2.0 GB | | Polars lazy with `engine="streaming"` | ≈ 0.7 GB | The full read pays for the 783 MB of Arrow plus the copies and the intermediate structures. The in-memory lazy query already saves (it reads only the columns it needs), and streaming drops to a quarter of the full read. With a lake ten times larger, the first three would grow in a similar way; the streaming one, much less, because it only keeps 12,000 groups. ## Out-of-core DuckDB DuckDB tackles the same problem from the other side: a memory manager with a **limit** (`memory_limit`) and a temporary folder (`temp_directory`) to which the operators that support it spill whatever does not fit (*spilling*): aggregations, joins, sorts and window functions[^4]. The test: a tick-by-tick log return with `lag()` over the 21 million rows, which forces each symbol to be sorted by time, with a 300 MB limit. ```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` is stored as `286.1 MiB` (DuckDB shows mebibytes, of 1,048,576 bytes). The query finishes: 20,783,757 ticks and 20,783,557 returns (`rendimientos`), 200 fewer, because the first tick of each symbol has no predecessor and `lag()` gives null. `suma_r2` is the sum of squared returns of the 200 symbols over the 60 sessions, including the jumps between one day's close and the next day's open: 5.427, about 0.027 per symbol. Spread over 60 sessions, √(0.027 / 60) ≈ 2.1% daily, the volatility the prices were generated with (2%) plus the noise from rounding to the cent, which in cheap stocks adds tick-by-tick variance. It is the known answer, and it serves to check that the window has paired each tick with the previous one of its symbol. What you do not see: measured on the lesson's machine, DuckDB wrote about 500 MB to `duckdb_tmp` during the query and the process did not go above about 530 MB of resident memory. Without a limit, the same query used ≈ 1.3 GB and wrote nothing to disk. The temporary files are deleted when the query finishes. With less memory, the number of threads matters: each thread needs its own buffers. With 100 MB and four threads, the query fails; with one thread, it finishes (more slowly): ```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)] ``` The error message comes with three remedies, in this order: fewer threads, `preserve_insertion_order = false` (DuckDB stops guaranteeing that rows come out in the order they went in when the query has no `ORDER BY`, which saves memory in large imports and exports) and more memory. With one thread, the result is identical to the previous one; measured on the lesson's machine, it took about eight seconds, versus about three with four threads and 300 MB. The DuckDB documentation spells out the limits of spilling: all blocking operators (grouping, joining, sorting and windows) can work with more data than memory, but some aggregate functions, such as `string_agg()`, do not spill to disk, and a query with several blocking operators can run out of memory even if each one on its own would not[^4]. ## What to use and when | Need | Tool | Why | |---|---|---| | Read few columns or few dates | Partitioned Parquet | Projection, partition pruning and statistics | | Filter by a field inside each file | Write order by that field | Statistics only help with sorted data | | Pass data between libraries | Arrow | Same buffers, no copy if there are no nulls or multiple chunks | | Python transformations with a small result | Polars lazy, `engine="streaming"` | Optimized plan and batched reading | | SQL, joins, large windows | DuckDB with `memory_limit` | Spills to disk whatever does not fit | | Several machines, terabytes of data | Dask, Spark (lesson 6.6.3) | When one machine is not enough, not before | The rule that sums up the lesson: before buying memory or setting up a cluster, look at the plan and count the bytes. In the lesson's lake, the difference between the worst and the best way to ask for BAGO's ticks for one day is 370 times (219 MB versus 0.59 MB), without changing anything in the hardware. [^1]: Apache Parquet, “Concepts”, “File Format” and “Configurations” (which recommends row groups of 512 MB to 1 GB): , accessed October 23, 2026. [^2]: Apache Arrow, “Arrow Columnar Format”: , accessed October 23, 2026. [^3]: Polars, user guide, “Streaming”: , accessed October 23, 2026. [^4]: DuckDB 1.5, “Tuning Workloads”: , accessed October 23, 2026. [^5]: DuckDB 1.5, “Configuration”: , accessed October 23, 2026. ## Exercises 1. Without filtering by `fecha`, ask the lake (`medir`) for the `ts` and `precio` columns of the 15:00 to 15:05 UTC window on March 14 (`franja`). How many bytes does it read and why is it not the 220 MB of the lake? **Solution.** ```python pd.DataFrame([medir("window without date", ["ts", "precio"], franja), medir("window with date", ["ts", "precio"], franja & f14)]) ``` ```output consulta filas MB_leidos pct_lago 0 window without date 4481 3.08 1.4 1 window with date 4481 3.08 1.4 ``` It reads 3.08 MB, 1.4% of the lake, with or without the date filter. The other 59 files are discarded anyway, but by another route: the minimum and maximum of `ts` in each row group of another session fall outside the window, so the statistics eliminate them without needing the partition. Within March 14, on the other hand, all six groups are read (sorted by symbol, they all cover the whole session), and only the `ts` and `precio` columns. If time-window queries were the usual ones, sorting by time would pay off. 2. Compute the same daily bars with DuckDB (with `arg_min` and `arg_max` for the open and the close) and check that they match the Polars ones. **Solution.** ```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} ``` The 12,000 bars match in all seven fields. `arg_min(precio, ts)` returns the price of the tick with the lowest `ts`: it is the open without depending on the read order, unlike `first()` in Polars, which trusts that the files are sorted. Comparing two independent engines on the same data is a cheap way to validate a computation before using it in a backtest. The last cell deletes the temporary data: the working folder is left as it was. ```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. A colleague wants to store the L2 book updates (lesson 1.1.3) of 500 stocks, about 2 billion rows a month, for two uses: backtests of one asset at a time and reconstructions of the whole market at specific moments. Propose the partitioning, the order and the row group size. **Solution.** Partition by date (one day per folder) and, within each day, several files by groups of symbols (for example, by first letter or by a hash of the symbol into 8-16 buckets), so that each file is in the hundreds of megabytes and not in the tens of gigabytes. Inside each file, sort by symbol and time: a single-asset backtest reads one file per day and, in it, the row groups of that symbol. For market reconstructions at an instant, that order forces reading all the files of the day, and the `ts` statistics do not help inside each one (all groups cover the session). If those queries are frequent, keep a second copy sorted by time (which is what many lakes do: storage is cheap compared to read time) or write row groups by symbol and time block. Row groups of tens to hundreds of megabytes: large enough to read sequentially, small enough for the statistics to discard. And before deciding, measure with the real queries as in this lesson. ## Self-assessment 1. In the lesson's lake, `simbolo = 'BAGO'` in one session reads 1 row group out of 6. With the same session written sorted by time, how many groups does it read for that filter? - A. 1, because statistics do not depend on the order - B. None, because the filter is resolved in the footer - C. All 6: sorted by time, each group contains ticks from almost every symbol and its `simbolo` minimum and maximum span the whole alphabet - D. 3, half of them *Answer C: All 6: sorted by time, each group contains ticks from almost every symbol and its `simbolo` minimum and maximum span the whole alphabet.* Statistics are the minimum and maximum of each column in each group. They only discard groups if the filter falls outside that interval, and that requires the data to be sorted (or clustered) by the filter column. Sorted by time, the opposite happens: a 5-minute window reads 1 group out of 6. 2. The query `precio > 1000` over the 21 million ticks reads 0 bytes of data. Why? - A. Because the maximum of `precio` in the footer of each row group is below 1,000: the reader knows no rows satisfy the filter without opening a single page - B. Because PyArrow has the query cached - C. Because the `precio` column is dictionary-encoded - D. Because it is a bug in the byte counter *Answer A: Because the maximum of `precio` in the footer of each row group is below 1,000: the reader knows no rows satisfy the filter without opening a single page.* The footers were read once before measuring (4.0 MB in 61 reads). With their statistics, the reader discards all 360 row groups. It is the extreme case of predicate pushdown. 3. `pa.array([101.5, None, 101.7]).to_numpy(zero_copy_only=True)` raises `ArrowInvalid`. What prevents it? - A. NumPy does not support numbers with decimals - B. The array is too small - C. Arrow stores arrays non-contiguously - D. The null: Arrow marks it in a separate bitmap and NumPy has no nulls in `float64`, so it has to copy and write `NaN` in the slot *Answer D: The null: Arrow marks it in a separate bitmap and NumPy has no nulls in `float64`, so it has to copy and write `NaN` in the slot.* With no nulls and a single chunk, NumPy can view the Arrow buffer as is (read-only). With nulls, multiple chunks or strings, the conversion copies. Polars and pandas with `ArrowDtype` keep the validity bitmap and do not copy. 4. Daily bars from 21 million ticks: ≈ 3.0 GB peak with pandas and ≈ 0.7 GB with Polars streaming. With 1-minute bars, streaming rises to ≈ 2.0 GB. Why? - A. Because the streaming engine does not work with timestamps - B. Because streaming bounds the memory of the input, not of the state: with 4.2 million open groups, the aggregation table takes up a lot - C. Because `truncate("1m")` forces the lake to be read twice - D. Because Polars switches to the pandas engine *Answer B: Because streaming bounds the memory of the input, not of the state: with 4.2 million open groups, the aggregation table takes up a lot.* The data is read in batches, but each group (`simbolo`, `minuto`) needs its state until the end. With 12,000 groups (daily bars) the state is small. For large results, write with `sink_parquet` or split the computation, for example one day at a time. 5. With `memory_limit = '100MB'` and four threads, the DuckDB window over the 21 million ticks fails with `Out of Memory Error`; with one thread, it finishes. What explains the difference? - A. With one thread DuckDB does not use `temp_directory` - B. With one thread the query is computed approximately - C. Each thread needs its own in-memory buffers: with fewer threads the work in progress fits and the rest spills to disk - D. With four threads the real limit is 25 MB *Answer C: Each thread needs its own in-memory buffers: with fewer threads the work in progress fits and the rest spills to disk.* The one-thread result is identical (20,783,557 returns, sum of squares 5.427), only slower. The error message itself suggests reducing threads, disabling `preserve_insertion_order` or raising the limit. ## Key points - A Parquet file has row groups and a footer with the minimum and maximum of each column in each group: the reader uses it to decide what to read. In the lesson's lake, BAGO's ticks for one day cost 0.59 MB out of 220 (0.27%) and `precio > 1000`, zero bytes. - Projection, partition pruning and statistics add up, but statistics only help if the data is sorted by the filter column: sorted by symbol, one symbol reads 1 group out of 6 and a 5-minute window reads all of them; sorted by time, the reverse. - Arrow is the common in-memory format: Polars, pandas with `ArrowDtype`, NumPy (read-only) and DuckDB use the same buffers without copying; nulls, multiple chunks and strings going to `object` force a copy. - The optimized Polars plan shows partition pruning (5 of 60 files), the columns read and the filter pushed down to the reader; `engine="streaming"` cut peak memory from ≈ 3.0 GB (pandas) to ≈ 0.7 GB, but it does not bound the state of an aggregation with millions of groups. - DuckDB spills to `temp_directory` whatever does not fit in `memory_limit` in aggregations, joins, sorts and windows: a window over 21 million rows finishes with 300 MB, and with 100 MB only with one thread. ## Resources ### Research - [Big Data Management. 1 - Introduction to Big Data for Trading with ArcticDB](https://pythonparatrading.com/en/big-data-for-trading/) · Research-0019 · 15 min ### Documentation - [Apache Parquet: concepts, file format and configuration (row group size)](https://parquet.apache.org/docs/file-format/configurations/) · parquet.apache.org - [Apache Arrow: in-memory columnar format](https://arrow.apache.org/docs/format/Columnar.html) · arrow.apache.org - [PyArrow: reading datasets with filters (pyarrow.dataset)](https://arrow.apache.org/docs/python/dataset.html) · arrow.apache.org - [Polars: the streaming engine](https://docs.pola.rs/user-guide/concepts/streaming/) · docs.pola.rs - [DuckDB 1.5: tuning workloads and out-of-core processing](https://duckdb.org/docs/1.5/guides/performance/how_to_tune_workloads) · duckdb.org ### Data - [Level 5 data: tickers of the stock panel (CSV)](https://pythonparatrading.com/curso/datos/nivel-5/panel5_activos.csv) · Download · panel5_activos.csv Previous lesson: [6.6.1 · When data becomes “big”: ticks, order books, options and universes](https://pythonparatrading.com/en/python-trading-course/level-6/when-financial-data-becomes-big-data/) Next lesson: [6.6.3 · Dask, Spark and Ray: from one machine to a cluster](https://pythonparatrading.com/en/python-trading-course/level-6/dask-spark-and-ray/)