The flow is relatively simple, it syncs files between an SFTP server and an S3 bucket. First it gets a secret from AWSSecretsManager, uses that to list file om an SFTP server using a task I wrote using paramiko, a separate task lists all files in an S3 bucket, then I’m comparing both file list and map all files that are not in the S3 bucket to an SFTP download task, which is followed by an S3Upload task

I’d be happy to share more code if it’d be helpful 🙂

from guppy import hpy; h=hpy()

I suspect that needs to be called in each task you’re interested in

Kidding

The original agent was a fargate with 1 vcpu and 3gb ram - my latest attempt was

flow.run_config = ECSRun(

labels=[f"data-{self._config.env}"], cpu=2048, memory=16384

)

thats 2 vCPUs in AWS parlance

i.e. thats about 5k children in the flow

one child for the download, one for the upload

woah, that explains it

my understanding was that flows should be designed such that we could easily transition to running them on a cluster, i.e. directly pass data between tasks instead of using local files

did I get that wrong?

should I be doing a flow of flows here, where I map chunks of the workload to child flows? (Though I’d find that a bit excessive, since effectively I’d be mapping my workload twice, once on the flow level and then on the task level)

I thought they should contain the data and the S3 result handler would upload them so that I could later resume if need be

but understood, in that case I actually need to do the download and upload in one task

Let me discuss with the team a bit more and try to provide some more insight

Hi @Dylan, just wondering if you had a chance to talk to the team about the way memory is managed for a flow?

Got a little more clarity here

The short answer is that Prefect doesn’t do memory management, so it’s the same behavior that you might see if you wrote a script yourself. Variables will be garbage collected once Python determines there are no more references to that variable — all task return values will be referenced via state objects and therefore will live in memory for the duration of the flow run. Running a workflow on a dask cluster helps because any intermediate memory usage within your task (that isn’t referenced in your final return value) will be run on a dask worker and cleaned up more aggressively, but the final return value will still get collected into memory back in the main flow runner process

If you’re using a LocalDaskExecutor, using process workers will clean up intermediate memory but thread workers will not