Streaming¶

By default, to_arrow() parses the whole file into memory. For files larger than your RAM, stream instead: rypipe reads the file in bounded chunks and yields one Arrow RecordBatch at a time. Parser memory stays within the configured budget when the adapter supplies a streaming implementation; fallback adapters materialize the table first.

The entry point is iter_record_batches():

from crxml import CrystalXMLSource

src = CrystalXMLSource("report.xml", row_tag="Details")

for batch in src.iter_record_batches(memory="64MB"):
    print(batch.num_rows)

The examples below run on report.xml; the same iterator pattern applies to larger inputs when the adapter supports streaming. Collected output still uses memory proportional to its size.

The memory parameter¶

memory= sets the budget per chunk. It accepts a string or an integer number of bytes:

src.iter_record_batches(memory="64MB")
src.iter_record_batches(memory="512KB")
src.iter_record_batches(memory=67_108_864)  # 64 MB in bytes

Supported units: B, KB, MB, GB, TB (1024-based, case-insensitive, no space between the number and the unit). The KiB/MiB binary forms are not accepted; "64MiB" raises invalid memory. The budget guides batch sizing; input storage, worker queues, export buffers, and Python allocations add memory.

Streaming to a DataFrame¶

Concatenate the batches into one pandas DataFrame:

import pandas as pd
from crxml import CrystalXMLSource

src = CrystalXMLSource("report.xml", row_tag="Details")

df = pd.concat(b.to_pandas() for b in src.iter_record_batches(memory="64MB"))
print(df.shape)  # (15, 5)

The whole DataFrame still ends up in memory, of course. The win is that the parser never holds the whole file at once, and you can process or write out each batch as it arrives (see Writing Parquet and Batch-level control).

Streaming with pipelines¶

Pipelines stream too. Fusable stages are pushed into the streaming parse loop, so they run at full parsing speed:

from crxml import CrystalXMLSource, CastTypes, FilterRows

src = CrystalXMLSource("report.xml", row_tag="Details")

pipeline = (
    src
    | CastTypes({"Amount": float})
    | FilterRows(field="Status", op="==", value="Active")
)

total = sum(b.num_rows for b in pipeline.iter_record_batches(memory="64MB"))
print(total)  # 12

Writing Parquet with constant memory¶

The classic large-file pattern: stream batches straight into a Parquet file. Only one batch is in memory at a time:

import pyarrow.parquet as pq
from crxml import CrystalXMLSource

src = CrystalXMLSource("report.xml", row_tag="Details")

batches = src.iter_record_batches(memory="64MB")
first = next(batches)

with pq.ParquetWriter("output.parquet", first.schema) as writer:
    writer.write_batch(first)
    for batch in batches:
        writer.write_batch(batch)

Streaming to Polars¶

import pyarrow as pa
import polars as pl
from crxml import CrystalXMLSource

src = CrystalXMLSource("report.xml", row_tag="Details")

table = pa.Table.from_batches(list(src.iter_record_batches(memory="64MB")))
df = pl.from_arrow(table)

Parallel streaming¶

Pass threads= to parse chunks in parallel while keeping the same memory bound. This gives throughput close to full-RAM parallel parsing:

total = 0
for batch in src.iter_record_batches(memory="64MB", threads=16):
    total += batch.num_rows

Batch-level control¶

Because you get one batch at a time, you can stop early. Here we search for one record without parsing the rest of the file:

from crxml import CrystalXMLSource

src = CrystalXMLSource("report.xml", row_tag="Details")

for batch in src.iter_record_batches(memory="64MB"):
    df = batch.to_pandas()
    matches = df[df["Name"] == "Ivy Anderson"]
    if not matches.empty:
        print(matches.iloc[0].to_dict())
        break

{'Name': 'Ivy Anderson', 'Department': 'Sales', 'Amount': '6300.75', 'Status': 'Inactive', 'Date': '2026-01-13'}

Recap¶

• iter_record_batches(memory="64MB") yields Arrow batches within a fixed memory budget, from Sources and from pipelines.
• Build a DataFrame with pd.concat, write Parquet incrementally with pq.ParquetWriter, or process batches one at a time and stop early.
• threads= enables parallel streaming at the same memory bound.
• Units are B/KB/MB/GB/TB (case-insensitive, no spaces), or an integer number of bytes.
• Not streaming? Then call .clear_cache() after each file's final sink so ETL jobs over many files do not accumulate cached tables (see Sinks).

Next: Configuration, all available options.