Hi everyone I have just created my first Prefect flow and de

Question

Hi everyone I have just created my first Prefect flow and deployed it to my manually set up Dask cluster via the Prefect Cloud Everything went smoothly and I have to say this by far the best data pipeline tool I have used However I noticed that when I run the Flow from the cloud instead of just a script `flow run` the memory consumption of my Dask Workers gradually grows until they run out of memory When I run the same flow from a script everything finishes correctly and there are no memory issues what so ever I am using the latest `prefect` and `dask` packages Has anyone experienced something similar Is there anything that could be done about it Thanks a lot

Jenny · Answer

Hi Zviri Welcome to Prefect That s not something we d expect to happen so we d like to look into it further Our Dask experts should be online a little later they re not on EST and I m going to ask for their input

Zviri · Answer

Thanks a a lot I ll be checking this thread regularly to provide more details

Jim Crist-Harif · Answer

Hi Zviri The difference between running with Dask and running with a local executor is that Dask will attempt to run your tasks in parallel This can be nice your tasks progress through faster since things can be done in parallel But if your tasks use lots of memory running them in parallel may exhaust your resources say one task takes 1 GB if you have 8 running at a time on a worker that s 8 GB Dask takes some of this into account and will generally try to do the right thing but there are certain task structures that will break things Generally Dask works best if your individual tasks are quick ish 0 5 60s each and use a modest amount of memory MiBs not GiBs and you break your work into many smaller tasks High memory tasks are harder to schedule in parallel without breaking things especially if you re using mapped tasks they can work but they may require a bit more care If you re already following the above practices I d need to know more about your flow and execution environment to help debug further

Zviri · Answer

Hi Jim thank for your input I would say that I follow the best practices see below I ve set up an OCR processing pipeline broken down into the following tasks 1 fetch URLs from a database returns list 2 mapped task fetch and save files to S3 3 mapped task preprocess PDF document loads stores data from to S3 4 mapped task OCR document loads stores data from to S3 I currently run relatively small flows each processes at most 2k documents at one time but ideally I would like to scale it up to much larger numbers Each task generally takes less than 60 seconds The workers are set up to use process level parallelism and only 1 thread most of the code is pure python so threads are not very effective I currently use 4GB of memory for each worker process which is plenty the worker rarely uses more than 2GB of memory when as I said start the flow from a python script However when I start the flow from the Cloud using a RemoteDaskEnvironment the same flow that finished when ran from the script does not go beyond the first two steps fetching URLs from the db and then fetching storing them to S3 without exhausting the whole 4GBs of memory

Jim Crist-Harif · Answer

Are the outputs from your tasks large in size the things your task returns not things you may write directly to s3 within the task Depending on execution order dask workers will store intermediate values between tasks locally these may build up over time but generally dask will try to run tasks that let it remove intermediate values before running tasks that create new intermediate values

Zviri · Answer

< Jim Crist Harif> sorry for not responding earlier I was out on Saturday The output of each task is small I only use some named tuples with a couple of fields to pass between tasks I never pass any actual data between tasks just some URIs etc I was reading into debugging profiling of dask workers and while there is plenty of materials around CPU performance I have not found much on memory profiling Is there any tool you guys use to profile the worker s memory to at least get an idea of what is pilling up Currently I m just trying things blindly which is not yielding much

Zviri · Answer

Ok so I did some digging around and memory profiled one of my dask workers using `heapy` I found out that what is clogging up the memory of my workers are these strings there are tons of them ``` lt p gt The error code is a string that uniquely identifies an error condition It is meant to be read and understood by programs that detect and handle errors by type lt p gt lt p class= title gt lt b gt Amazon S3 error codes lt b gt lt p gt lt ul gt lt li gt lt ul gt lt li gt lt p gt lt i gt Code lt i gt AccessDenied lt p gt lt li gt lt li gt lt p gt lt i gt Description lt i gt Access Denied lt p gt lt li gt lt li gt lt p gt lt i gt HTTP Status Code lt i gt 403 Forbidden lt p gt lt li gt lt li gt lt p gt lt i gt SOAP Fault Code Prefix lt i gt Client lt p gt lt li gt lt ul gt lt li gt ``` So at first I assumed a library of mine used for accessing s3 is doing this so I stripped down my flow to only 2 steps 1 fetch a list of URLs from db 2 download them using requests but do not store the results anywhere So now there was no code of mine doing anything with s3 But the issue was still occurring So I figured the only remaining culprit could be the S3 storage that I used for my flows To test this hypothesis I replaced the S3 storage with Local storage As expected the issue disappeared after this I have no idea what the root cause of this is so I will continue digging But it seems to be connected to boto3 and to originate in the S3 storage

Zviri · Answer

< Jim Crist Harif> I managed to work around this problem using the Local storage in combination with sshfs I m not very happy about it since it is quite fragile but it will do for the time being At this point I m 100 it s the S3 storage causing the huge memory consumption I was not able to find the root cause yet and the strings I pasted above are probably just some side effects I checked the S3 storage and S3ResultHandler classes and nothing there struck me as causing any leakage AFAIK boto3 sessions clients don t need any explicit release management I m using ` nprocs` with the workers so it might have to do something with the forking logic

Jim Crist-Harif · Answer

Hmmm interesting Great debugging work < Zviri> I m not sure what would cause that with prefect s s3 storage If you can create a and file an issue we could help debug and track down what s going on

Zviri · Answer

< Jim Crist Harif> will do

Zviri · Answer

< Jim Crist Harif> I noticed that in the meantime just the last week or so you went from version 0 10 3 to 0 10 7 in which you also refactored the S3 storage and it seems the issue went away So I m not going to submit any issues for now You guys are on a roll keep up the good work muscle

Jim Crist-Harif · Answer

Glad to hear it