Recipes¶
Short solutions to common tasks. The recipes build on each other and start from:
>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import ParquetDal
>>> data_dir = Path("data")
>>> dal = ParquetDal(data_dir)
Keep a snapshot of a full refresh¶
When a source is reloaded in full, write each load as a new version. Readers always get the newest complete snapshot, and earlier ones stay on disk.
>>> for day, rows in [("2026-10-22", 3), ("2026-10-23", 5)]:
... snapshot = pl.DataFrame({"day": [day] * rows, "n": range(rows)})
... _ = dal.write(snapshot, "inventory")
>>> dal.read("inventory").collect().height
5
Ingest data in batches¶
When new rows arrive over time, append each batch. Each batch becomes its own file, so a failed batch never damages earlier ones.
>>> _ = dal.write(pl.DataFrame({"event": ["start"], "ms": [0]}), "events")
>>> for batch in ([("click", 120)], [("click", 340), ("stop", 900)]):
... _ = dal.append(pl.DataFrame(batch, schema=["event", "ms"], orient="row"), "events")
>>> dal.read("events").collect()["event"].to_list()
['start', 'click', 'click', 'stop']
To start a profile that may not exist yet, append falls back
to a write and warns. Write the first batch explicitly to avoid the warning.
Query without loading everything¶
read is lazy. Filter and select before collect so polars
only reads the rows and columns it needs.
>>> (
... dal.read("events")
... .filter(pl.col("event") == "click")
... .select(pl.col("ms").mean())
... .collect()
... .item()
... )
230.0
Filter by column values¶
For the common case of keeping rows where columns equal given values,
SimpleIo takes the filters as keyword arguments to
read. It stores parquet, like ParquetDal,
so it reads the same data directory. None matches nulls, and a misspelt column
name raises straight away rather than at collect.
>>> from pyrosome import SimpleIo
>>> io = SimpleIo(data_dir)
>>> io.read("events", event="click").collect()["ms"].to_list()
[120, 340]
>>> io.read("events", evnt="click")
Traceback (most recent call last):
...
polars.exceptions.ColumnNotFoundError: No column named evnt in events
Floats can’t be filtered on, because float equality is unreliable. Use
filter with a tolerance for those.
write and append also take pandas
data frames. Named index levels, such as the keys of a groupby, are stored as
columns, and an unnamed index is dropped. Converting pandas data needs the
pandas extra, which installs pandas and pyarrow.
>>> import pandas as pd
>>> views = pd.DataFrame({"page": ["home", "cart", "home"], "ms": [50, 70, 90]})
>>> io.write(views[views["ms"] > 60], "views").name
'v0'
>>> io.read("views", page="home").collect().rows()
[('home', 90)]
Group profiles into a dataset¶
A Dataset reads the profiles below the same keys with the same filters,
so every profile in it must have the filtered columns.
get_subset narrows it further.
>>> from pyrosome import Dataset
>>> orders = pl.DataFrame({"region": ["eu", "us", "eu"], "amount": [1, 2, 3]})
>>> _ = io.write(orders, "shop", "orders")
>>> eu = Dataset(io, "shop", region="eu")
>>> eu.get_profile("orders").collect()["amount"].to_list()
[1, 3]
>>> eu.get_subset(amount=3).get_profile("orders").collect()["amount"].to_list()
[3]
To derive a profile from others, subclass Dataset and add a
_calc_<profile> method returning its data. The first
get_cache_profile stores it, calculated from every row rather
than just the dataset’s subset, and later calls read the stored copy. It isn’t
updated when the profiles it’s calculated from change: pass refresh=True to
recalculate it.
>>> class Shop(Dataset):
... def _calc_totals(self):
... orders = self.get_profile("orders")
... return orders.group_by("region").agg(pl.col("amount").sum())
>>> eu = Shop(io, "shop", region="eu")
>>> eu.get_cache_profile("totals").collect().rows()
[('eu', 4)]
>>> _ = io.append(pl.DataFrame({"region": ["eu"], "amount": [10]}), "shop", "orders")
>>> eu.get_cache_profile("totals").collect().rows()
[('eu', 4)]
>>> eu.get_cache_profile("totals", refresh=True).collect().rows()
[('eu', 14)]
Read data that may not exist yet¶
read raises FileNotFoundError when a profile has no data.
>>> def read_or_empty(dal, schema, *keys):
... try:
... return dal.read(*keys)
... except FileNotFoundError:
... return pl.LazyFrame(schema=schema)
>>> read_or_empty(dal, {"user": pl.String}, "users").collect().shape
(0, 1)
Combine several profiles¶
Use keys to split data, for example by region, then concatenate the lazy frames to query them together.
>>> for region, total in [("eu", 10.0), ("us", 4.5)]:
... _ = dal.write(pl.DataFrame({"total": [total]}), "sales", region)
>>> combined = pl.concat(
... dal.read("sales", region).with_columns(region=pl.lit(region))
... for region in ["eu", "us"]
... )
>>> combined.collect().sort("region")["region"].to_list()
['eu', 'us']
Compact many small appends¶
Many small appended files make reads slower. Rewrite the profile as a new version with one file. The lazy read streams straight into the new version.
>>> compacted = dal.write(dal.read("events"), "events")
>>> sorted(path.name for path in compacted.iterdir())
['0.parquet']
>>> dal.read("events").collect().height
4
Change the schema¶
append rejects data with different columns. To add a column,
write a new version that includes it.
>>> try:
... dal.append(pl.DataFrame({"event": ["start"], "ms": [5], "user": ["ana"]}), "events")
... except ValueError as error:
... print("rejected")
rejected
>>> widened = dal.read("events").with_columns(user=pl.lit(None, dtype=pl.String))
>>> _ = dal.write(widened, "events")
>>> _ = dal.append(pl.DataFrame({"event": ["start"], "ms": [5], "user": ["ana"]}), "events")
>>> dal.read("events").collect().columns
['event', 'ms', 'user']
Read an older version¶
Old versions are plain directories, so scan them directly.
>>> profile_dir = data_dir / "inventory"
>>> pl.scan_parquet(profile_dir / "v0" / "*.parquet").collect().height
3
Roll back to an older version¶
Write the old version again as a new version. History is kept, and the rollback can itself be undone.
>>> _ = dal.write(pl.scan_parquet(profile_dir / "v0" / "*.parquet"), "inventory")
>>> dal.read("inventory").collect().height
3
Remove old versions¶
Delete version directories older than the last few. Only do this when no reader
can still be using them, since a lazy frame from an earlier
read fails if its files disappear.
>>> import shutil
>>> import stat
>>> from pyrosome.dal import DataManifestFile
>>> def remove_read_only(remove, path, _error):
... # data files are read only, and Windows can't delete read-only files
... Path(path).chmod(stat.S_IWRITE)
... remove(path)
>>> def prune(dal, *keys, keep=2):
... profile_dir = dal.data_directory.joinpath(*keys)
... latest = DataManifestFile(profile_dir).get_latest_data_dir()
... oldest_kept = int(latest.name.removeprefix("v")) - keep + 1
... for version in profile_dir.glob("v*"):
... index = version.name.removeprefix("v")
... if version.is_dir() and index.isdigit() and int(index) < oldest_kept:
... shutil.rmtree(version, onexc=remove_read_only)
>>> prune(dal, "inventory", keep=2)
>>> sorted(path.name for path in profile_dir.iterdir())
['manifest.json', 'v1', 'v2']
Versions newer than the manifest’s latest are kept, because another process may still be writing them.
Write from several processes¶
Writes and appends are safe from several processes at once, so a process pool can append batches to the same profile directly.
Worker functions must be importable by the pool, so this recipe is a script rather than an interactive session:
"""Append batches to one profile from a pool of processes."""
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
import polars as pl
from pyrosome import ParquetDal
DATA_DIR = Path("data")
def ingest(batch_number: int) -> Path:
"""Append one batch of metrics.
Args:
batch_number: The batch to generate and append.
Returns:
The file the batch was appended as.
"""
batch = pl.DataFrame({"batch": [batch_number], "value": [batch_number * 10]})
return ParquetDal(DATA_DIR).append(batch, "metrics")
if __name__ == "__main__":
ParquetDal(DATA_DIR).write(pl.DataFrame({"batch": [0], "value": [0]}), "metrics")
with ProcessPoolExecutor() as pool:
files = list(pool.map(ingest, range(1, 100)))
rows = ParquetDal(DATA_DIR).read("metrics").collect().height
print(f"{len(files)} batches appended as {len(set(files))} files, {rows} rows")
Running it prints:
99 batches appended as 99 files, 100 rows
Exchange data as CSV¶
Convert between formats by reading with one class and writing with the other. Keep each format in its own data directory; see the limitations in Concepts.
>>> from pyrosome import CsvDal
>>> csv_dal = CsvDal(data_dir / "export")
>>> export_dir = csv_dal.write(dal.read("sales", "eu"), "sales", "eu")
>>> (export_dir / "0.csv").read_text().splitlines()
['total', '10.0']
>>> _ = dal.write(csv_dal.read("sales", "eu"), "sales", "eu_from_csv")
Recover from a crashed writer¶
If a process dies while updating a manifest, its lock file stays behind and
later writes raise FileAlreadyLockedError after
LOCK_TIMEOUT seconds. Once you are sure no
writer is running, remove the lock file.
>>> from pyrosome.simple_file_lock import file_is_locked
>>> manifest_path = data_dir / "events" / "manifest.json"
>>> lock_file = manifest_path.with_name("manifest.json.write_lock")
>>> lock_file.touch() # simulate a crashed writer
>>> file_is_locked(manifest_path)
True
>>> lock_file.unlink(missing_ok=True)
>>> file_is_locked(manifest_path)
False