Split & Gather
Branch your pipeline into independent streams, merge them back
.split(fn, n=2) routes each item to 0-N branches based on fn's return value. .gather(*pipes) merges N pipelines into one stream. Ideal for processing different item categories with distinct logic.
Signature
Pipeline.split(split_function, n=None) -> tuple[Pipeline, ...]
Pipeline.gather(*others: Pipeline) -> Pipelinefrom typing import Optional, Tuple
from olympipe import Pipeline
def route_by_type(item: dict) -> Tuple[Optional[dict], Optional[dict]]:
if item["type"] == "image":
return (item, None) # → image branch
return (None, item) # → text branch
images, texts = Pipeline(mixed_dataset).split(route_by_type, n=2)
results = (
images.task(process_image, count=4) # image pipeline
.gather(texts.task(process_text, count=2)) # + text pipeline
.wait_for_result()
)⚡ Performance
5 000 mixed items, differentiated processing
How it works
Each branch is an independent Pipeline with its own queue. None in the return value means "don't send to this branch". gather connects every pipeline to a shared output queue: items are merged as they arrive, with no ordering guarantee.
split without n, and 3 or 4 branches
Signature: .split(split_function, n=None). If n is omitted it is inferred from the return annotation (Tuple[...] with 2, 3, 4 or more elements); without an annotation, split raises ValueError. The function returns a tuple of n values: each non-None value goes to the matching branch.
from typing import Optional, Tuple
from olympipe import Pipeline
Route = Tuple[Optional[int], Optional[int], Optional[int]]
def by_remainder(x: int) -> Route:
out: list = [None, None, None]
out[x % 3] = x
return tuple(out)
r0, r1, r2 = Pipeline(range(9)).split(by_remainder) # 3 branches, n inferred
print(r0.wait_for_results([r1, r2]))
def no_annotation(x):
return (x, None)
try:
Pipeline(range(3)).split(no_annotation)
except ValueError as e:
print("ValueError:", e)[[0, 3, 6], [1, 4, 7], [2, 5, 8]]
ValueError: Cannot infer number of outputs for 'no_annotation'. Add a return annotation (e.g. Tuple[A, B]) or pass n= explicitly.