Collections API¶

Collections¶

`Streaming`([stream, example, stream_type])	Superclass for streaming collections
`Streaming.map_partitions`(func, args, *kwargs)	Map a function across all batch elements of this stream
`Streaming.accumulate_partitions`(self, func, …)	Accumulate a function with state across batch elements
`Streaming.verify`(self, x)	Verify elements that pass through this stream

Batch¶

`Batch`([stream, example])	A Stream of tuples or lists
`Batch.filter`(self, predicate)	Filter elements by a predicate
`Batch.map`(self, func, **kwargs)	Map a function across all elements
`Batch.pluck`(self, ind)	Pick a field out of all elements
`Batch.to_dataframe`(self)	Convert to a streaming dataframe
`Batch.to_stream`(self)	Concatenate batches and return base Stream

Dataframes¶

`DataFrame`(args, *kwargs)	A Streaming Dataframe
`DataFrame.groupby`(self, other)	Groupby aggregations
`DataFrame.rolling`(self, window[, …])	Compute rolling aggregations
`DataFrame.assign`(self, **kwargs)	Assign new columns to this dataframe
`DataFrame.sum`(self[, start])	Sum frame.
`DataFrame.mean`(self[, start])	Average frame
`DataFrame.cumsum`(self)	Cumulative sum
`DataFrame.cumprod`(self)	Cumulative product
`DataFrame.cummin`(self)	Cumulative minimum
`DataFrame.cummax`(self)	Cumulative maximum

`GroupBy`(root, grouper[, index])	Groupby aggregations on streaming dataframes
`GroupBy.count`(self[, start])	Groupby-count
`GroupBy.mean`(self[, with_state, start])	Groupby-mean
`GroupBy.size`(self)	Groupby-size
`GroupBy.std`(self[, ddof])	Groupby-std
`GroupBy.sum`(self[, start])	Groupby-sum
`GroupBy.var`(self[, ddof])	Groupby-variance

`Rolling`(sdf, window, min_periods, …)	Rolling aggregations
`Rolling.aggregate`(self, args, *kwargs)	Rolling aggregation
`Rolling.count`(self, args, *kwargs)	Rolling count
`Rolling.max`(self)	Rolling maximum
`Rolling.mean`(self)	Rolling mean
`Rolling.median`(self)	Rolling median
`Rolling.min`(self)	Rolling minimum
`Rolling.quantile`(self, args, *kwargs)	Rolling quantile
`Rolling.std`(self, args, *kwargs)	Rolling standard deviation
`Rolling.sum`(self)	Rolling sum
`Rolling.var`(self, args, *kwargs)	Rolling variance

`DataFrame.window`(self[, n, value, …])	Sliding window operations
`Window.apply`(self, func)	Apply an arbitrary function over each window of data
`Window.count`(self)	Count elements within window
`Window.groupby`(self, other)	Groupby-aggregations within window
`Window.sum`(self)	Sum elements within window
`Window.size`	Number of elements within window
`Window.std`(self[, ddof])	Compute standard deviation of elements within window
`Window.var`(self[, ddof])	Compute variance of elements within window

`Rolling.aggregate`(self, args, *kwargs)	Rolling aggregation
`Rolling.count`(self, args, *kwargs)	Rolling count
`Rolling.max`(self)	Rolling maximum
`Rolling.mean`(self)	Rolling mean
`Rolling.median`(self)	Rolling median
`Rolling.min`(self)	Rolling minimum
`Rolling.quantile`(self, args, *kwargs)	Rolling quantile
`Rolling.std`(self, args, *kwargs)	Rolling standard deviation
`Rolling.sum`(self)	Rolling sum
`Rolling.var`(self, args, *kwargs)	Rolling variance

Random([freq, interval, dask]) A streaming dataframe of random data

Details¶

class streamz.collection.Streaming(stream=None, example=None, stream_type=None)¶

Superclass for streaming collections

Do not create this class directly, use one of the subclasses instead.

Parameters:	stream: streamz.Stream example: object An object to represent an example element of this stream

See also

streamz.dataframe.StreamingDataFrame
streamz.dataframe.StreamingBatch

