First: Emit a row containing only the first value from a side, if any.
Last: Emit a row containing only the last value from a side, if any.
Product: Emit a row with every combination of values for all sides. This is similar to a SQL cross join.

JoinEmitMode: TypeAlias#

When should a join emit rows downstream.

Complete: Emit once a value has been seen from each side. Then discard the state.
Final: Emit when the upstream is EOF or the window closes. Then discard the state.

This mode only works on finite data streams and only returns a result once the upstream is EOF. You’ll need to use a different mode on infinite data.
Running: Emit every time a new value is seen on any side. Retain the state forever.

Classes#

class BranchOut#

Bases:

Generic[X, Y]

Streams returned from the branch operator.

trues: Stream[X]#

falses: Stream[Y]#

class StatefulBatchLogic#

Bases:

ABC
Generic[V, W, S]

Abstract class to define a stateful operator.

The operator will call these methods in order: on_batch once with all items queued, then on_notify if the notification time has passed, then on_eof if the upstream is EOF and no new items will be received this execution. If the logic is retained after all the above calls then notify_at will be called. snapshot is periodically called.

RETAIN: bool#

This logic should be retained after this returns.

If you always return this, this state will never be deleted and if your key-space grows without bound, your memory usage will also grow without bound.

DISCARD: bool#: This logic should be discarded immediately after this returns.

abstract on_batch( values: List[V], ) → Tuple[Iterable[W], bool]#

Called on each new upstream item.

This will be called multiple times in a row if there are multiple items from upstream.

Parameters:: value – The value of the upstream (key, value).
Returns:: A 2-tuple of: any values to emit downstream and wheither to discard this logic. Values will be wrapped in (key, value) automatically.

on_notify() → Tuple[Iterable[W], bool]#

Called when the scheduled notification time has passed.

Defaults to emitting nothing and retaining the logic.

Returns:: A 2-tuple of: any values to emit downstream and wheither to discard this logic. Values will be wrapped in (key, value) automatically.

on_eof() → Tuple[Iterable[W], bool]#

The upstream has no more items on this execution.

This will only be called once per awake after on_batch is called.

Defaults to emitting nothing and retaining the logic.

Returns:: 2-tuple of: any values to emit downstream and wheither to discard this logic. Values will be wrapped in (key, value) automatically.

notify_at() → Optional[datetime]#

Return the next notification time.

This will be called once right after the logic is built, and if any of the on_* methods were called if the logic was retained.

This must always return the next notification time. The operator only stores a single next time, so if there are a series of times you would like to notify at, store all of them but only return the soonest.

Defaults to returning None.

Returns:: Scheduled time. If None, no on_notify callback will occur.

abstract snapshot() → S#

Return a immutable copy of the state for recovery.

This will be called periodically by the runtime.

The value returned here will be passed back to the builder function of stateful when resuming.

The state must be pickle-able.

Danger

The state must be effectively immutable! If any of the other functions in this class might be able to mutate the state, you must copy.deepcopy or something equivalent before returning it here.

Returns:: The immutable state to be pickled.

class StatefulLogic#

Bases:

ABC
Generic[V, W, S]

Abstract class to define a stateful operator.

The operator will call these methods in order: on_item once for any items queued, then on_notify if the notification time has passed, then on_eof if the upstream is EOF and no new items will be received this execution. If the logic is retained after all the above calls then notify_at will be called. snapshot is periodically called.

RETAIN: bool#

This logic should be retained after this returns.

If you always return this, this state will never be deleted and if your key-space grows without bound, your memory usage will also grow without bound.

DISCARD: bool#: This logic should be discarded immediately after this returns.

abstract on_item( value: V, ) → Tuple[Iterable[W], bool]#

Called on each new upstream item.

This will be called multiple times in a row if there are multiple items from upstream.

Parameters:: value – The value of the upstream (key, value).
Returns:: A 2-tuple of: any values to emit downstream and wheither to discard this logic. Values will be wrapped in (key, value) automatically.

