polars learning notes: lazy scanning, streaming collect, and a 39GB read that crashed

thirteen things that actually stuck from three practice notebooks, and the eager read that undid all of them at the one file where it mattered

Closer to a personal reference sheet, for lookup or re-learning, than a piece meant to be read cold. Sorry if it doesn’t feel too clear on a cold read. These are things I learned from my own practice notebooks. Dataset: a 71.8 million row financial sentence corpus.

expr = pl.col("sentenceID").unique()
print("Expression object:", expr)          # prints a plan node, not data

result = df.select([expr])                 # now it's data

unique_sentence_ids = df.get_column("sentenceID").unique().to_list()   # eager, immediate, a plain list

1. pl.col(...) is a plan, not a value, until something evaluates it. Polars has two parallel worlds that look similar and aren’t. df.get_column(...) walks into the eager world and hands back a concrete column right away. Call .to_list() on it and you get a plain Python list. pl.col(...) walks into the expression world instead. It builds a small unevaluated tree that describes what to compute, and it only becomes data once you hand it to select, with_columns, or .agg(...). Printing the bare expression prints the plan, not the answer. That’s the moment that trips you up the first time, because nothing in the syntax warns you which world you’re in. Every later pattern in this list (window functions, struct unpacking, lazy scans) is really just more things you can put inside this expression tree before evaluating it once.

df.with_columns(
    y = pl.col.x + 3,
    x_max = pl.col("x").list.max(),
    x_sum = pl.col("x").list.sum()
).with_columns(
    z2 = pl.col("x").list.eval((pl.element() >= 2) & (pl.element() <= 4))
)

tickers_norm = (
    pl.col("tickers")
      .list.eval(pl.element().str.to_uppercase())   # a string method, run *inside* every list cell
      .list.unique().list.sort()
)

2. A list column isn’t something you iterate. It’s a mini expression scope. .list.max() and .list.sum() are what you’d expect: one aggregate number per row. .list.eval() is the interesting one. It takes an expression written in terms of pl.element(), the placeholder for “the current item inside this list,” and runs it vectorized against every element of every list in the column, no Python for loop unpacking anything. The first example runs a numeric predicate elementwise (>= 2 & <= 4). The second runs a string method elementwise (.str.to_uppercase()). Worth keeping: .list.eval() isn’t a special tool for numbers. It’s the entire expression language, just scoped one level down into the list.

exploded = toy.select(["docID", "tickers"]).explode("tickers")
result1 = (
    exploded.group_by("docID")
    .agg(pl.col("tickers").unique().sort().alias("merged_tickers"))
)

result = (
    toy.group_by("docID")
    .agg(pl.col("tickers").flatten())          # no explode step at all, it just works
)

row_concat = toy.select([
    pl.concat_list(["tickers", "exchanges", "alt"]).alias("combined")   # merges across COLUMNS, same row
])

3. “Merge these lists” splits into two different questions, and each has its own function. Merging lists down a column, across rows in a group (three filings’ ticker lists into one) is what explode, then group_by, then re-aggregate does. Turns out to be unnecessary ceremony, because .flatten() inside .agg(...) does the same collapse in one line, no intermediate row-per-element table needed. Merging lists across columns, within the same row (this row’s tickers plus this row’s exchanges into one combined list) is a completely different operation, pl.concat_list. Confusing the two is an easy mistake, because both start from “I have several lists and I want one list.”

df.with_columns(
    pl.when(pl.col("x").str.starts_with("foo"))
      .then(pl.col(pl.String).str.to_uppercase())
      .otherwise(pl.col(pl.String))
)

kpi_sentences.filter(
    pl.col("likely_kpi") | pl.col("has_numbers")
).filter(
    pl.col("section").is_in([9, 10])
)

4. Conditional logic in Polars is either a rewrite or a filter, and they’re not interchangeable. pl.when().then().otherwise() rewrites values. Every row survives, some just get a different value written into them (here: uppercase every string column, but only on rows whose x starts with "foo"). .filter(A | B) removes rows. Nothing gets rewritten, some rows just stop existing in the output. Both are boolean logic over columns, both avoid a Python-level branch. One shrinks the table and one doesn’t. Picking the wrong one is a subtle bug, not an error message.

