Réduction & checkpoint

Agréger tout le flux en une valeur, ou en quelques items

.reduce(acc, reducer) replie le flux avec reducer(item, acc) et émet un seul item, l'accumulateur final, quand le flux se termine. .wait_and_reduce(acc, reducer) fait la même chose et renvoie directement la valeur (bloquant). Attention à l'ordre des arguments : l'item d'abord, l'accumulateur ensuite.

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

# Bloquant : renvoie directement l'accumulateur final
print(Pipeline(range(101)).wait_and_reduce(0, add))

# Comme étape : émet un seul item à la fin du flux
print(Pipeline(range(101)).reduce(0, add).task(lambda t: t * 2).wait_for_result())

Sortie

5050
[10100]

Comment ça fonctionne

La réduction tourne dans un seul processus, qui garde l'accumulateur en mémoire. Si une erreur survient pendant la réduction, l'accumulateur partiel n'est pas émis : le pipeline échoue avec PipelineExecutionError.

checkpoint : accumuler un état puis émettre plusieurs items

.checkpoint(initial, reducer, emitter) met à jour un état avec reducer(item, state) (qui doit renvoyer l'état), puis, une fois le flux terminé, appelle emitter(state) et émet chaque élément de l'itérable renvoyé. Pratique pour un top-N, un histogramme ou une déduplication.

python

from collections import Counter
from olympipe import Pipeline

def count_word(word: str, counts: Counter) -> Counter:
    counts[word] += 1
    return counts            # le reducer doit renvoyer l'état

def top_words(counts: Counter) -> list[tuple[str, int]]:
    return counts.most_common(2)   # items émis un par un

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

Sortie

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

Précédent Explode Suivant Limite & budget de temps