spdl.pipeline.PipelineBuilder

class PipelineBuilder

Build Pipeline object.

Note

PipelineBuilder supports only chain of operations. If you need to build a pipeline composed of multiple sub-pipelines, use PipelineConfig.

See also

Building and Running Pipeline

Explains the basic usage of PipelineBuilder and Pipeline.

⚠ Caveats ⚠

Lists known anti-patterns that can cause a deadlock.

Pipeline Parallelism

Covers how to switch (or combine) multi-threading and multi-processing in detail.

Example: Pipeline definitions

Illustrates how to build a complex pipeline that PipelineBuilder does not support.

Methods

add_sink([buffer_size])	Attach a buffer to the end of the pipeline.
add_source(source, *[, continuous])	Attach an iterator to the source buffer.
aggregate(input, /, *[, drop_last])	Buffer the items in the pipeline.
build(*, num_threads[, max_failures, ...])	Build the pipeline.
disaggregate()	Disaggregate the items in the pipeline.
get_config()	Get the pipeline configuration.
path_variants(router, paths[, name, batched])	Route items to different processing paths based on a router function.
pipe(op, /, *[, concurrency, executor, ...])	Apply an operation to items in the pipeline.
to(target, /)	[Experimental] Designate where the subsequent stages execute.

add_sink(buffer_size: int = 3) → PipelineBuilder[T, U]

Attach a buffer to the end of the pipeline.

Parameters:

buffer_size – The size of the buffer. Pass 0 for unlimited buffering.

add_source(source: Iterable[T] | AsyncIterable[T], *, continuous: bool = False) → PipelineBuilder[T, U]

Attach an iterator to the source buffer.

Parameters:

• source –

  A lightweight iterator that generates data.

  Warning

  The source iterator must be lightweight as it is executed in async event loop. If the iterator performs a blocking operation, the entire pipeline will be blocked.

• continuous – If True, the source continuously re-iterates, injecting a sentinel object representing an epoch boundary between iterations. This enables multi-epoch pipeline reuse without rebuilding. Use is_epoch_end() to detect epoch boundaries in custom merge operations if needed; regular pipe stage functions do not need to handle it.

aggregate(input: int | Aggregator, /, *, drop_last: bool = False) → PipelineBuilder[T, U]

Buffer the items in the pipeline.

Parameters:

• input –

  Either an integer specifying the number of items to buffer, or an Aggregator instance for custom aggregation logic.

  • If int: Buffers that many items before emitting. It uses Collate aggregator class.

  • If Aggregator: Custom aggregation using the accumulate() and flush() methods.

• drop_last –

  Drop the last aggregation if incomplete. - When drop_last=False (default): Calls flush()

  at EOF

  • When drop_last=True: Does NOT call flush(), dropping incomplete batches

build(*, num_threads: int, max_failures: int | Fraction = -1, report_stats_interval: float = -1, queue_class: type[AsyncQueue] | None = None, task_hook_factory: Callable[[StageInfo], list[TaskHook]] | None = None, stage_id: int = 0, use_thread_output_queue: bool = False) → Pipeline[U]

Build the pipeline.

Parameters:

• num_threads – The number of threads in the thread pool attached to async event loop.

• max_failures – The maximum number (int) or rate (Fraction) of failures each pipe stage can have before the pipeline is halted. When an int is provided, it specifies the maximum count of failures. Setting -1 (default) disables it. When a Fraction is provided (e.g., Fraction(1, 10) for 10%), it specifies the maximum failure rate (failures / invocations).

• report_stats_interval –

  When provided, report the pipeline performance stats every given interval. Unit: [sec]

  This is only effective if there is no custom hook or custom AsyncQueue provided for stages. The argument is passed to TaskStatsHook and StatsQueue.

  If a custom stage hook is provided and stats report is needed, you can instantiate TaskStatsHook and include it in the hooks provided to PipelineBuilder.pipe().

  Similarly if you are providing a custom AsyncQueue class, you need to implement the same logic by your self.

• queue_class – If provided, override the queue class used to connect stages. Must be a class (not an instance) inherits AsyncQueue.

• task_hook_factory – If provided, used to create task hook objects, given a name of the stage. If None, a default hook, TaskStatsHook is used. To disable hooks, provide a function that returns an empty list.

• stage_id – The index of the initial stage used for logging.

• use_thread_output_queue – If True, replace the sink’s output queue with a queue.Queue-backed queue for lower-latency batch handoff. Default: False.

See also

to()

Run a region of stages in a subprocess (or subinterpreter) worker pool.

disaggregate() → PipelineBuilder[T, U]

Disaggregate the items in the pipeline.

get_config() → PipelineConfig[U]

Get the pipeline configuration.

Returns:

A PipelineConfig object representing the current pipeline configuration.

Raises:

• RuntimeError – If source or sink is not set, or a subinterpreter region is used on Python < 3.14.

• ValueError – If an execution region opened by to() is not closed with to(MAIN_PROCESS) before the sink, or a stage inside a region uses output_order="input".

