Type Alias timely::dataflow::stream::Stream

pub type Stream<S, D> = StreamCore<S, Vec<D>>;

A stream batching data in vectors.

Aliased Type

struct Stream<S, D> { /* private fields */ }

Trait Implementations

impl<G: Scope, D: Data> Accumulate<G, D> for Stream<G, D>

fn accumulate<A: Data>( &self, default: A, logic: impl Fn(&mut A, RefOrMut<'_, Vec<D>>) + 'static, ) -> Stream<G, A>

Accumulates records within a timestamp.

fn count(&self) -> Stream<G, usize>

Counts the number of records observed at each time.

impl<S: Scope, K: ExchangeData + Hash + Eq, V: ExchangeData> Aggregate<S, K, V> for Stream<S, (K, V)>

fn aggregate<R: Data, D: Default + 'static, F: Fn(&K, V, &mut D) + 'static, E: Fn(K, D) -> R + 'static, H: Fn(&K) -> u64 + 'static>( &self, fold: F, emit: E, hash: H, ) -> Stream<S, R>

where S::Timestamp: Eq,

Aggregates data of the form (key, val), using user-supplied logic.

impl<S: Scope, D: Data> Branch<S, D> for Stream<S, D>

fn branch( &self, condition: impl Fn(&S::Timestamp, &D) -> bool + 'static, ) -> (Stream<S, D>, Stream<S, D>)

Takes one input stream and splits it into two output streams. For each record, the supplied closure is called with a reference to the data and its time. If it returns true, the record will be sent to the second returned stream, otherwise it will be sent to the first.

impl<G: Scope, D: ExchangeData> Broadcast<D> for Stream<G, D>

fn broadcast(&self) -> Stream<G, D>

Broadcast records to all workers.

impl<G: Scope, D: Data> Delay<G, D> for Stream<G, D>

fn delay<L: FnMut(&D, &G::Timestamp) -> G::Timestamp + 'static>( &self, func: L, ) -> Self

Advances the timestamp of records using a supplied function.

fn delay_total<L: FnMut(&D, &G::Timestamp) -> G::Timestamp + 'static>( &self, func: L, ) -> Self

where G::Timestamp: TotalOrder,

Advances the timestamp of records using a supplied function.

fn delay_batch<L: FnMut(&G::Timestamp) -> G::Timestamp + 'static>( &self, func: L, ) -> Self

Advances the timestamp of batches of records using a supplied function.

impl<G: Scope, D: Data> Filter<D> for Stream<G, D>

fn filter<P: FnMut(&D) -> bool + 'static>(&self, predicate: P) -> Stream<G, D>

Returns a new instance of self containing only records satisfying predicate.

impl<S: Scope, D: Data> Map<S, D> for Stream<S, D>

fn map_in_place<L: FnMut(&mut D) + 'static>(&self, logic: L) -> Stream<S, D>

Updates each element of the stream and yields the element, re-using memory where possible.

fn flat_map<I: IntoIterator, L: FnMut(D) -> I + 'static>( &self, logic: L, ) -> Stream<S, I::Item>

where I::Item: Data,

Consumes each element of the stream and yields some number of new elements.

fn map<D2: Data, L: FnMut(D) -> D2 + 'static>(&self, logic: L) -> Stream<S, D2>

Consumes each element of the stream and yields a new element.

impl<G: Scope, D: Data, D2: Data, F: Fn(D) -> (u64, D2) + 'static> Partition<G, D, D2, F> for Stream<G, D>

fn partition(&self, parts: u64, route: F) -> Vec<Stream<G, D2>>

Produces parts output streams, containing records produced and assigned by route.

impl<S: Scope, T: Data, E: Data> ResultStream<S, T, E> for Stream<S, Result<T, E>>

fn ok(&self) -> Stream<S, T>

Returns a new instance of self containing only ok records.

fn err(&self) -> Stream<S, E>

Returns a new instance of self containing only err records.

fn map_ok<T2: Data, L: FnMut(T) -> T2 + 'static>( &self, logic: L, ) -> Stream<S, Result<T2, E>>

Returns a new instance of self applying logic on all Ok records.

fn map_err<E2: Data, L: FnMut(E) -> E2 + 'static>( &self, logic: L, ) -> Stream<S, Result<T, E2>>

Returns a new instance of self applying logic on all Err records.

fn and_then<T2: Data, L: FnMut(T) -> Result<T2, E> + 'static>( &self, logic: L, ) -> Stream<S, Result<T2, E>>

Returns a new instance of self applying logic on all Ok records, passes through Err records.

fn unwrap_or_else<L: FnMut(E) -> T + 'static>(&self, logic: L) -> Stream<S, T>

Returns a new instance of self applying logic on all Ok records.

impl<S: Scope, K: ExchangeData + Hash + Eq, V: ExchangeData> StateMachine<S, K, V> for Stream<S, (K, V)>

fn state_machine<R: Data, D: Default + 'static, I: IntoIterator<Item = R>, F: Fn(&K, V, &mut D) -> (bool, I) + 'static, H: Fn(&K) -> u64 + 'static>( &self, fold: F, hash: H, ) -> Stream<S, R>

where S::Timestamp: Hash + Eq,

Tracks a state for each presented key, using user-supplied state transition logic.