Output Storage

Offload large task and workflow outputs to external storage.

Offloads task and workflow outputs to an external store instead of keeping them inline in the event log. task.with_options(output_storage=...) and the matching workflow.with_options(...) route results — here a pandas DataFrame — through a LocalFileStorage backend. Reach for this when outputs are large enough that inlining them would bloat the execution record.

Run it

python examples/output_storage.py

from __future__ import annotations

import pandas as pd

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow
from flux.output_storage import LocalFileStorage


file_storage = LocalFileStorage()


@task.with_options(output_storage=file_storage)
async def load_data(file_name: str) -> pd.DataFrame:
    return pd.read_csv(file_name)


@workflow.with_options(output_storage=file_storage)
async def output_storage(ctx: ExecutionContext[str]):
    if not ctx.input:
        raise TypeError("Input not provided")
    data = await load_data(ctx.input)
    return data


if __name__ == "__main__":  # pragma: no cover
    ctx = output_storage.run("examples/data/sample.csv")
    print(ctx.to_json())

A single LocalFileStorage instance is shared by both the load_data task and the output_storage workflow. With output_storage set, Flux writes the result to the backend and stores only a reference in the event log; the value is rehydrated on read. Swap LocalFileStorage for an object-store backend to keep the same workflow code in a distributed setup.