Parallel streams with buffers

[john-jam]

In a simple use case like downloading files and process them on a single machine, how could one achieve parallelization of downloads and processes with buffers?

Example:

import time
from streamz import Stream
from tornado.ioloop import IOLoop


def download_file(file_id: int):
    time.sleep(1)
    print(f"Downloaded file: {file_id}")
    return file_id


def process_file(file_id: int):
    time.sleep(2)
    print(f"Processed file : {file_id}")
    return file_id


async def streamz_run():
    s = Stream(asynchronous=True)
    s.map(download_file).buffer(4).sink(process_file)
    for i in range(10):
        await s.emit(i)


if __name__ == '__main__':
    start = time.time()
    IOLoop().run_sync(streamz_run)
    print(f"Streamz run took: {time.time() - start}s")

The download_file is properly buffered but not executed at the same time as process_file. The whole thing takes ~30s to run while we could expect 21s with parallel downloads/processes. Is using Dask the intended way in that case?

[martindurant]

Streamz is not a parallelism framework, but it can be concurrent ("async") for tasks that spend most of their time waiting. If your functions were async and you used asyncio.wait, they would have a shorter total run time. However, whether you can actually get parallelism depends on exactly what calls you make, and mixing CPU with IO is always tricky. Often a separate worker thread would end up running CPU loads (but python's GIL means you still might not get parallelism).

Dask can well be the parallelism engine for you, and it has various cluster topologies you can set up. From streamz's point of view, dask is a handy way to hand of mini-batches of events; but it could also be long-running tasks like downloads, in theory. In fact, if download/process is all you are doing, you can just use dask without srteamz (the delayed or client.submit patterns).

Note that no one is developing streamz these days, but I believe it can do what you want, if you have the interest to dig in.

[john-jam]

@martindurant Thanks for your prompt and useful answer!

When you indicate to use async methods, does streamz support this? When I try, it indicates that my download_file method was never awaited. Or maybe you were indicating to use async methods and asyncio.wait outside of streamz?

Anyway, I guess Dask can do what I want as you mentioned but I need some sets of tasks (e.g. download + process) to be executed on the same dask worker since it can be a different machine (different fs).

I didn't catch the last commit date, but streamz still looks useful! Thanks

[martindurant]

When you indicate to use async methods, does streamz support this?

Yes, there should be some examples of this.

Dask can do what I want as you mentioned but I need some sets of tasks (e.g. download + process) to be executed on the same dask worker

There are various ways to do this kind of thing, but having shared storage is a useful thing for a cluster. I'm not immediately sure how you would phrase "download X can happen anywhere, but processing must happen where its associated download happened".