Reduce & checkpoint

Fold the whole stream into one value, or a few items

.reduce(acc, reducer) folds the stream with reducer(item, acc) and emits a single item, the final accumulator, when the stream ends. .wait_and_reduce(acc, reducer) does the same and returns the value directly (blocking). Mind the argument order: item first, accumulator second.

Signature

Pipeline.reduce(accumulator, reducer) -> Pipeline[OutPacket]
Pipeline.wait_and_reduce(accumulator, reducer) -> OutPacket
Pipeline.checkpoint(initial, reducer, emitter) -> Pipeline[OutPacket]
# reducer(item, accumulator) -> accumulator

python

from olympipe import Pipeline

def add(item: int, total: int) -> int:   # reducer(item, accumulator)
    return total + item

# Blocking: returns the final accumulator directly
print(Pipeline(range(101)).wait_and_reduce(0, add))

# As a step: emits a single item when the stream ends
print(Pipeline(range(101)).reduce(0, add).task(lambda t: t * 2).wait_for_result())

Output

5050
[10100]

How it works

The reduction runs in a single process that keeps the accumulator in memory. If an error occurs during the reduction, the partial accumulator is not emitted: the pipeline fails with PipelineExecutionError.

checkpoint: accumulate a state, then emit several items

.checkpoint(initial, reducer, emitter) updates a state with reducer(item, state) (which must return the state), then, once the stream ends, calls emitter(state) and emits each element of the returned iterable. Handy for a top-N, a histogram or deduplication.

python

from collections import Counter
from olympipe import Pipeline

def count_word(word: str, counts: Counter) -> Counter:
    counts[word] += 1
    return counts            # the reducer must return the state

def top_words(counts: Counter) -> list[tuple[str, int]]:
    return counts.most_common(2)   # emitted items, one by one

words = "a b a c b a d".split()
print(
    Pipeline(words)
    .checkpoint(Counter(), count_word, top_words)
    .wait_for_result()
)

Output

[('a', 3), ('b', 2)]

Previous Explode Next Limit & time budget