Sinks¶

Sinks materialize pipeline results into tables, DataFrames, or files. This page explains each sink and shows how to implement them in your adapter.

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
# Department: string
# Amount: string
# Status: string
# Date: string

to_pandas()¶

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

df = src.to_pandas()
print(df.dtypes)
# Name       string[pyarrow]
# Department string[pyarrow]
# Amount     string[pyarrow]
# Status     string[pyarrow]
# Date       string[pyarrow]

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

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

to_polars()¶

Returns a Polars DataFrame:

import polars as pl

df = src.to_polars()
print(df.columns)
# ['Name', 'Department', 'Amount', 'Status', 'Date']

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:

from crxml import collect

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

to_arrow¶

Materializes a pipeline to a pyarrow.Table:

from crxml import to_arrow

table = to_arrow(pipeline)

to_pandas¶

Materializes a pipeline to a pandas DataFrame:

from crxml import to_pandas

df = to_pandas(pipeline)

to_polars¶

Materializes a pipeline to a Polars DataFrame:

from crxml import to_polars

df = to_polars(pipeline)

to_csv¶

Writes pipeline results to a CSV file:

from crxml import to_csv

to_csv(pipeline, "output.csv")

# Custom delimiter and encoding
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¶

Writes pipeline results to a Parquet file:

from crxml import to_parquet

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: crxml.collect(), crxml.to_arrow(), crxml.to_pandas(), crxml.to_polars(), crxml.to_csv(), crxml.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.