Methods

`accumulate_partitions`(self, func, *args, …)	Accumulate a function with state across batch elements
`map_partitions`(func, args, *kwargs)	Map a function across all batch elements of this stream
`verify`(self, x)	Verify elements that pass through this stream

emit

accumulate_partitions(self, func, *args, **kwargs)¶

Accumulate a function with state across batch elements

See also

Streaming.map_partitions

static map_partitions(func, *args, **kwargs)¶

Map a function across all batch elements of this stream

The output stream type will be determined by the action of that function on the example

See also

Streaming.accumulate_partitions

verify(self, x)¶: Verify elements that pass through this stream

class streamz.batch.Batch(stream=None, example=None)¶

A Stream of tuples or lists

This streaming collection manages batches of Python objects such as lists of text or dictionaries. By batching many elements together we reduce overhead from Python.

This library is typically used at the early stages of data ingestion before handing off to streaming dataframes

Examples

>>> text = Streaming.from_file(myfile)  # doctest: +SKIP
>>> b = text.partition(100).map(json.loads)  # doctest: +SKIP

Methods

`accumulate_partitions`(self, func, *args, …)	Accumulate a function with state across batch elements
`filter`(self, predicate)	Filter elements by a predicate
`map`(self, func, **kwargs)	Map a function across all elements
`map_partitions`(func, args, *kwargs)	Map a function across all batch elements of this stream
`pluck`(self, ind)	Pick a field out of all elements
`sum`(self)	Sum elements
`to_dataframe`(self)	Convert to a streaming dataframe
`to_stream`(self)	Concatenate batches and return base Stream
`verify`(self, x)	Verify elements that pass through this stream

emit

accumulate_partitions(self, func, *args, **kwargs)¶

Accumulate a function with state across batch elements

See also

Streaming.map_partitions

filter(self, predicate)¶: Filter elements by a predicate

map(self, func, **kwargs)¶: Map a function across all elements

static map_partitions(func, *args, **kwargs)¶

Map a function across all batch elements of this stream

The output stream type will be determined by the action of that function on the example

See also

Streaming.accumulate_partitions

pluck(self, ind)¶: Pick a field out of all elements

sum(self)¶: Sum elements

to_dataframe(self)¶

Convert to a streaming dataframe

This calls pd.DataFrame on all list-elements of this stream

to_stream(self)¶

Concatenate batches and return base Stream

Returned stream will be composed of single elements

verify(self, x)¶: Verify elements that pass through this stream

class streamz.dataframe.DataFrame(*args, **kwargs)¶

A Streaming Dataframe

This is a logical collection over a stream of Pandas dataframes. Operations on this object will translate to the appropriate operations on the underlying Pandas dataframes.

See also

Series

Attributes:	columns dtypes index `size` size of frame

Methods

`accumulate_partitions`(self, func, *args, …)	Accumulate a function with state across batch elements
`assign`(self, **kwargs)	Assign new columns to this dataframe
`count`(self[, start])	Count of frame
`cummax`(self)	Cumulative maximum
`cummin`(self)	Cumulative minimum
`cumprod`(self)	Cumulative product
`cumsum`(self)	Cumulative sum
`groupby`(self, other)	Groupby aggregations
`map_partitions`(func, args, *kwargs)	Map a function across all batch elements of this stream
`mean`(self[, start])	Average frame
`reset_index`(self)	Reset Index
`rolling`(self, window[, min_periods, …])	Compute rolling aggregations
`round`(self[, decimals])	Round elements in frame
`set_index`(self, index, **kwargs)	Set Index
`sum`(self[, start])	Sum frame.
`tail`(self[, n])	Round elements in frame
`to_frame`(self)	Convert to a streaming dataframe
`verify`(self, x)	Verify consistency of elements that pass through this stream
`window`(self[, n, value, with_state, start])	Sliding window operations

aggregate
astype
emit
map
query

accumulate_partitions(self, func, *args, **kwargs)¶

Accumulate a function with state across batch elements

See also

Streaming.map_partitions

assign(self, **kwargs)¶

Assign new columns to this dataframe