path_variants(router: Callable, paths: Sequence, name: str | None = None, batched: bool = False) → PipelineBuilder[T, U]

Route items to different processing paths based on a router function.

Added in version 0.6.0: The batched argument.

Parameters:

• router – A callable that selects the path for each input. In per-item mode (default) it takes an item and returns an int index. In batched mode it takes a batch (list) and returns one int index per element (same length as the batch).

• paths – A sequence of paths, where each path is a sequence of pipe configs.

• name – Optional name for the stage.

• batched – If True, route whole batches instead of single items: the batch is partitioned into per-path sub-batches (each path op takes and returns a list) and merged back into one batch. Aggregate the source into batches upstream of this stage. See PathVariants().

pipe(op: Callable[[T_], U_] | Callable[[T_], Iterable[U_]] | Callable[[T_], Awaitable[U_]] | Callable[[T_], AsyncIterable[U_]] | SupportsGetItem[T_, U_] | Mapping[T_, U_] | Sequence[T_], /, *, concurrency: int = 1, executor: Executor | None = None, name: str | None = None, output_order: str = 'completion', max_failures: int | Fraction | None = None) → PipelineBuilder[T, U]

Apply an operation to items in the pipeline.

Parameters:

• op –

  A function, callable or container with __getitem__ method (such as dict, list and tuple). If it’s function or callable, it is inovked with the input from the input queue. If it’s container type, the input is passed to __getitem__ method.

  The function or callable must take exactly one argument, which is the output from the upstream. If passing around multiple objects, take them as a tuple or use dataclass and define a custom protocol.

  If the result of applying op to an input item is None, the pipeline skips absorb the result and it won’t be propagated to the downstream stages.

  Optionally, the op can be a generator function, async function or async generator function.

  If op is (async) generator, the items yielded are put in the output queue separately.

  Warning

  If op is synchronous geneartor, and executor is an instance of concurrent.futures.ProcessPoolExecutor, the output items are not put in the output queue until the generator is exhausted.

  Async generator, or synchronous generator without ProcessPoolExecutor does not have this issue, and the yielded items are put in the output queue immediately.

  Tip

  When passing an async op, make sure that the op does not call sync function inside. If calling a sync function, use asyncio.loop.run_in_executor() or asyncio.to_thread() to delegate the execution to the thread pool.

• concurrency – The maximum number of async tasks executed concurrently.

• executor –

  A custom executor object to be used to convert the synchronous operation into asynchronous one. If None, the default executor is used.

  It is invalid to provide this argument when the given op is already async.

• name – The name (prefix) to give to the task.

• output_order – If "completion" (default), the items are put to output queue in the order their process is completed. If "input", then the items are put to output queue in the order given in the input queue.

• max_failures – The maximnum number (int) or rate (Fraction) of failures allowed before the pipe operation is considered failure and the whole Pipeline is shutdown. When an int is provided, it specifies the maximum count of failures. When a Fraction is provided (e.g., Fraction(1, 10) for 10%), it specifies the maximum failure rate (failures / invocations). This overrides the value provided to the build() method.

to(target: ProcessPoolExecutorConfig | InterpreterPoolExecutorConfig | _MainProcess, /) → PipelineBuilder[T, U]

[Experimental] Designate where the subsequent stages execute.

Added in version 0.6.0.

Opens (or closes) an execution region: every stage added after this call runs on target until the next to(). A pipeline starts on the main process, so a region is opened by to(ProcessPoolExecutorConfig(...)) or to(InterpreterPoolExecutorConfig(...)) and closed by to(MAIN_PROCESS). The stages inside a region are fused into one nested pipeline that runs together in a worker process (or subinterpreter), so the value handed from one stage to the next stays in the worker — it is not copied back to the main process between stages and need not be picklable. Only the region’s inputs and outputs cross the boundary.

Unlike passing executor= to individual pipe() calls, a region also carries aggregate(), disaggregate(), and path_variants() stages into the worker, and gives the worker-pool configuration a single home.

Each stage’s concurrency applies within each worker, so a stage’s effective concurrency across the pool is concurrency × max_workers. For example, pipe(op, concurrency=2) in a pool with max_workers=4 runs up to 8 invocations of op at once. Size each stage’s concurrency together with max_workers to stay within your CPU budget.

Parameters:

target –

Where the following stages run.

• ProcessPoolExecutorConfig — a pool of worker processes.

• InterpreterPoolExecutorConfig — a pool of subinterpreters (Python 3.14+; the region’s ops must avoid NumPy/PyTorch, which cannot be imported in a subinterpreter).

• MAIN_PROCESS — close the current region; the following stages run in the main process.

A live Executor is not accepted — pass a spec so the pipeline stays expressible as static config. To run a single stage on a custom executor, use pipe(executor=...) instead.

Note

The region must be closed with to(MAIN_PROCESS) before add_sink(), a stage inside a region may not use output_order="input" (order cannot be preserved across independent workers), and a subinterpreter region requires Python 3.14+. These are checked when the pipeline is built.