CodeRef. Polars
All guides
Polars / Chapter 6 of 7

Data pipeline examples

Follow complete Polars pipelines for messy sales exports, schema drift, ledger reconciliation, and files larger than memory.

4 topicsUpdated
39 / 50

Complete pipeline: clean a messy sales export#

Read raw strings, quarantine bad rows, check IDs, calculate totals, and export a report.

This input has whitespace, mixed date formats, a duplicate order, an invalid amount, and a negative quantity. One row represents one order in this export, so order ID must be unique after the winner rule. The accepted amount format is USD with comma thousands separators and two decimal places. Quarantine means retaining invalid source rows in a separate output instead of silently dropping or repairing them.

TextSample input — sales.csv
order_id,updated,day,customer,qty,price
001,1,2026-09-01, a ,2,"USD 1,000.00"
002,1,02/09/2026,b,1,USD 250.00
002,2,02/09/2026,b,2,USD 250.00
003,1,2026-09-03,c,1,unknown
004,1,2026-09-04,d,-1,USD 100.00
PythonA runnable sales pipeline with a reject file and typed outputs
from pathlib import Path
from tempfile import TemporaryDirectory
from decimal import Decimal
import polars as pl

csv = """order_id,updated,day,customer,qty,price
001,1,2026-09-01, a ,2,"USD 1,000.00"
002,1,02/09/2026,b,1,USD 250.00
002,2,02/09/2026,b,2,USD 250.00
003,1,2026-09-03,c,1,unknown
004,1,2026-09-04,d,-1,USD 100.00
"""
with TemporaryDirectory() as folder:
    root = Path(folder)
    source = root / "sales.csv"
    source.write_text(csv, encoding="utf-8")
    raw = pl.scan_csv(source, infer_schema=False).with_row_index("source_row")
    parsed = raw.with_columns(
        pl.col("order_id").str.strip_chars(),
        pl.col("customer").str.strip_chars().str.to_uppercase(),
        pl.col("updated").cast(pl.Int64, strict=False).alias("version"),
        pl.col("qty").cast(pl.Int64, strict=False).alias("quantity"),
        pl.coalesce(
            pl.col("day").str.to_date("%Y-%m-%d", strict=False),
            pl.col("day").str.to_date("%d/%m/%Y", strict=False),
        ).alias("date"),
        pl.col("price")
        .str.strip_chars()
        .str.replace_all("USD ", "", literal=True)
        .str.replace_all(",", "", literal=True)
        .cast(pl.Decimal(18, 2), strict=False)
        .alias("unit_price"),
    )
    checked = parsed.with_columns(
        (
            pl.col("order_id").is_not_null()
            & (pl.col("order_id") != "")
            & pl.col("customer").is_not_null()
            & (pl.col("customer") != "")
            & pl.col("version").is_not_null()
            & pl.col("date").is_not_null()
            & (pl.col("quantity") > 0)
            & (pl.col("unit_price") >= 0)
        )
        .fill_null(False)
        .alias("valid")
    )
    rejected = checked.filter(~pl.col("valid")).collect()
    rejected.write_parquet(root / "rejected.parquet")
    accepted = (
        checked.filter(pl.col("valid"))
        .sort(["order_id", "version", "source_row"])
        .unique(subset="order_id", keep="last", maintain_order=True)
        .with_columns((pl.col("quantity") * pl.col("unit_price")).alias("revenue"))
    )
    lines = accepted.collect()
    assert lines["order_id"].n_unique() == lines.height
    report = (
        lines.group_by("customer")
        .agg(pl.len().alias("orders"), pl.col("revenue").sum())
        .sort("customer")
    )
    report.write_parquet(root / "report.parquet")
    report.write_csv(root / "report.csv")
    print("rejected IDs:", rejected["order_id"].to_list())
    print(
        "accepted orders:",
        lines.select("order_id", "quantity").sort("order_id").to_dicts(),
    )
    print("report:", report.to_dicts())
    assert report["revenue"].sum() == Decimal("2500.00")
    print("written: rejected.parquet, report.parquet, report.csv")
TextExpected output
rejected IDs: ['003', '004']
accepted orders: [{'order_id': '001', 'quantity': 2}, {'order_id': '002', 'quantity': 2}]
report: [{'customer': 'A', 'orders': 1, 'revenue': Decimal('2000.00')}, {'customer': 'B', 'orders': 1, 'revenue': Decimal('500.00')}]
written: rejected.parquet, report.parquet, report.csv
StepEffect on this input
Read as stringsPreserves 001 and other IDs; retains original date and price text.
Parse and normalizeCustomer a becomes A, both date formats become Date, price becomes exact Decimal.
Quality check003 fails amount parsing; 004 has a negative quantity. Both remain in the reject file.
Latest valid version002 version 2 replaces version 1; two accepted orders remain.
Transform and summarizeA totals 2000.00, B totals 500.00.
ExportParquet retains decimal types; CSV needs a schema when read back.
Make winner and reject policies explicit