print(df.schema.get("returns"))
print(df.select(pl.col("returns")).head(2))     # look before you unpack

df_flat = df.with_columns([
    pl.col("returns").struct.field("1d").struct.field("ret").alias("ret_1d"),
    pl.col("returns").struct.field("5d").struct.field("ret").alias("ret_5d"),
    pl.col("returns").struct.field("30d").struct.field("ret").alias("ret_30d"),
])

5. Real filing data doesn’t arrive flat. SEC returns ship as a nested struct. returns isn’t a number column, it’s a struct with three named sub-structs (1d, 5d, 30d), each holding its own ret field, like returns.5d.ret, three levels deep. .struct.field(...) chains exactly like a path: one call per level. The habit worth keeping is the first snippet, not the second. Check df.schema.get(...) or peek at one row before writing the unpacking chain. Guessing field names on a nested struct wastes more time than the two-second look.

df_year = df.with_columns(
    pl.col("reportDate").str.strptime(pl.Date, strict=False).dt.year().alias("fiscal_year")
)
comp_rollup = df_year.filter(
    (pl.col("fiscal_year") > 2014) & (pl.col("section").is_in(TOP10))
)

kpi_sentences = (
    df.lazy()
    .filter(pl.col("reportDate").str.slice(0, 4).is_in(TARGET_YEARS))   # cheaper: no real date parse
    .filter(pl.col("section").is_in(KPI_SECTIONS))
)

6. The column you want to filter on sometimes doesn’t exist yet. You derive it once, then treat it like any other column. reportDate arrives as a string. fiscal_year doesn’t exist until strptime plus .dt.year() make it exist. Worth noticing: two different notebooks solve “get the year” two different ways. One is a real typed Date parse (correct, handles malformed dates via strict=False, costs an actual parse). The other is a raw str.slice(0, 4) on the assumption the field always starts with a 4-digit year (cheaper, and silently wrong the moment that assumption breaks). Neither is “the” right answer. It’s a real correctness versus cost tradeoff that shows up again later at a much bigger scale.

DATA_PATH = Path("../data/exports/sec_filings_small_full.parquet")
df = pl.read_parquet(DATA_PATH)     # eager, loads now
lf = pl.scan_parquet(DATA_PATH)     # lazy, loads nothing yet, just describes a plan

lf_base = (
    pl.scan_parquet(DATA_PATH)
    .with_columns(...)
    .filter(...)
)
lf_cached = lf_base.cache()          # pin one filtered slice, branch multiple aggregations off it

7. Going lazy is a decision you check, not a habit you trust. pl.scan_parquet instead of pl.read_parquet costs nothing to write and means something completely different. The file isn’t read until something downstream forces it. Everything between the scan and the final .collect() is a plan the query optimizer gets to rewrite: pushing filters down to the file reader, dropping unread columns, before a single row moves. .cache() on a lazy frame is the companion move. Once you’ve built one filtered or derived slice you plan to reuse for several different aggregations, cache it so the shared filtering work happens once. My own note from around this point says to check with .explain() that only the needed columns and rows are actually being scanned, and that doing this catches real speedups. I never wrote down a measured number for that, so treat it as a habit worth keeping, not a benchmark.

docs = (
    df_year
    .group_by("docID")
    .agg([
        pl.first("cik").alias("cik"),
        pl.first("fiscal_year").alias("fiscal_year"),
        pl.n_unique("section").alias("n_sections_doc"),
        pl.count().alias("n_sentences_doc"),
    ])
)

company_year = (
    docs
    .group_by(["cik", "name", "fiscal_year"])
    .agg([
        pl.n_unique("docID").alias("n_docs"),
        pl.sum("n_sentences_doc").alias("n_sentences"),
    ])
)

8. The “grain” of a table has to be nailed down before you aggregate across it, or the higher-up numbers quietly double count. The raw table is one row per sentence. Rolling straight from sentences to a company-year summary in one aggregation would double count anything that varies per sentence but is really a per-filing fact. So it happens in two explicit steps. First, collapse to one row per docID (a real filing), carrying forward the doc-level facts with pl.first(...) since they’re constant within a filing. Only then roll that table up to company-year. Two aggregations instead of one looks like more code. It’s actually the thing that keeps the second aggregation’s counts honest.

