I m running into some memory limits with a flow using an ECS

Question

I m running into some memory limits with a flow using an ECSRun config on a LocalCluster DaskExecutor The first task in the flow is to generate a list of file keys and then I map over that list to process the files I would have expected that when an individual Mapped Child task completes any memory used in that task would have been released aside from the return value But it seems like that is not the case as the process eventually runs out of memory after several Mapped Children finish Is there something I can do in the configuration of the flow or task to mitigate this As an aside when I run this same flow using a FargateCluster it will complete without issue I m not sure if that s because the FargateCluster I m using just has more memory or if there s something inherently different between the two different DaskExecutor classes Thanks as always for any help you can provide

Anna Geller · Answer

can you share your ECSRun run config

Scott Aefsky · Answer

```ECSRun image=ECR IMAGE execution role arn=f arn aws iam account role prefect ecs execution role task role arn=f arn aws iam account role prefect ecs task role run task kwargs= cluster PrefectCluster labels= account ```

Anna Geller · Answer

and now your executor slightly smiling face how do you attach the executor to your flow

Scott Aefsky · Answer

And using a task definition of ```networkMode awsvpc cpu 1024 memory 2048 containerDefinitions name flow image not a real image logConfiguration logDriver awslogs options awslogs group prefect ecs task log ACCOUNT awslogs region us east 1 awslogs stream prefix ecs prefect awslogs create group True ```

Scott Aefsky · Answer

Exceutor is ```def dynamic executor if prefect context parameters use fargate == true return FargateCluster image= ECR IMAGE n workers= 5 Must specify n workers cluster arn= fr arn aws ecs us east 1 account cluster PrefectCluster task role arn= fr arn aws iam account role prefect ecs task role vpc= vpc subnets= subnet security groups= sg cloudwatch logs group=f prefect ecs task log account else return LocalCluster flow1 executor = DaskExecutor cluster class=dynamic executor ```

Anna Geller · Answer

in order for the executor to take effect it must be properly set in your Flow because it s read from storage It s best to attach it to the Flow directly like so ```with Flow yourflow executor=DaskExecutor cluster class=dynamic executor as flow ```

Scott Aefsky · Answer

I thought ```with Flow flow executor=myExecutor as f ``` and ```with Flow flow as f lt tasks gt f executor=myExecutor``` are functionally equivalent

Kevin Kho · Answer

They are but if you do ```from some other file import flow1 flow1 storage = flow1 executor = flow1 register ``` the executor is not stored because it s pulled from the Flow definition in storage Is it in the same file for you

Anna Geller · Answer

They are but sometimes people attach it too late and it must be available on the flow object in order for it to be used E g a common antipattern is ```with Flow flow as f lt tasks gt if name == main f executor=DaskExecutor ``` and main is not evaluated when flow is already registered a likely reason why your flow runs out of memory when using a local dask cluster is that it only has 1 CPU and 2 GB of memory so it sounds reasonable that in this setting your flow run can get OOM errors

Scott Aefsky · Answer

Yes everything is in the same file and in the same scope

Scott Aefsky · Answer

I get that increasing the memory for the memory would help but I am using relatively small batches I was expecting that each of the mapped tasks would only use the memory needed for what they return Is there some way I can tell Prefect Dask to dump the memory after the task is done

Kevin Kho · Answer

How many tasks are you mapping over We normally see issues around maybe the 50k mapped to 100k mapped task scale

Scott Aefsky · Answer

50ish

Scott Aefsky · Answer

Each task handles 500 small files

Kevin Kho · Answer

Ah ok Have I shown you <https www youtube com watch v=nwR6iGR0mb0|this> video to handle unmanaged memory

Scott Aefsky · Answer

You have not

Scott Aefsky · Answer

I will take a look

Kevin Kho · Answer

I dunno May or may not help release that memory

Scott Aefsky · Answer

Alright I ll give a couple of things a shot

Scott Aefsky · Answer

Unfortunately the jobs take a while so I won t know if it helped for a couple of hours

Scott Aefsky · Answer

Thanks as always

Kevin Kho · Answer

You added the env variable

Scott Aefsky · Answer

I m going to try to do that

Scott Aefsky · Answer

I m adding it to my default task definition ```networkMode awsvpc cpu 1024 memory 2048 containerDefinitions name flow environment name MALLOC TRIM THRESHOLD value 65536 image not a real image logConfiguration logDriver awslogs options awslogs group prefect ecs task log ACCOUNT awslogs region us east 1 awslogs stream prefix ecs prefect awslogs create group True ```

Scott Aefsky · Answer

Does that seem like the right way to do it

Kevin Kho · Answer

Looks right to me

Scott Aefsky · Answer

Alright I ll let you know if that improves things

Arseniy Aseev · Answer

Interesting I have run into a similar issue recently and thought I was losing my mind I m using a local DaskCluster and 64 threads are handling 100 1000 files worth 10 500 kB each but the tasks are long running and RAM usage grows relatively quickly tens of minutes to an hour to an insane 2 4GiB I have been debugging this for several days now Looking at the Dask dashboard it s clear that the amount of unmanaged memory keeps growing very fast and doesn t reset or go down when some tasks are done as you d expect If you set a memory limit on the worker it doesn t spill anything to disk and just crashes restarts The MALLOC TRIM THRESHOLD option is not available on my MacOS so it would be good to know if it works in your case Official documentation for reference <https distributed dask org en stable worker memory html automatically trim memory>