Struct renoir::WindowedStream

pub struct WindowedStream<Op, O: Data, WinDescr>
where
    Op: Operator,
    Op::Out: KeyedItem,
    WinDescr: WindowDescription<<Op::Out as KeyedItem>::Value>,
{ /* private fields */ }

A WindowedStream is a data stream partitioned by Key, where elements of each partition are divided in groups called windows. Each element can be assigned to one or multiple windows.

Windows are handled independently for each partition of the stream. Each partition may be processed in parallel.

The windowing logic is implemented through 3 traits:

• A WindowDescription contains the parameters and logic that characterize the windowing strategy, when given a WindowAccumulator it instantiates a WindowManager.
• A WindowManger is the struct responsible for creating the windows and forwarding the input elements to the correct window which will should pass it to its WindowAccumulator.
• A WindowAccumulator contains the logic that should be applied to the elements of each window.

There are a set of provided window descriptions with their respective managers:

• EventTimeWindow
• ProcessingTimeWindow
• CountWindow
• SessionWindow
• TransactionWindow

Implementations

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn fold<NewOut: Data, F>( self, init: NewOut, fold: F ) -> KeyedStream<impl Operator<Out = (Key, NewOut)>>

where F: FnMut(&mut NewOut, Out) + Clone + Send + 'static,

Folds the elements of each window into an accumulator value

fold() takes two arguments: the initial value of the accumulator and a closure used to accumulate the elements of each window.

The closure is called once for each element of each window with two arguments: a mutable reference to the accumulator and the element of the window. The closure should modify the accumulator, without returning anything.

Example

let s = env.stream_iter(0..5);
let res = s
    .group_by(|&n| n % 2)
    .window(CountWindow::tumbling(2))
    .fold(1, |acc, n| *acc *= n)
    .collect_vec();

env.execute_blocking();

let mut res = res.get().unwrap();
res.sort_unstable();
assert_eq!(res, vec![(0, 0 * 2), (1, 1 * 3)]);

pub fn fold_first<F>( self, fold: F ) -> KeyedStream<impl Operator<Out = (Key, Out)>>

where F: FnMut(&mut Out, Out) + Clone + Send + 'static,

Folds the elements of each window into an accumulator value, starting with the first value

TODO DOCS

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data + Ord,

pub fn map<NewOut: Data, F: Fn(Vec<Out>) -> NewOut + Send + Clone + 'static>( self, f: F ) -> KeyedStream<impl Operator<Out = (Key, NewOut)>>

Prefer other aggregators if possible as they don’t save all elements

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn count(self) -> KeyedStream<impl Operator<Out = (Key, usize)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data + Ord,

pub fn max(self) -> KeyedStream<impl Operator<Out = (Key, Out)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn max_by_key<K: Ord, F: Fn(&Out) -> K + Clone + Send + 'static>( self, get_key: F ) -> KeyedStream<impl Operator<Out = (Key, Out)>>

pub fn max_by<F: Fn(&Out, &Out) -> Ordering + Clone + Send + 'static>( self, compare: F ) -> KeyedStream<impl Operator<Out = (Key, Out)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data + Ord,

pub fn min(self) -> KeyedStream<impl Operator<Out = (Key, Out)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn min_by_key<K: Ord, F: Fn(&Out) -> K + Clone + Send + 'static>( self, get_key: F ) -> KeyedStream<impl Operator<Out = (Key, Out)>>

pub fn min_by<F: Fn(&Out, &Out) -> Ordering + Clone + Send + 'static>( self, compare: F ) -> KeyedStream<impl Operator<Out = (Key, Out)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn first(self) -> KeyedStream<impl Operator<Out = (Key, Out)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn last(self) -> KeyedStream<impl Operator<Out = (Key, Out)>>

impl<Key, Out, WindowDescr, OperatorChain> WindowedStream<OperatorChain, Out, WindowDescr>

where WindowDescr: WindowDescription<Out>, OperatorChain: Operator<Out = (Key, Out)> + 'static, Key: DataKey, Out: Data,

pub fn sum<NewOut: Data + Default + AddAssign<Out>>( self ) -> KeyedStream<impl Operator<Out = (Key, NewOut)>>