on_notify() → Tuple[Iterable[W], bool]#

Called when the scheduled notification time has passed.

Returns:: A 2-tuple of: any values to emit downstream and wheither to discard this logic. Values will be wrapped in (key, value) automatically.

on_eof() → Tuple[Iterable[W], bool]#

The upstream has no more items on this execution.

This will only be called once per execution after on_item is done being called.

Returns:: 2-tuple of: any values to emit downstream and wheither to discard this logic. Values will be wrapped in (key, value) automatically.

notify_at() → Optional[datetime]#

Return the next notification time.

This will be called once right after the logic is built, and if any of the on_* methods were called if the logic was retained.

This must always return the next notification time. The operator only stores a single next time, so if

Returns:: Scheduled time. If None, no on_notify callback will occur.

abstract snapshot() → S#

Return a immutable copy of the state for recovery.

This will be called periodically by the runtime.

The value returned here will be passed back to the builder function of stateful when resuming.

The state must be pickle-able.

Danger

The state must be effectively immutable! If any of the other functions in this class might be able to mutate the state, you must copy.deepcopy or something equivalent before returning it here.

Returns:: The immutable state to be pickled.

class TTLCache#

Bases:

Generic[DK, DV]

A simple TTL cache.

v_getter: Callable[[DK], DV]#

now_getter: Callable[[], datetime]#

ttl: timedelta#

get(k: DK) → DV#

Get the cached value for a key.

Will cache and return the updated value if TTL has expired.

Parameters:: k – Key.
Returns:: Value.

remove(k: DK) → None#

Remove the cached value for this key.

Parameters:: k – Key.

Functions#

branch( step_id: str, up: Stream[X], predicate: Callable[[X], bool], ) → BranchOut#

Divide items into two streams with a predicate.

import bytewax.operators as op
from bytewax.dataflow import Dataflow
from bytewax.testing import run_main, TestingSource

flow = Dataflow("branch_eg")
nums = op.input("nums", flow, TestingSource([1, 2, 3, 4, 5]))
b_out = op.branch("even_odd", nums, lambda x: x % 2 == 0)
evens = b_out.trues
odds = b_out.falses
_ = op.inspect("evens", evens)
_ = op.inspect("odds", odds)

branch_eg.odds: 1
branch_eg.evens: 2
branch_eg.odds: 3
branch_eg.evens: 4
branch_eg.odds: 5

Parameters:

step_id – Unique ID.
up – Stream to divide.
predicate –
Function to call on each upstream item. Items for which this returns True will be put into one branch stream; False the other branch stream.

If this function is a typing.TypeGuard, the downstreams will be properly typed.

Returns:

A stream of items for which the predicate returns True, and a stream of items for which the predicate returns False.

flat_map_batch( step_id: str, up: Stream[X], mapper: Callable[[List[X]], Iterable[Y]], ) → Stream[Y]#

Transform an entire batch of items 1-to-many.

The batch size received here depends on the exact behavior of the upstream input sources and operators. It should be used as a performance optimization when processing multiple items at once has much reduced overhead.

See also the batch_size parameter on various input sources.

See also the collect operator, which collects multiple items next to each other in a stream into a single list of them flowing through the stream.

Parameters:

step_id – Unique ID.
up – Stream.
mapper – Called once with each batch of items the runtime receives. Returns the items to emit downstream.

Returns:

A stream of each item returned by the mapper.

input( step_id: str, flow: Dataflow, source: Source[X], ) → Stream[X]#

Introduce items into a dataflow.

See bytewax.inputs for more information on how input works. See bytewax.connectors for a buffet of our built-in connector types.

Parameters:

step_id – Unique ID.
flow – The dataflow.
source – To read items from.

Returns:

A stream of items from the source. See your specific Source documentation for what kind of item that is.

This stream might be keyed. See your specific Source.

inspect_debug( step_id: str, up: Stream[X], inspector: Callable[[str, X, int, int], None] = _default_debug_inspector, ) → Stream[X]#

Observe items, their worker, and their epoch for debugging.

