Side effects

A pipeline often ends by saving data, recording metrics, or printing a report. These actions are side effects.

Use @side_effect to mark a transform that returns None and does a side effect:

@side_effect
def write_output(df: Annotated[DataFrame, Project(OutputSchema)], path: str) -> None:
    df.write.parquet(path)

@side_effect
def log_ml_model(model: LinearRegression, experimet_id: str) -> None:
    mlflow.log_model(model, experiment_id)

Running without side effects

You might want to run a pipeline without it's side effects to inspect the intermediate results. Or if you a running in an environment where you don't have write permissions, you might want to skip the steps that write to disk. Pass include_side_effects=False when running pipelines to run all steps leading up to side effects:

from data_joinery import Pipeline, side_effect, transform


@transform
def make_report() -> str:
    return "3 records"


@side_effect
def print_report(report: str) -> None:
    print(report)


pipeline = Pipeline()
report = pipeline.add_step(make_report)
printer = pipeline.add_step(print_report)
report >> printer

quiet_result = pipeline.run(include_side_effects=False)
print("Quiet run:", quiet_result.get_output(report))

pipeline.run()
print_report("Direct call")

Outputs:

Quiet run: 3 records
3 records
Direct call

In the first run, the pipeline creates the report but skips print_report. The second run prints the report as normal.

Running without side effects can also be a great CICD check to ensure that your pipeline will run without mutating the state of your environment.

Polars Lazy Sinks

Polars supports lazy sinks. With lazy=True, each write step returns a query plan instead of writing a file. Mark only the step that calls pl.collect_all as a side effect to run both sinks together:

from pathlib import Path
from tempfile import TemporaryDirectory

import polars as pl

from data_joinery import Pipeline, side_effect, transform

with TemporaryDirectory() as directory:
    parquet_path = Path(directory) / "orders.parquet"
    ipc_path = Path(directory) / "orders.ipc"

    @transform
    def make_orders() -> pl.LazyFrame:
        return pl.LazyFrame({"order_id": [1, 2], "amount": [10, 25]})

    @transform
    def write_parquet(orders: pl.LazyFrame) -> pl.LazyFrame:
        return orders.sink_parquet(parquet_path, lazy=True)

    @transform
    def write_ipc(orders: pl.LazyFrame) -> pl.LazyFrame:
        return orders.sink_ipc(ipc_path, lazy=True)

    @side_effect
    def collect_sinks(parquet: pl.LazyFrame, ipc: pl.LazyFrame) -> None:
        pl.collect_all([parquet, ipc])
        print(f"Wrote {parquet_path.name} and {ipc_path.name}")

    pipeline = Pipeline()
    orders = pipeline.add_step(make_orders)
    parquet = pipeline.add_step(write_parquet)
    ipc = pipeline.add_step(write_ipc)
    collect = pipeline.add_step(collect_sinks)
    orders >> parquet
    orders >> ipc
    pipeline.connect(parquet, collect, param="parquet")
    pipeline.connect(ipc, collect, param="ipc")
    pipeline.run()

Outputs:

Wrote orders.parquet and orders.ipc