Batching
Group items to saturate your GPU or database
.batch(n) accumulates n items before emitting them as a list. .temporal_batch(s) groups everything arriving within s seconds — perfect for real-time streams (audio, sensors, WebSocket).
Signature
Pipeline.batch(batch_size=2, keep_incomplete_batch=True) -> Pipeline[list[InPacket]]
Pipeline.temporal_batch(time_interval: float) -> Pipeline[list[InPacket]]python
from olympipe import Pipeline
import time
# Fixed batch: group 64 items before processing
results = (
Pipeline(image_paths) # 10 000 images
.task(load_image, count=4) # parallel loading
.batch(64) # group into batches
.task(model.predict) # GPU inference
.wait_for_result()
)
# Temporal batch: 500ms sliding window
results = (
Pipeline(sensor_stream)
.task(parse_frame)
.temporal_batch(0.5) # everything arriving within 500ms
.task(aggregate)
.wait_for_result()
)⚡ Performance
10 000 images, GPU inference
No batching 50.0s
batch=16 12.0s
batch=64 3.2s
batch=128 2.1s
🚀 23.8× faster
How it works
Without batching, each item triggers a separate GPU round-trip. With a batch of 64, a single call processes 64 images — the fixed GPU latency is amortized over 64× more work.
keep_incomplete_batch
Signature: .batch(batch_size=2, keep_incomplete_batch=True). By default the last, incomplete batch is emitted at the end of the stream; with keep_incomplete_batch=False it is dropped (useful when the next step requires a fixed size).
python
from olympipe import Pipeline
print(Pipeline(range(7)).batch(3).wait_for_result())
print(Pipeline(range(7)).batch(3, keep_incomplete_batch=False).wait_for_result())Output
[[0, 1, 2], [3, 4, 5], [6]]
[[0, 1, 2], [3, 4, 5]]