This example chooses the latest valid version, not “the latest version, then reject if invalid.” Those policies can differ. A production job should record invalid reasons, duplicate losers, source file identities, input/accepted/rejected counts, and whether superseded rows are expected. Keep transformations lazy while useful; collect only when the result size and validation needs are understood.

Official reference: Complete pipeline: clean a messy sales export.

40 / 50

Complete pipeline: merge evolving customer files#

Normalize schemas, keep provenance, choose the latest record, and write a checked snapshot.

The files use different ID fields and schemas. Normalize them to one schema, retain the source name, and break ties by version and source. Versions are increasing integers; timestamps need an explicit time zone.

PythonCombine two schema versions and keep the latest customer snapshot
from pathlib import Path
from tempfile import TemporaryDirectory
import polars as pl

with TemporaryDirectory() as folder:
    root = Path(folder)
    (root / "v1.csv").write_text(
        "id,version,email\nA,1,old@a.test\nB,1,b@b.test\n", encoding="utf-8"
    )
    (root / "v2.csv").write_text(
        "customer_id,version,email,country\nA,2,new@a.test,BD\nC,1,c@c.test,BD\n",
        encoding="utf-8",
    )
    old = (
        pl.scan_csv(root / "v1.csv", infer_schema=False)
        .rename({"id": "customer_id"})
        .with_columns(
            pl.lit(None, dtype=pl.String).alias("country"),
            pl.lit("v1.csv").alias("source"),
        )
    )
    new = pl.scan_csv(root / "v2.csv", infer_schema=False).with_columns(
        pl.lit("v2.csv").alias("source")
    )
    schema = {
        "customer_id": pl.String,
        "version": pl.Int64,
        "email": pl.String,
        "country": pl.String,
        "source": pl.String,
    }

    def normalize(frame):
        return frame.select(
            *(pl.col(name).cast(dtype).alias(name) for name, dtype in schema.items())
        )

    merged = pl.concat([normalize(old), normalize(new)])
    invalid_count = (
        merged.filter(pl.col("customer_id").is_null() | (pl.col("customer_id") == ""))
        .select(pl.len())
        .collect()
        .item()
    )
    if invalid_count:
        raise ValueError("snapshot has missing IDs")
    snapshot = (
        merged.sort(["customer_id", "version", "source"])
        .unique(subset="customer_id", keep="last", maintain_order=True)
        .sort("customer_id")
    )
    target = root / "customers.parquet"
    snapshot.sink_parquet(target, engine="streaming")
    result = pl.read_parquet(target)
    assert dict(result.schema) == schema
    assert result["customer_id"].n_unique() == result.height
    print(result.to_dicts())
    print("written: customers.parquet")
TextExpected output
[{'customer_id': 'A', 'version': 2, 'email': 'new@a.test', 'country': 'BD', 'source': 'v2.csv'}, {'customer_id': 'B', 'version': 1, 'email': 'b@b.test', 'country': None, 'source': 'v1.csv'}, {'customer_id': 'C', 'version': 1, 'email': 'c@c.test', 'country': 'BD', 'source': 'v2.csv'}]
written: customers.parquet
What changes

The newer record supplies A’s email and country; B’s missing country remains null. The source column records each winner. Strict casts reject incompatible versions. Use stable input files when validation and export run separate queries.

Official reference: Complete pipeline: merge evolving customer files.

41 / 50

Complete pipeline: reconcile ledger and bank files#

Use source-presence flags so missing records and missing amounts stay distinguishable.

PythonNormalize, validate, compare, and export reconciliation results
from pathlib import Path
from tempfile import TemporaryDirectory
import polars as pl