lf_doc_returns_long = (
    lf_docs
    .select("docID", "cik", "name", "fiscal_year", "ret_1d", "ret_5d", "ret_30d")
    .unpivot(
        index=["docID", "cik", "name", "fiscal_year"],
        on=["ret_1d", "ret_5d", "ret_30d"],
        variable_name="horizon",
        value_name="ret"
    )
    .filter(pl.col("ret").is_not_null())
    .with_columns(
        horizon = pl.col("horizon").str.replace("^ret_", "", literal=False),
        ret_safe = pl.col("ret").clip(lower_bound=-0.9999),
        # if the Polars build lacks Expr.clip(), the fallback written down at the time was:
        # ret_safe = pl.when(pl.col("ret") < -0.9999).then(-0.9999).otherwise(pl.col("ret"))
    )
)

9. Three parallel columns become one column with a label, so one aggregation replaces three. ret_1d, ret_5d, ret_30d are the same kind of fact at three horizons. Exactly the shape .unpivot() is for. index names what stays fixed per row, on names the columns being folded down into two new ones: horizon (which column it came from) and ret (the value). One group_by(["cik","horizon"]) downstream now covers all three horizons instead of three near-identical blocks of aggregation code. The ret_safe clip is a small piece of domain hygiene riding along. A return below minus 100% would make the next step’s log() blow up, so it gets floored first, with a plain when/then fallback written down in a comment in case .clip() isn’t available in a given Polars build. Worth keeping as a pattern in general.

comp_ret = ((pl.col("ret_safe") + 1).log().sum()).exp() - 1     # compounded return, one group, one year

log1p_ret = (pl.col("comp_ret") + 1).log()
cagr           = pl.col("log1p_ret").mean().exp() - 1            # across years: mean, not sum, then back out
total_comp_ret = pl.col("log1p_ret").sum().exp() - 1

10. Averaging returns arithmetically is wrong. The fix is log-sum-exp, and it’s just expression chaining. Compounding is multiplicative, (1+r1)(1+r2)..., and a plain mean of the r’s overstates it. The standard fix converts each return to log(1+r), sums them (or means them, for an annualized CAGR across years), and exponentiates back. sum for total compounded return across docs in one year, mean for CAGR across years, same log1p groundwork underneath both. Nothing about this needs a special finance function. It composes with group_by().agg() like any other expression. The “hard” quant technique and the “boring” aggregation machinery are the same tool.

df_ranked = df.with_columns([
    (pl.int_range(0, pl.len()).over("store") + 1).alias("row_num"),
    pl.col("sales").rank().over("store").alias("rank"),
    (pl.col("sales") == pl.col("sales").max().over("store")).alias("is_best_day"),
])

lf_first3_per_doc = (
    lf_base
    .sort(["docID", "sentenceCount"])
    .with_columns([
        pl.int_range(0, pl.len()).over("docID").alias("pos_in_doc"),
        pl.int_range(0, pl.len()).over(["docID", "section"]).alias("pos_in_sect"),
    ])
)

11. group_by collapses a group to one row. .over(...) keeps every row and stamps the group fact back onto each one. pl.col("sales").sum().over("store") computes the same total a group_by("store").agg(sum) would, but the toy dataframe still has 6 rows afterward, not 2. Every row now carries its store’s total alongside it, which is what makes per-row things like rank, row-number, and “is this the best day for this store” possible without a join back to a collapsed table. The real version does the same thing with a compound key. .over(["docID", "section"]) numbers each sentence’s position within its own section of its own filing, a compound window key instead of a single one. That’s the version that later feeds into “is this the first sentence of a section” flags.

company_horizon_rollup = (
    lf_yearly
    .group_by(["cik", "name", "horizon"])
    .agg(
        _best  = pl.struct(["fiscal_year", "mean_ret"]).sort_by("mean_ret", descending=True).first(),
        _worst = pl.struct(["fiscal_year", "mean_ret"]).sort_by("mean_ret", descending=False).first(),
    )
    .with_columns(
        best_year  = pl.col("_best").struct.field("fiscal_year"),
        best_ret   = pl.col("_best").struct.field("mean_ret"),
        worst_year = pl.col("_worst").struct.field("fiscal_year"),
    )
)