import bytewax.operators as op
from bytewax.testing import TestingSource
from bytewax.dataflow import Dataflow

flow = Dataflow("inspect_debug_eg")
s = op.input("inp", flow, TestingSource(range(3)))
_ = op.inspect_debug("help", s)

inspect_debug_eg.help W0 @1: 0
inspect_debug_eg.help W0 @1: 1
inspect_debug_eg.help W0 @1: 2

Parameters:

step_id – Unique ID.
up – Stream.
inspector – Called with the step ID, each item in the stream, the epoch of that item, and the worker processing the item. Defaults to printing out all the arguments.

Returns:

The upstream unmodified.

merge( step_id: str, *ups: Stream[Any], ) → Stream[Any]#

Combine multiple streams together.

Parameters:

step_id – Unique ID.
*ups – Streams.

Returns:

A single stream of the same type as all the upstreams with items from all upstreams merged into it unmodified.

output( step_id: str, up: Stream[X], sink: Sink[X], ) → None#

Write items out of a dataflow.

See bytewax.outputs for more information on how output works. See bytewax.connectors for a buffet of our built-in connector types.

Parameters:

step_id – Unique ID.
up – Stream of items to write. See your specific Sink documentation for the required type of those items.
sink – Write items to.

redistribute( step_id: str, up: Stream[X], ) → Stream[X]#

Redistribute items randomly across all workers.

Bytewax’s execution model has workers executing all steps, but the state in each step is partitioned across workers by some key. Bytewax will only exchange an item between workers before stateful steps in order to ensure correctness, that they interact with the correct state for that key. Stateless operators (like filter) are run on all workers and do not result in exchanging items before or after they are run.

This can result in certain ordering of operators to result in poor parallelization across an entire execution cluster. If the previous step (like a bytewax.operators.windowing.reduce_window or input with a FixedPartitionedSource) concentrated items on a subset of workers in the cluster, but the next step is a CPU-intensive stateless step (like a map), it’s possible that not all workers will contribute to processing the CPU-intesive step.

This operation has a overhead, since it will need to serialize, send, and deserialize the items, so while it can significantly speed up the execution in some cases, it can also make it slower.

A good use of this operator is to parallelize an IO bound step, like a network request, or a heavy, single-cpu workload, on a machine with multiple workers and multiple cpu cores that would remain unused otherwise.

A bad use of this operator is if the operation you want to parallelize is already really fast as it is, as the overhead can overshadow the advantages of distributing the work. Another case where you could see regressions in performance is if the heavy CPU workload already spawns enough threads to use all the available cores. In this case multiple processes trying to compete for the cpu can end up being slower than doing the work serially. If the workers run on different machines though, it might again be a valuable use of the operator.

Use this operator with caution, and measure whether you get an improvement out of it.

Once the work has been spread to another worker, it will stay on those workers unless other operators explicitely move the item again (usually on output).

Parameters:

step_id – Unique ID.
up – Stream.

Returns:

Stream unmodified.

stateful_batch( step_id: str, up: KeyedStream[V], builder: Callable[[Optional[S]], StatefulBatchLogic[V, W, S]], ) → KeyedStream[W]#

Advanced generic stateful operator.

This is the lowest-level operator Bytewax provides and gives you full control over all aspects of the operator processing and lifecycle. Usualy you will want to use a higher-level operator than this.

Subclass StatefulBatchLogic to define its behavior. See documentation there.

Parameters:

step_id – Unique ID.
up – Keyed stream.
builder – Called whenver a new key is encountered with the resume state returned from StatefulBatchLogic.snapshot for this key, if any. This should close over any non-state configuration and combine it with the resume state to return the prepared StatefulBatchLogic for the new key.

Returns:

Keyed stream of all items returned from StatefulBatchLogic.on_batch, StatefulBatchLogic.on_notify, and StatefulBatchLogic.on_eof.

stateful( step_id: str, up: KeyedStream[V], builder: Callable[[Optional[S]], StatefulLogic[V, W, S]], ) → KeyedStream[W]#