Alternatively use setitem syntax

Examples

>>> sdf = sdf.assign(z=sdf.x + sdf.y)  # doctest: +SKIP
>>> sdf['z'] = sdf.x + sdf.y  # doctest: +SKIP

count(self, start=None)¶

Count of frame

Parameters:	start: None or resulting Python object type from the operation Accepts a valid start state.

cummax(self)¶: Cumulative maximum

cummin(self)¶: Cumulative minimum

cumprod(self)¶: Cumulative product

cumsum(self)¶: Cumulative sum

groupby(self, other)¶: Groupby aggregations

static map_partitions(func, *args, **kwargs)¶

Map a function across all batch elements of this stream

The output stream type will be determined by the action of that function on the example

See also

Streaming.accumulate_partitions

mean(self, start=None)¶

Average frame

Parameters:	start: None or resulting Python object type from the operation Accepts a valid start state.

reset_index(self)¶: Reset Index

rolling(self, window, min_periods=1, with_state=False, start=())¶

Compute rolling aggregations

When followed by an aggregation method like sum, mean, or std this produces a new Streaming dataframe whose values are aggregated over that window.

The window parameter can be either a number of rows or a timedelta like ``”2 minutes”` in which case the index should be a datetime index.

This operates by keeping enough of a backlog of records to maintain an accurate stream. It performs a copy at every added dataframe. Because of this it may be slow if the rolling window is much larger than the average stream element.

Parameters:	window: int or timedelta Window over which to roll with_state: bool (False) Whether to return the state along with the result as a tuple (state, result). State may be needed downstream for a number of reasons like checkpointing. start: () or resulting Python object type from the operation Accepts a valid start state.
Returns:	Rolling object

See also

DataFrame.window: more generic window operations

round(self, decimals=0)¶: Round elements in frame

set_index(self, index, **kwargs)¶: Set Index

size¶: size of frame

sum(self, start=None)¶

Sum frame.

Parameters:	start: None or resulting Python object type from the operation Accepts a valid start state.

tail(self, n=5)¶: Round elements in frame

to_frame(self)¶: Convert to a streaming dataframe

verify(self, x)¶: Verify consistency of elements that pass through this stream

window(self, n=None, value=None, with_state=False, start=None)¶

Sliding window operations

Windowed operations are defined over a sliding window of data, either with a fixed number of elements:

>>> df.window(n=10).sum()  # sum of the last ten elements

or over an index value range (index must be monotonic):

>>> df.window(value='2h').mean()  # average over the last two hours

Windowed dataframes support all normal arithmetic, aggregations, and groupby-aggregations.

Parameters:	n: int Window of number of elements over which to roll value: str Window of time over which to roll with_state: bool (False) Whether to return the state along with the result as a tuple (state, result). State may be needed downstream for a number of reasons like checkpointing. start: None or resulting Python object type from the operation Accepts a valid start state.

See also

DataFrame.rolling: mimic’s Pandas rolling aggregations

Examples

>>> df.window(n=10).std()
>>> df.window(value='2h').count()

>>> w = df.window(n=100)
>>> w.groupby(w.name).amount.sum()
>>> w.groupby(w.x % 10).y.var()

class streamz.dataframe.Rolling(sdf, window, min_periods, with_state, start)¶

Rolling aggregations

This intermediate class enables rolling aggregations across either a fixed number of rows or a time window.

Examples

>>> sdf.rolling(10).x.mean()  # doctest: +SKIP
>>> sdf.rolling('100ms').x.mean()  # doctest: +SKIP

Methods

`aggregate`(self, args, *kwargs)	Rolling aggregation
`count`(self, args, *kwargs)	Rolling count
`max`(self)	Rolling maximum
`mean`(self)	Rolling mean
`median`(self)	Rolling median
`min`(self)	Rolling minimum
`quantile`(self, args, *kwargs)	Rolling quantile
`std`(self, args, *kwargs)	Rolling standard deviation
`sum`(self)	Rolling sum
`var`(self, args, *kwargs)	Rolling variance

aggregate(self, *args, **kwargs)¶: Rolling aggregation

count(self, *args, **kwargs)¶: Rolling count

max(self)¶: Rolling maximum

mean(self)¶: Rolling mean

median(self)¶: Rolling median

min(self)¶: Rolling minimum

quantile(self, *args, **kwargs)¶: Rolling quantile

std(self, *args, **kwargs)¶: Rolling standard deviation

sum(self)¶: Rolling sum

var(self, *args, **kwargs)¶: Rolling variance

class streamz.dataframe.Window(sdf, n=None, value=None, with_state=False, start=None)¶

Windowed aggregations

This provides a set of aggregations that can be applied over a sliding window of data.

See also

DataFrame.window: contains full docstring

Attributes:	columns dtypes example index `size` Number of elements within window

Methods

`apply`(self, func)	Apply an arbitrary function over each window of data
`count`(self)	Count elements within window
`groupby`(self, other)	Groupby-aggregations within window
`mean`(self)	Average elements within window
`std`(self[, ddof])	Compute standard deviation of elements within window
`sum`(self)	Sum elements within window
`value_counts`(self)	Count groups of elements within window
`var`(self[, ddof])	Compute variance of elements within window

aggregate
full
map_partitions
reset_index

apply(self, func)¶: Apply an arbitrary function over each window of data

count(self)¶: Count elements within window

groupby(self, other)¶: Groupby-aggregations within window

mean(self)¶: Average elements within window

size¶: Number of elements within window

std(self, ddof=1)¶: Compute standard deviation of elements within window

sum(self)¶: Sum elements within window

value_counts(self)¶: Count groups of elements within window

var(self, ddof=1)¶: Compute variance of elements within window

class streamz.dataframe.GroupBy(root, grouper, index=None)¶

Groupby aggregations on streaming dataframes

Methods

`count`(self[, start])	Groupby-count
`mean`(self[, with_state, start])	Groupby-mean
`size`(self)	Groupby-size
`std`(self[, ddof])	Groupby-std
`sum`(self[, start])	Groupby-sum
`var`(self[, ddof])	Groupby-variance

count(self, start=None)¶

Groupby-count

Parameters:	start: None or resulting Python object type from the operation Accepts a valid start state.

mean(self, with_state=False, start=None)¶

Groupby-mean

Parameters:	start: None or resulting Python object type from the operation Accepts a valid start state.

size(self)¶: Groupby-size

std(self, ddof=1)¶: Groupby-std

sum(self, start=None)¶

Groupby-sum

Parameters:	start: None or resulting Python object type from the operation Accepts a valid start state.

var(self, ddof=1)¶: Groupby-variance

class streamz.dataframe.Random(freq='100ms', interval='500ms', dask=False)¶

A streaming dataframe of random data

The x column is uniformly distributed. The y column is poisson distributed. The z column is normally distributed.

This class is experimental and will likely be removed in the future

Parameters:	freq: timedelta The time interval between records interval: timedelta The time interval between new dataframes, should be significantly larger than freq
Attributes:	columns dtypes index `size` size of frame

Methods

`accumulate_partitions`(self, func, *args, …)	Accumulate a function with state across batch elements
`assign`(self, **kwargs)	Assign new columns to this dataframe
`count`(self[, start])	Count of frame
`cummax`(self)	Cumulative maximum
`cummin`(self)	Cumulative minimum
`cumprod`(self)	Cumulative product
`cumsum`(self)	Cumulative sum
`groupby`(self, other)	Groupby aggregations
`map_partitions`(func, args, *kwargs)	Map a function across all batch elements of this stream
`mean`(self[, start])	Average frame
`reset_index`(self)	Reset Index
`rolling`(self, window[, min_periods, …])	Compute rolling aggregations
`round`(self[, decimals])	Round elements in frame
`set_index`(self, index, **kwargs)	Set Index
`sum`(self[, start])	Sum frame.
`tail`(self[, n])	Round elements in frame
`to_frame`(self)	Convert to a streaming dataframe
`verify`(self, x)	Verify consistency of elements that pass through this stream
`window`(self[, n, value, with_state, start])	Sliding window operations

aggregate
astype
emit
map
query
stop