From a6c8c342b755c5479aed96408e3c6f0fde90cc73 Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Tue, 7 Jan 2020 14:41:50 -0800 Subject: [PATCH] Better document guarantees provided by par iter API (#6726) * update * Update doc/source/iter.rst Co-Authored-By: Edward Oakes * Update doc/source/iter.rst Co-Authored-By: Edward Oakes Co-authored-by: Edward Oakes --- doc/source/iter.rst | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/doc/source/iter.rst b/doc/source/iter.rst index ecff24480..42bd6ec2c 100644 --- a/doc/source/iter.rst +++ b/doc/source/iter.rst @@ -97,6 +97,42 @@ each shard should only be read by one process at a time: >>> ray.get([do_sum.remote(s) for s in it.shards()]) [5552778, 16661667, 27780555] +Semantic Guarantees +~~~~~~~~~~~~~~~~~~~ + +The parallel iterator API guarantees the following semantics: + +**Fetch ordering**: When using ``it.gather_sync().foreach(fn)`` or +``it.gather_async().foreach(fn)`` (or any other transformation after a gather), +``fn(x_i)`` will be called on the element ``x_i`` before the next +element ``x_{i+1}`` is fetched from the source actor. This is useful if you need to +update the source actor between iterator steps. Note that for async gather, this +ordering only applies per shard. + +**Operator state**: Operator state is preserved for each shard. This means that you can pass a stateful callable to ``.foreach()``: +For example, this means you can pass a stateful callable to ``.foreach()``: + +.. code-block:: python + + class CumulativeSum: + def __init__(self): + self.total = 0 + + def __call__(self, x): + self.total += x + return (self.total, x) + + it = ray.experimental.iter.from_range(5, 1) + for x in it.for_each(CumulativeSum()).gather_sync(): + print(x) + + ## This prints: + #(0, 0) + #(1, 1) + #(3, 2) + #(6, 3) + #(10, 4) + Example: Streaming word frequency count ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~