Hi all! I’m having a RAM issue when running one of my flows (diagram in thread). Essentially, I have a list of calls I need to make to a source system that I store in a list and the flow is for each item in that list

ping source system for dataset -> send dataset to pandas DF -> store dataset as a file in S3 -> push contents of said file to our datastore

I’m using the

LocalDaskExecutor

to do DFE and that’s helped a ton as the flow used to fail at the first step because it was doing Breadth first and holding all of the datasets in memory. I expected that once the task tree was completed for each dataset that it’d release that memory but that doesn’t seem to be the case. Am I doing something wrong/poorly or is there a way to free up that memory once the terminal task has completed for each branch?

Hi @Michael Kennely! Can you share a bit more info about your memory usage reporting here, what monitoring is indicating that memory isn’t being released over the course of the Flow Run?

The “monitoring” is pretty back of the envelope 😅 As the flow is running I have the

top

command running on the machine that I have my agent installed on and there’s a process that’s listed as

prefect

and it’s

%MEM

is continually increasing while the available memory on that machine continually decreases.

Got it, thanks that’s helpful!

Obviously, there are easy solutions for this, but I didn’t want to jus do something easy that wasn’t a good solution

Hey I haven’t forgotten about this, just having a quick discussion

Prefect Cloud - on the Developer Tier

Local agent and run config?

Local agent and

flow.environment = LocalEnvironment(
    executor=LocalDaskExecutor()
)

flow.storage = GitHub(redacted)

Hi! We’ve been talking about this as a team — it’s a bit of a tricky situation. Basically, we have to store the return values from every task so they can be used downstream. We can’t just “drop” this from some “intermediate” tasks because

dask

is managing the values. We’ll continue to think about this in the future, but because of this complexity we can’t promise a perfect solution right now. The best recommendation I have for this is to reduce the amount of data that needs to be passed between tasks. Passing large amounts of data from task to task won’t scale to a distributed setting anyway. I’d recommend moving your tasks that use the dataframe into a single task. Alternatively, you can write the dataframe to disk and pass the filepath to the next task.

Thanks for the replies @Zanie and @Dylan!! I greatly appreciate your time and attention on this. I appreciate the guidance and I think that the suggestions you gave are totally viable