[Python] Lightweight python library to process streams of ordered data / to process batch of...

I am seeking advice on appropriate (lightweight) python library to process streams of ordered data / to process batch of ordered data?

An example of workflow (not very complex) can be:

from typing import Dict
from pandas import DataFrame
from numpy import array

def my_iterator(*args) -> Dict[str, DataFrame]:
# Parse files with some processing already
yield {"key1": df1, "key2": df2}

def analyze(dfs: Dict[str, DataFrame], buffer: dict) -> array:
# Analysis on some columns of some DataFrame
pass

def custom_join(dfs: Dict[str, DataFrame], settings: array, buffer: dict) -> DataFrame:
# Custom join from multiple Dataframe to a single one.
pass

def write_to_database(key: str, df: DataFrame):
# Write df to a database using key.
pass


def main():
buffer_analysis = {}
buffer_join = {}
for dfs_dict in my_iterator(*args):
# Process data
analysis = analyze(dfs_dict, buffer_analysis)
res = custom_join(dfs_dict, analysis, buffer_join)
# Write result to disk.
write_to_database("key_res", res)


I do have some other worlfows, a bit more complex, here again with an iterator yielding dicts of Dataframes, and requiring filtering, segmented cumulative aggregation and ending with a write step.

Requirements

• Parallel looping is not possible. Data is ordered and analysis and custom join will keep a state of the data they have processed so far, taking it into account to process new data.
• But support of asyncio can be of interest (the write step may takes some time for instance, while the next loop can already be started)

Identified solutions

• streamz really caught my interest, as being lightweight; but it appears not to be maintained. With it, i would have defined my workflow with map() function, then I could have run the iterator with the stream already defined. Something like (if I understand it correctly):

from streamz import Stream
# define 'analysis' and 'custom_join'

source = Stream()
analysis = source.map(analyze)
source.combine_latest(analysis).map(custom_join).sink(write_to_database)

def main():
buffer_analysis = {}
buffer_join = {}
for dfs_dict in my_iterator(*args):
# Yet have to workout how to use the buffers.
source.emit(dfs_dict)

• dask of course, with dask.delayed; but the documentation (and answers in some tickes) makes it clear you cannot apply dask.delayed on an iterator. This means I would apply dask.delayed within the for loop to analyze, custom_join and write_to_database, and finally, still within the loop, dask.compute(). This would trigger rebuilding the same graph at each iteration, and this appears to me not very efficient. The thing is that all dicts of DataFrames that are yielded cannot be materialized in memory as a list because it would not fit into RAM.

Please, what would be your advices?

Continue reading...