12. “Argmax, but I also want the other columns from that row” doesn’t have a built-in. You pack a struct, sort it, and unpack it back. .max() inside an aggregation gives you the best value, and nothing else about the row it came from. Wrapping the columns you actually need into pl.struct([...]), sorting that list of structs by the field you care about, and taking .first() (or .last(), or flipping descending) gets you the whole row’s worth of context back. Here, which fiscal year was the best and worst for a company, not just the number itself. It’s a pattern, not a one-off: pack, sort_by, first, unpack. Generalizes to any “best row in the group, with its context” problem.

kpi_sentences = (
    df.lazy()
    .filter(pl.col("reportDate").str.slice(0, 4).is_in(TARGET_YEARS))
    .filter(pl.col("section").is_in(KPI_SECTIONS))
)

s9_kpi_df = (
    kpi_sentences
    .filter(pl.col("likely_kpi") | pl.col("has_numbers"))
    .filter(pl.col("section").is_in([9, 10]))
    .sort(["year", "section", "name"])
    .head(400)
)

13. By the last notebook, the tools stop being demonstrations and start being a real narrowing pipeline. Nothing here is a new technique. It’s .lazy(), .filter(), .sort(), .head(), all already covered. What changed is the target: from “show that this works” to “get from 600K+ sentences down to a small KPI-likely candidate set.” Each filter is independently readable (right years, right sections, looks numeric-or-KPI-ish, cap the result) rather than one dense boolean condition. That’s the actual skill: chaining simple filters instead of writing one clever one.

where the practice habits met the real file

Every one of the thirteen things above ran against sec_filings_small_full.parquet. Small enough that eager versus lazy barely showed up as a difference, filters pushed down cheaply, and .explain() came back clean. It was a sandbox, and a good one, but it never once tested the thing it was implicitly training for.

The real corpus is 71,866,962 rows. Converting the raw HuggingFace shards to a single Polars DataFrame was the very first step, before any of the thirteen patterns even applied, and it put 39.03GB in memory. That got written down to a 1.53GB Parquet file (28.7x compression). The next real line of code against that file was df_large = pl.read_parquet("sec_filings_large_full.parquet"). Eager, unscanned, no .lazy() in sight. The kernel crashed with no traceback. A Windows access violation, not a catchable Python exception. That happened after a dozen cells of correctly reaching for .scan_parquet() on the small file. The habit was there, and it still didn’t fire on the one file where it mattered.

The fix exists, but only as a rule, not as a rerun. platform_core_contract.py already says it plainly: kernel crashes mean switch to lazy Polars, scan_parquet, collect at the end. It works elsewhere in the project, at the embeddings scale, 614,787 rows, roughly 2GB. It was never pointed back at the 71.8M-row file that actually broke. The bulk work at that scale went to DuckDB instead. So whether lazy Polars actually holds at 71.8M rows is still open. Nobody ever ran it again on the file that raised the question in the first place.

things I think I forgot to write down, worth adding

The one sentence that ties all thirteen points to the crash, and I don’t think I ever stated it this plainly: lazy defers work, it does not shrink the result. .scan_parquet() buys you predicate and projection pushdown, real savings. But if the query you eventually .collect() still has to materialize the full wide result (every column, every row, no filter having survived to the end), laziness bought you nothing. That line, found almost verbatim in my own TechNotes_MemoryExp_Handling.md, is the actual thesis connecting the twelve clean examples to the one crash. Deserves to be written where I’ll see it again, not buried in a debugging note.

A quiet API drift worth flagging for future me: the working code across this project calls .collect(streaming=True). Current Polars docs show the same feature as .collect(engine="streaming"). Nothing says the old form is deprecated, but if I ever copy this code into something new, copy the current spelling, not the one that already works here.

Row-group size is the same “grain” idea from point 8, one layer down. I set the DuckDB row group size parameter to 100000. This is what happened: smaller row groups give finer grained min/max stats, so a filter has more chances to skip a group entirely, at some cost in footer metadata per group. The source file’s own row groups average 261,334 rows per group, much bigger than what I chose.

Open experiment, still todo: does pl.scan_parquet(large_file).filter(...).collect(engine="streaming") handle the 71.8M row file that crashed the kernel? DuckDB already solved the operational problem. This would answer the Polars question specifically.