Question on using an external Dask cluster (DaskExecutor) running in adaptive mode -- we notice that running a Flow that maps over elements (say 100 elements) only seems to use the number of Dask workers that existed at the time the flow run started. We would have expected the Dask cluster to begin scaling up workers (since it's in adaptive mode) and starting to run more Prefect tasks in parallel. I'm not sure we've really characterized this behavior well, but we've observed it enough that it seems to be a pattern. (We have a notebook that - using the k8s APIs - plots the number of requested workers, running workers, & nodes and could show some of what we're seeing if it would help.)

For my understanding you’re running into a scenario where you have a dask cluster in adaptive mode and (for example) you have N workers, start the flow run, cluster scales up to M workers, but it is still only using the original N?

could this be related to work stealing being disabled? the prefect docs say to do so but it does seem like that limits the effectiveness of an adaptive cluster

@josh what we see is that the cluster doesn't even try to scale up beyond the original N workers. There are a lot of available (Pending) Prefect Tasks but in our case only 2 dask workers running 4 Tasks (2 vCPUs for each worker).

Interesting theory @Brett Naul I think I’ve seen an issue similar to this in the past with not every dask worker being used from an adaptive standpoint. I was using the

dask-kubernetes

library and I can’t say for sure if I had work stealing disabled

@Brett Naul We did recently disable work stealing, but saw the same behavior prior to that.

Are you using dask-kubernetes for your cluster?

Yup, running on AWS EKS.

Interesting, I wonder if the dask scheduler thinks it doesn’t need to create more workers for itself. I’m trying to recreate this on a cluster to see if I get the same issue

In case it helps, one aspect of the behavior we observed is that N Prefect tasks start running initially where N = w * c (w = number of initial Dask workers, c = vCPUs which is the same as nthreads that we start the workers with)

There’s a couple things I think could be happening: 1. It’s starting with the amount of processes available (2 workers w/ 2 CPUs so 4 total) and sticking with that so it doesn’t think it needs to create more workers (e.g. only ever trying to do 4 things at once) 2. The mapped tasks aren’t trying to be executed in parallel but instead are happening sequentially and therefore dask isn’t creating more workers because it doesn’t think it needs to cc @Chris White

@josh thanks for the test case! We can give this a shot. Much appreciated.

Just catching up - this is interesting; we could also explore replicating this with Dask directly and file a bug report; under the hood, Prefect makes a call to

Client.map

https://distributed.dask.org/en/latest/api.html#distributed.Client.map; in Prefect, this call is actually made with a

worker_client

, which could also be relevant

@Chris White it does seem like a potential bug/issue. At a minimum, it would certainly be desirable behavior to scale up workers when mapping.

yup yup! I’ll sync up with @josh and try to recreate with Dask - we’ll keep you in the loop!

@Chris White Sounds good. Thanks for the update, Chris.

@Joe Schmid / @Jie Lou are the Tasks that you are mapping very fast tasks? What’s a typical duration of the run for these tasks?

@Chris White they're quite slow -- often minutes.

Interesting. OK so that is not what I was expecting; just to give you some insight, while your tasks are running, you can do the following:

from distributed import Client

c = Client("<tcp://scheduler-address>")
c.run_on_scheduler(lambda scheduler: scheduler.adaptive_target())

and call this function over and over as your Tasks progress; (note you might need to exec into one of your pods to have access to the scheduler). This is the endpoint dask-k8s calls to determine whether scaling should occur — we found that for tasks whose runtime was on the same order as the network communication cost, this function would return 1, but for longer tasks it would very quickly begin returning large values — not a solution, but just so you know the approach we were taking

The update from our side is that we put our external Dask cluster in non-adaptive mode this morning and saw significantly better reliability today. (We did some work to make it easy to dynamically adjust our Dask cluster, e.g. adaptive on/off, set min/max # workers, set target # workers for non-adaptive, etc.)

That sounds good!