Hi in prefect 2 0 is there a way to perform yields we have s

Question

Hi in prefect 2 0 is there a way to perform yields we have some unfortunately long running tasks where we stream some content which takes a while +30 min very large zipped json files and I can t seem to figure out if I can yield partial result futures in my task incrementally and start processing them in down stream tasks Currently the work flow is Stream read big annoying file into small chunks save them in S3 incrementally in a separate thread inside the task and return list of all the chunked file URI s as single result = single `prefectfuture list str ` object this is where I would like to yield Run some down stream processing on each chunk Here we use future result to loop over and start each task this works 100 but if we could use yield in the first task we could begin processing immediately Is something like this supported going to be supported I tried making it work with contextmanager and tasks but ran into nothing but trouble

Malthe Karbo · Answer

something like pseudo code ```from prefect import flow task from import load file from concurrent futures import ThreadpoolExecutor as completed task contextmanager def stream big file file bytes = load file futures = with ThreadpoolEexecutor as executor with mystreamer file bytes as stream for chunked content in stream chunk uri str = executor submit persist func chunked content futures append chunk uri for future in as completed futures yield future result task def process chunk chunk uri flow def yield flow uri iterator = stream big file process chunk map uri iterator should not be blocking```

Malthe Karbo · Answer

On a related note is anything like `as completed` from concurrent futures supported for prefectfutures

Anna Geller · Answer

re as completed check out this example from Michael <https discourse prefect io t how to get finished futures as soon as they complete in a non blocking way 790> how are you processing this file e g `pandas` has a way of reading files in chunks so that you wouldn t have to rely on async or any Prefect functionality for that

Anna Geller · Answer

and mapping is not there yet in 2 0 so you can t use this syntax yet ```process chunk map uri iterator ``` you would need a for loop

Malthe Karbo · Answer

We read multiple 160Gb zipped json files we can t change this logic as the data format comes from third parties No pandas at all once we chunk the content out we persist it in raw datalake storage Current limitation from our implementation is that we need to stream all 160gb first and then we can begin processing chunks in parallel by using a loop over `list of chunks result ` and passing the chunks to next task I was hoping there was a way to get around this by yielding incremental future chunks from a task slightly smiling face I don t want to get around the async functionality from prefect I really like the future setup of 2 0

Malthe Karbo · Answer

And thanks for the discourse link That will do for now until an equivalent would become included in prefect smiley

Anna Geller · Answer

Gotcha Thanks for this great and detailed explanation And a really tough problem I think you re right that the example from Michael seems like the right approach for now but once we have mapping in 2 0 this will get easier

Malthe Karbo · Answer

It is the right approach for `as completed` but as for my initial inquiry single future of iterable where yield would make a big difference it does not help unfortunately it is equivalent to my current implementation of `for chunk result in future list result ` The current implementation works it would just be nice to cut off 1 hour of processing per flow run to save cost it adds up when you run many workers of size in k8s Map would also not solve it I think what I map over has to be an iterable of futures but currently a task can only deliver a single future to my knowledge this would be really powerful feature if yieldable was possible though but maybe not so easy to implement or maybe it is implemented and I just can not figure out how to do it Sorry for the confusion with two questions in a single thread sweat smile

Malthe Karbo · Answer

Is github issues the place for feature requests or is slack community the prefered way I don t mind writing up something about this issue and how something like yield would help me in a more detailed way for evaluation as a feature

Anna Geller · Answer

Great idea If you could submit a feature request with a description of the problem that yield would solve for you that would be awesome You can do that <https github com PrefectHQ prefect issues new choose|here> just pick 2 0 Not sure if this is relevant but you need to consider that task runs are stateless while you may pass data from one task to downstream tasks they shouldn t share stateful information with each other which could be relevant when you build such logic to process one giant file

Malthe Karbo · Answer

Thank you I will do so slightly smiling face

Malthe Karbo · Answer

<https github com PrefectHQ prefect issues 5708> for anyone searching for yield