First Steps¶

This page explains the complete example from the Tutorial line by line. By the end, you will understand what a Source is, how the pipeline operator works, and what rypipe does behind the scenes.

The example¶

Here is the complete code we will explain:

from crxml import CrystalXMLSource, RenameFields, CastTypes, FilterRows

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

df = (
    source
    | RenameFields({"Name": "name"})
    | CastTypes({"Amount": float})
    | FilterRows(field="Status", op="==", value="Active")
).to_pandas()

print(df)

Step 1: Create a Source¶

from crxml import CrystalXMLSource

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

A Source is a handle over one input file. It does not parse the file yet. It stores the path and configuration, and waits for you to ask for data.

The row_tag="Details" argument is adapter-specific: it tells the crxml adapter which XML element represents a row. Each adapter accepts its own kwargs; check your adapter's documentation for what it supports.

What a Source gives you¶

Step 2: Parse with to_arrow()¶

source = CrystalXMLSource("report.xml", row_tag="Details")
table = source.to_arrow()

The first time you call .to_arrow(), rypipe parses the file:

1. The Splitter finds row boundaries in the byte stream.
2. The RecordParser extracts field values from each row.
3. The Engine accumulates values into Arrow columns.
4. The result is a pyarrow.Table.

The table is cached. Subsequent calls to .to_arrow(), .to_pandas(), etc. reuse the cached table without re-parsing.

What rypipe does automatically¶

With just that one call, rypipe:

• Splits the file into chunks for parallel parsing.
• Discovers the schema from the data.
• Builds Arrow column arrays with near-zero copy.
• Returns a pyarrow.Table you can use directly.

Step 3: Chain stages with |¶

result = (
    source
    | RenameFields({"Name": "name"})
    | CastTypes({"Amount": float})
    | FilterRows(field="Status", op="==", value="Active")
)

The | operator chains transformation stages. Each stage transforms the data as it flows through, like a Unix pipe:

• RenameFields renames the "Name" column to "name".
• CastTypes casts the "Amount" column from string to float.
• FilterRows keeps only rows where Status equals "Active".

Each | returns a new Pipeline — the original Source is not modified.

What rypipe does automatically¶

When you call .to_pandas() on the pipeline, rypipe:

1. Splits the stages into fusable and non-fusable groups.
2. Pushes fusable stages (RenameFields, DropFields, CastTypes, constant FilterRows) into the Rust parse loop via the plan.
3. Runs remaining stages (lambda predicates, complex combinators) over Arrow batches in Python.

Fusable stages run at Rust speed during parsing: they never touch Python.

Step 4: Get a DataFrame¶

df = result.to_pandas()
print(df)
#     name  Amount  Status
# 0  Alice   150.0  Active
# 2  Carol   200.0  Active

.to_pandas() materializes the pipeline into a pandas DataFrame. You can also use:

• .to_arrow() for a pyarrow.Table
• .to_polars() for a Polars DataFrame
• .to_parquet(path) to write to a Parquet file
• rypipe.collect() to get a list of dicts

See Sinks for the full reference.

Recap¶

• A Source is a handle over one input file. It parses lazily and caches.
• The | operator chains stages into a Pipeline.
• Stages (RenameFields, CastTypes, FilterRows) transform data.
• Sinks (.to_pandas(), .to_arrow()) materialize results.
• rypipe pushes fusable stages into the Rust parse loop automatically.

Next: Building an Adapter, the basic scaffolding.