Advanced generic stateful operator.

This is a low-level operator Bytewax provides and gives you control over most aspects of the operator processing and lifecycle. Usualy you will want to use a higher-level operator than this. Also see stateful_batch for even more control.

Subclass StatefulLogic to define its behavior. See documentation there.

Parameters:

step_id – Unique ID.
up – Keyed stream.
builder – Called whenver a new key is encountered with the resume state returned from StatefulLogic.snapshot for this key, if any. This should close over any non-state configuration and combine it with the resume state to return the prepared StatefulLogic for the new key.

Returns:

Keyed stream of all items returned from StatefulLogic.on_item, StatefulLogic.on_notify, and StatefulLogic.on_eof.

collect( step_id: str, up: KeyedStream[V], timeout: timedelta, max_size: int, ) → KeyedStream[List[V]]#

Collect items into a list up to a size or a timeout.

See bytewax.operators.windowing.collect_window for more control over time.

Parameters:

step_id – Unique ID.
up – Stream of individual items.
timeout – Timeout before emitting the list, even if max_size was not reached.
max_size – Emit the list once it reaches this size, even if timeout was not reached.

Returns:

A stream of upstream items gathered into lists.

count_final( step_id: str, up: Stream[X], key: Callable[[X], str], ) → KeyedStream[int]#

Count the number of occurrences of items in the entire stream.

This only works on finite data streams and only return counts once the upstream is EOF. You’ll need to use bytewax.operators.windowing.count_window on infinite data.

Parameters:

step_id – Unique ID.
up – Stream of items to count.
key – Function to convert each item into a string key. The counting machinery does not compare the items directly, instead it groups by this string key.

Returns:

A stream of (key, count). Only once the upstream is EOF.

enrich_cached( step_id: str, up: Stream[X], getter: Callable[[DK], DV], mapper: Callable[[TTLCache[DK, DV], X], Y], ttl: timedelta = timedelta.max, _now_getter: Callable[[], datetime] = _get_system_utc, ) → Stream[Y]#

Enrich / join items using a cached lookup.

Use this if you’d like to join items in the dataflow with unsychronized pulled / polled data from an external service. This assumes that the joined data is static and will not emit updates. Since there is no integration with the recovery system, it’s possible that results will change across resumes if the joined data source changes.

The joined data is cached by a key you specify.

import bytewax.operators as op
from bytewax.dataflow import Dataflow
from bytewax.testing import TestingSource, run_main


def query_icon_url_service(code):
    if code == "dog_ico":
        return "http://domain.invalid/static/dog_v1.png"
    elif code == "cat_ico":
        return "http://domain.invalid/static/cat_v2.png"
    elif code == "rabbit_ico":
        return "http://domain.invalid/static/rabbit_v1.png"


flow = Dataflow("param_eg")
inp = op.input(
    "inp",
    flow,
    TestingSource(
        [
            {"user_id": "1", "avatar_icon_code": "dog_ico"},
            {"user_id": "3", "avatar_icon_code": "rabbit_ico"},
            {"user_id": "2", "avatar_icon_code": "dog_ico"},
        ]
    ),
)
op.inspect("check_inp", inp)


def icon_code_to_url(cache, msg):
    code = msg.pop("avatar_icon_code")
    msg["avatar_icon_url"] = cache.get(code)
    return msg


with_urls = op.enrich_cached(
    "with_url",
    inp,
    query_icon_url_service,
    icon_code_to_url,
)
op.inspect("check_with_url", with_urls)

If you have a join source which is push-based or need to emit updates when either side of the join changes, instead consider having that be a second input to the dataflow and using a running join. This reduces cache misses and startup overhead.

Each worker will keep a local cache of values. There is no max size.

You can also use a map step in the same way to manage the cache yourself manually.

Parameters:

step_id – Unique ID.
up – Stream.
getter – On cache miss, get the new updated value for a key.
mapper – Called on each item with access to the cache. Each return value is emitted downstream.
ttl – Re-get values in the cache that are older than this.

