Sinks¶

Sinks materialize pipeline results into tables, DataFrames, or files. You can use them as methods on a Source or as standalone functions on a Pipeline.

Source methods¶

Every Source has built-in sink methods:

from crxml import CrystalXMLSource

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

to_arrow()¶

Returns a pyarrow.Table. This is the default materialization:

table = src.to_arrow()
print(table.schema)
# name: string
# amount: double

to_pandas()¶

Returns a pandas DataFrame with PyArrow-backed dtypes by default:

df = src.to_pandas()
print(df.dtypes)
# name      string[pyarrow]
# amount    double[pyarrow]

You can disable PyArrow backing with dtype_backend="numpy":

df = src.to_pandas(dtype_backend="numpy")

to_dataframe()¶

Alias for to_pandas():

df = src.to_dataframe()  # same as src.to_pandas()

to_polars()¶

Returns a Polars DataFrame:

import polars as pl

df = src.to_polars()
print(df.columns)
# ['name', 'amount']

to_parquet()¶

Writes the table to a Parquet file:

src.to_parquet("output.parquet")

# Pass additional pyarrow.parquet options
src.to_parquet("output.parquet", compression="snappy")

clear_cache()¶

Drops the cached Arrow table to free memory:

src.clear_cache()
# Next to_arrow() call will re-parse the file

Pipeline functions¶

When working with a Pipeline (the result of src | stage), use the standalone sink functions from the adapter:

from crxml import CrystalXMLSource, FilterRows, collect

src = CrystalXMLSource("report.xml", row_tag="Details")
pipeline = src | FilterRows(field="status", op="==", value="active")

collect()¶

Collects all rows into a list of dicts:

rows = rypipe.collect(pipeline)
print(rows[0])
# {"name": "Alice", "amount": 150.0, "status": "active"}

to_arrow() (function)¶

Materializes a pipeline to a pyarrow.Table:

table = rypipe.to_arrow(pipeline)

to_pandas() (function)¶

Materializes a pipeline to a pandas DataFrame:

df = rypipe.to_pandas(pipeline)

to_polars() (function)¶

Materializes a pipeline to a Polars DataFrame:

df = rypipe.to_polars(pipeline)

to_csv() (function)¶

Writes pipeline results to a CSV file:

rypipe.to_csv(pipeline, "output.csv")

# Custom delimiter and encoding
rypipe.to_csv(pipeline, "output.tsv", delimiter="\t", encoding="utf-8")

Parameters:

• pipeline: iterable of dicts.
• path: output file path.
• encoding: file encoding (default: "utf-8").
• delimiter: column delimiter (default: ",").
• fieldnames: optional list of column names. If omitted, uses the keys from the first row.

to_parquet() (function)¶

Writes pipeline results to a Parquet file:

rypipe.to_parquet(pipeline, "output.parquet")

Which sink should I use?¶

Recap¶

• Source methods: .to_arrow(), .to_pandas(), .to_polars(), .to_parquet(), .clear_cache().
• Standalone functions: rypipe.collect(), rypipe.to_arrow(), rypipe.to_pandas(), rypipe.to_polars(), rypipe.to_csv(), rypipe.to_parquet().
• Source methods reuse the cached table. Standalone functions re-parse if the pipeline hasn't been materialized yet.

Next: Streaming: processing large files with bounded memory.