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])	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, fuse_subprocess_stages: 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.

• fuse_subprocess_stages –

  If True, fuse runs of two or more adjacent pipe stages that share the same process-pool (or interpreter-pool) executor instance into a single stage that runs the run as one nested pipeline inside a worker pool. This eliminates the inter-stage IPC that otherwise round-trips data back to this process between each stage (so intermediate values need not be picklable), while each fused stage keeps its own concurrency and per-stage stats. A path_variants stage whose branches all use the same pool executor is fused too (router and branches move into the worker), and fuses on its own. An aggregate/disaggregate between pool stages is not fused and runs in the main process with its usual batching. Default: False.

  Added in version 0.6.0: The fuse_subprocess_stages argument.

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) → PipelineBuilder[T, U]

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

Parameters:

• router – A callable that takes an item and returns an int index selecting which path the item should be routed to.

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

• name – Optional name for the stage.

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.

  When op is already async, executor must be an isolating-pool (process or interpreter) executor and is not used to run the op – an async op always runs on the event loop. It serves only as a subprocess fusion-group tag: with fuse_subprocess_stages=True (see build()), adjacent stages sharing the same executor instance are fused into one worker sub-pipeline, so tagging an async op lets it join such a run (it then runs on the worker’s own event loop and, like any fused stage, must be picklable). Unfused, the tag is ignored and the op runs in the main process. Passing a non-isolating executor (e.g. a thread pool) with an async op is an error.

  Changed in version 0.6.0: An async op may now be given an isolating-pool executor as a subprocess fusion-group tag; previously any executor on an async op was rejected.

• 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.

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.

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.

Added in version 0.6.0.