Returns:

A stream of items returned by the mapper.

flat_map( step_id: str, up: Stream[X], mapper: Callable[[X], Iterable[Y]], ) → Stream[Y]#

Transform items one-to-many.

This is like a combination of map and flatten.

It is commonly used for:

Tokenizing
Flattening hierarchical objects
Breaking up aggregations for further processing

import bytewax.operators as op
from bytewax.testing import TestingSource
from bytewax.dataflow import Dataflow

flow = Dataflow("flat_map_eg")

inp = ["hello world"]
s = op.input("inp", flow, TestingSource(inp))

def split_into_words(sentence):
    return sentence.split()

s = op.flat_map("split_words", s, split_into_words)

_ = op.inspect("out", s)

flat_map_eg.out: 'hello'
flat_map_eg.out: 'world'

Parameters:

step_id – Unique ID.
up – Stream.
mapper – Called once on each upstream item. Returns the items to emit downstream.

Returns:

A stream of each item returned by the mapper.

flat_map_value( step_id: str, up: KeyedStream[V], mapper: Callable[[V], Iterable[W]], ) → KeyedStream[W]#

Transform values one-to-many.

Parameters:

step_id – Unique ID.
up – Keyed stream.
mapper – Called once on each upstream value. Returns the values to emit downstream.

Returns:

A keyed stream of each value returned by the mapper.

flatten( step_id: str, up: Stream[Iterable[X]], ) → Stream[X]#

Move all sub-items up a level.

Parameters:

step_id – Unique ID.
up – Stream of iterables.

Returns:

A stream of the items within each iterable in the upstream.

filter( step_id: str, up: Stream[X], predicate: Callable[[X], bool], ) → Stream[X]#

Keep only some items.

It is commonly used for:

Selecting relevant events
Removing empty events
Removing sentinels
Removing stop words

import bytewax.operators as op
from bytewax.testing import TestingSource
from bytewax.dataflow import Dataflow


flow = Dataflow("filter_eg")
s = op.input("inp", flow, TestingSource(range(4)))

def is_odd(item):
    return item % 2 != 0

s = op.filter("filter_odd", s, is_odd)
_ = op.inspect("out", s)

filter_eg.out: 1
filter_eg.out: 3

Parameters:

step_id – Unique ID.
up – Stream.
predicate – Called with each upstream item. Only items for which this returns true True will be emitted downstream.

Returns:

A stream with only the upstream items for which the predicate returns True.

filter_value( step_id: str, up: KeyedStream[V], predicate: Callable[[V], bool], ) → KeyedStream[V]#

Selectively keep only some items from a keyed stream.

Parameters:

step_id – Unique ID.
up – Keyed stream.
predicate – Will be called with each upstream value. Only values for which this returns True will be emitted downstream.

Returns:

A keyed stream with only the upstream pairs for which the predicate returns True.

filter_map( step_id: str, up: Stream[X], mapper: Callable[[X], Optional[Y]], ) → Stream[Y]#

A one-to-maybe-one transformation of items.

This is like a combination of map and then filter with a predicate removing None values.

import bytewax.operators as op
from bytewax.testing import TestingSource
from bytewax.dataflow import Dataflow

flow = Dataflow("filter_map_eg")
s = op.input(
    "inp",
    flow,
    TestingSource(
        [
            {"key": "a", "val": 1},
            {"bad": "obj"},
        ]
    ),
)

def validate(data):
    if type(data) != dict or "key" not in data:
        return None
    else:
        return data["key"], data

s = op.filter_map("validate", s, validate)
_ = op.inspect("out", s)

filter_map_eg.out: ('a', {'key': 'a', 'val': 1})

Parameters:

step_id – Unique ID.
up – Stream.
mapper – Called on each item. Each return value is emitted downstream, unless it is None.

Returns:

A stream of items returned from mapper, unless it is None.

filter_map_value( step_id: str, up: KeyedStream[V], mapper: Callable[[V], Optional[W]], ) → KeyedStream[W]#