with TemporaryDirectory() as folder:
    root = Path(folder)
    (root / "ledger.csv").write_text(
        "tx,cents\n t1 ,100\nt2,200\nt3,\n", encoding="utf-8"
    )
    (root / "bank.csv").write_text(
        "tx,cents\nT1,100\nT2,190\nT4,400\n", encoding="utf-8"
    )

    def load(name, side):
        raw = pl.read_csv(root / name, infer_schema=False)
        data = raw.select(
            pl.col("tx").str.strip_chars().str.to_uppercase(),
            pl.col("cents").cast(pl.Int64).alias(side + "_cents"),
            pl.lit(True).alias(side + "_present"),
        )
        if data["tx"].null_count() or data.filter(pl.col("tx") == "").height:
            raise ValueError("missing transaction key")
        if data["tx"].n_unique() != data.height:
            raise ValueError("duplicate transaction key")
        return data

    ledger, bank = load("ledger.csv", "ledger"), load("bank.csv", "bank")
    report = ledger.join(bank, on="tx", how="full", coalesce=True, validate="1:1")
    report = report.with_columns(
        pl.when(pl.col("ledger_present").is_null())
        .then(pl.lit("bank_only"))
        .when(pl.col("bank_present").is_null())
        .then(pl.lit("ledger_only"))
        .when(pl.col("ledger_cents").is_null() | pl.col("bank_cents").is_null())
        .then(pl.lit("missing_amount"))
        .when(pl.col("ledger_cents") != pl.col("bank_cents"))
        .then(pl.lit("amount_diff"))
        .otherwise(pl.lit("matched"))
        .alias("status")
    )
    report = report.sort("tx")
    report.write_parquet(root / "reconciliation.parquet")
    report.filter(pl.col("status") != "matched").write_csv(root / "exceptions.csv")
    print(report.select("tx", "ledger_cents", "bank_cents", "status").to_dicts())
    print("exceptions:", report.filter(pl.col("status") != "matched").height)
TextExpected output
[{'tx': 'T1', 'ledger_cents': 100, 'bank_cents': 100, 'status': 'matched'}, {'tx': 'T2', 'ledger_cents': 200, 'bank_cents': 190, 'status': 'amount_diff'}, {'tx': 'T3', 'ledger_cents': None, 'bank_cents': None, 'status': 'ledger_only'}, {'tx': 'T4', 'ledger_cents': None, 'bank_cents': 400, 'status': 'bank_only'}]
exceptions: 3
Source-presence flags

T3 has a ledger row with a null amount and no bank row, so it is ledger_only. If both rows exist but an amount is null, it is missing_amount. Strict parsing rejects malformed numbers; add quarantine handling if processing should continue.

Official reference: Complete pipeline: reconcile ledger and bank files.

42 / 50

Complete pipeline: files larger than memory#

Scan stable inputs, validate small summaries, and stream accepted rows into partitioned output.

Scan source files, validate summary counts, and write accepted rows by day. The separate region summary stays small. Keep generated outputs out of input globs to avoid reprocessing them.

PythonA streaming pipeline with quality counts and partitioned output
from pathlib import Path
from tempfile import TemporaryDirectory
import polars as pl

with TemporaryDirectory() as folder:
    root = Path(folder)
    source = root / "source"
    source.mkdir()
    for part in range(3):
        pl.DataFrame(
            {
                "day": ["2026-09-01", "2026-09-02", "2026-09-02"],
                "region": ["Dhaka", "Sylhet", "Dhaka"],
                "cents": [100 + part, 200 + part, -10],
            }
        ).write_parquet(source / f"part-{part}.parquet")
    raw = pl.scan_parquet(str(source / "part-*.parquet"))
    if dict(raw.collect_schema()) != {
        "day": pl.String,
        "region": pl.String,
        "cents": pl.Int64,
    }:
        raise TypeError("unexpected input schema")
    checked = raw.with_columns(pl.col("day").str.to_date(strict=False).alias("date"))
    valid = (
        pl.col("date").is_not_null()
        & pl.col("region").is_not_null()
        & (pl.col("cents") >= 0)
    ).fill_null(False)
    counts = checked.select(
        pl.len().alias("input"),
        valid.sum().alias("accepted"),
        (~valid).sum().alias("rejected"),
    ).collect(engine="streaming")
    accepted = checked.filter(valid).select("date", "region", "cents")
    rejected = checked.filter(~valid)
    rejected.sink_parquet(root / "rejected.parquet", engine="streaming")
    accepted.sink_parquet(
        pl.PartitionBy(
            root / "clean", key="date", include_key=True, max_rows_per_file=100_000
        ),
        mkdir=True,
        engine="streaming",
    )
    summary = accepted.group_by("region").agg(pl.col("cents").sum()).sort("region")
    summary.sink_parquet(root / "summary.parquet", engine="streaming")
    print("quality:", counts.to_dicts())
    print("summary:", pl.read_parquet(root / "summary.parquet").to_dicts())
    print("written partitions:", len(list((root / "clean").rglob("*.parquet"))) > 0)
    assert counts.item(0, "input") == counts.item(0, "accepted") + counts.item(
        0, "rejected"
    )
TextExpected output
quality: [{'input': 9, 'accepted': 6, 'rejected': 3}]
summary: [{'region': 'Dhaka', 'cents': 303}, {'region': 'Sylhet', 'cents': 603}]
written partitions: True
Memory and publication

PartitionBy is unstable. Test its layout and peak memory with representative files, partition counts, and operations. Use immutable inputs and a new output directory for each run. Verify all files before publishing a manifest; file sinks do not form a transaction across outputs.

Official reference: Complete pipeline: files larger than memory.