Transform values one-to-maybe-one.

This is like a combination of map_value and then filter_value with a predicate removing None values.

Parameters:

step_id – Unique ID.
up – Stream.
mapper – Called on each value. Each return value is emitted downstream, unless it is None.

Returns:

A keyed stream of values returned from the mapper, unless the value is None. The key is unchanged.

fold_final( step_id: str, up: KeyedStream[V], builder: Callable[[], S], folder: Callable[[S, V], S], ) → KeyedStream[S]#

Build an empty accumulator, then combine values into it.

This only works on finite data streams and only returns a result once the upstream is EOF. You’ll need to use bytewax.operators.windowing.fold_window or bytewax.operators.stateful_flat_map on infinite data.

It is like reduce_final but uses a function to build the initial value.

Parameters:

step_id – Unique ID.
up – Keyed stream.
builder – Called the first time a key appears and is expected to return the empty accumulator for that key.
folder – Combines a new value into an existing accumulator and returns the updated accumulator. The accumulator is initially the empty accumulator.

Returns:

A keyed stream of the accumulators. Only once the upstream is EOF.

inspect( step_id: str, up: Stream[X], inspector: Callable[[str, X], None] = _default_inspector, ) → Stream[X]#

Observe items for debugging.

import bytewax.operators as op
from bytewax.testing import TestingSource
from bytewax.dataflow import Dataflow

flow = Dataflow("my_flow")
s = op.input("inp", flow, TestingSource(range(3)))
_ = op.inspect("help", s)

my_flow.help: 0
my_flow.help: 1
my_flow.help: 2

Parameters:

step_id – Unique ID.
up – Stream.
inspector – Called with the step ID and each item in the stream. Defaults to printing the step ID and each item.

Returns:

The upstream unmodified.

join( step_id: str, *sides: KeyedStream[Any], insert_mode: JoinInsertMode = 'last', emit_mode: JoinEmitMode = 'complete', ) → KeyedStream[Tuple]#

Gather together the value for a key on multiple streams.

See Joins for more information.

Parameters:

step_id – Unique ID.
*sides – Keyed streams.
insert_mode – Mode of this join. See JoinInsertMode for more info. Defaults to "last".
emit_mode – Mode of this join. See JoinEmitMode for more info. Defaults to "complete".

Returns:

Emits a tuple with the value from each stream in the order of the argument list. See JoinEmitMode for when tuples are emitted.

key_on( step_id: str, up: Stream[X], key: Callable[[X], str], ) → KeyedStream[X]#

Add a key for each item.

This allows you to use all the keyed operators that require the upstream to be a KeyedStream.

Parameters:

step_id – Unique ID.
up – Stream.
key – Called on each item and should return the key for that item.

Returns:

A stream of 2-tuples of (key, item) AKA a keyed stream. The keys come from the return value of the key function; upstream items will automatically be attached as values.

key_rm( step_id: str, up: KeyedStream[X], ) → Stream[X]#

Discard keys.

KeyedStreams are 2-tuples of (key, value). This will discard the key so you just have the values if you don’t need the keys anymore.

Parameters:

step_id – Unique ID.
up – Keyed stream.

Returns:

A stream of just values.

map( step_id: str, up: Stream[X], mapper: Callable[[X], Y], ) → Stream[Y]#

Transform items one-by-one.

It is commonly used for:

Serialization and deserialization.
Selection of fields.

import bytewax.operators as op
from bytewax.testing import TestingSource
from bytewax.dataflow import Dataflow

flow = Dataflow("map_eg")
s = op.input("inp", flow, TestingSource(range(3)))

def add_one(item):
    return item + 10

s = op.map("add_one", s, add_one)
_ = op.inspect("out", s)

map_eg.out: 10
map_eg.out: 11
map_eg.out: 12

Parameters:

step_id – Unique ID.
up – Stream.
mapper – Called on each item. Each return value is emitted downstream.

Returns:

A stream of items returned from the mapper.

map_value( step_id: str, up: KeyedStream[V], mapper: Callable[[V], W], ) → KeyedStream[W]#

Transform values one-by-one.

Parameters:

step_id – Unique ID.
up – Keyed stream.
mapper – Called on each value. Each return value is emitted downstream.

Returns:

A keyed stream of values returned from the mapper. The key is unchanged.

max_final( step_id: str, up: KeyedStream[V], by=_identity, ) → KeyedStream#

Find the maximum value for each key.

This only works on finite data streams and only returns a result once the upstream is EOF. You’ll need to use bytewax.operators.windowing.max_window on infinite data.

Parameters:

step_id – Unique ID.
up – Keyed stream.
by – A function called on each value that is used to extract what to compare.

Returns:

A keyed stream of the max values. Only once the upstream is EOF.

min_final( step_id: str, up: KeyedStream[V], by=_identity, ) → KeyedStream#

Find the minimum value for each key.

This only works on finite data streams and only returns a result once the upstream is EOF. You’ll need to use bytewax.operators.windowing.min_window on infinite data.

Parameters:

step_id – Unique ID.
up – Keyed stream.
by – A function called on each value that is used to extract what to compare.

Returns:

A keyed stream of the min values. Only once the upstream is EOF.

raises(step_id: str, up: Stream[Any]) → None#

Raise an exception and crash the dataflow on any item.

Parameters:

step_id – Unique ID.
up – Any item on this stream will throw a RuntimeError.

reduce_final( step_id: str, up: KeyedStream[V], reducer: Callable[[V, V], V], ) → KeyedStream[V]#

Distill all values for a key down into a single value.

This only works on finite data streams and only returns a result once the upstream is EOF. You’ll need to use bytewax.operators.windowing.reduce_window or bytewax.operators.stateful_flat_map on infinite data.

It is like fold_final but the first value is the initial accumulator.

Parameters:

step_id – Unique ID.
up – Keyed stream.
reducer – Combines a new value into an old value and returns the combined value.

Returns:

A keyed stream of the accumulators. Only once the upstream is EOF.

stateful_flat_map( step_id: str, up: KeyedStream[V], mapper: Callable[[Optional[S], V], Tuple[Optional[S], Iterable[W]]], ) → KeyedStream[W]#

Transform values one-to-many, referencing a persistent state.

Parameters:

step_id – Unique ID.
up – Keyed stream.
mapper – Called whenever a value is encountered from upstream with the last state or None, and then the upstream value. Should return a 2-tuple of (updated_state, emit_values). If the updated state is None, discard it.

Returns:

A keyed stream.

stateful_map( step_id: str, up: KeyedStream[V], mapper: Callable[[Optional[S], V], Tuple[Optional[S], W]], ) → KeyedStream[W]#

Transform values one-to-one, referencing a persistent state.

It is commonly used for:

Anomaly detection
State machines

import bytewax.operators as op
from bytewax.testing import TestingSource, run_main
from bytewax.dataflow import Dataflow

flow = Dataflow("stateful_map_eg")

inp = [
    "a",
    "a",
    "a",
    "b",
    "a",
]
s = op.input("inp", flow, TestingSource(inp))

s = op.key_on("self_as_key", s, lambda x: x)

def check(running_count, _item):
    if running_count is None:
        running_count = 0
    running_count += 1
    return (running_count, running_count)

s = op.stateful_map("running_count", s, check)
_ = op.inspect("out", s)

stateful_map_eg.out: ('a', 1)
stateful_map_eg.out: ('a', 2)
stateful_map_eg.out: ('a', 3)
stateful_map_eg.out: ('b', 1)
stateful_map_eg.out: ('a', 4)

Parameters:

step_id – Unique ID.
up – Keyed stream.
mapper – Called whenever a value is encountered from upstream with the last state or None, and then the upstream value. Should return a 2-tuple of (updated_state, emit_value). If the updated state is None, discard it.

Returns:

A keyed stream.

bytewax.operators#

Submodules#

Data#

Classes#

Functions#

`bytewax.operators`#