Hi Team, using Prefect Server (version 0.14.1) wit...
# prefect-server
Hi Team, using Prefect Server (version 0.14.1) with Dask Gateway. I am using the
method under Flow to run the same task in parallel for different inputs. It is working fine but not using all the Dask workers, which is causing Dask to run into resource/communication errors. How can I tell Prefect flow to run only one mapped task per worker?
Error from Dask worker logs
This mapped task has 15 runs but all running on the same worker - 1 or 2 at a time. And after 12 runs, worker shows this message above and the flow does not proceed further at all.
I am not sure. Does your cluster have the workers beforehand or does it autoscale up?
GKE cluster has autoscaling enabled. Dask cluster is spun up when the Flow run starts.
How many workers does it start with?
I see that Dask cluster is spinning up all the workers I requested but it is using only one to run all iterations of the mapped task. Hence, running into errors.
I have set it to have minimum 5. Tried upto 20 as well. It is able to start all those workers, just doesn’t use them.
Can I see how you set up your Dask cluster?
Someone does seem wrong with the scheduler, what behavior do you see without Prefect?
Using this function with the
and this is how the cluster is setup:
Copy code
def create_cluster(min_workers, max_workers, docker_image, proxy_address):
        secret_client = secretmanager.SecretManagerServiceClient.from_service_account_json(

        response = secret_client.access_secret_version(
        secrets = json.loads(response.payload.data.decode('UTF-8'))

        # connect to dask-gateway proxy
        auth = BasicAuth(password=secrets['dask_gateway_secret'])
        gateway = Gateway(address=proxy_address, auth=auth)

        options = gateway.cluster_options()

        options.environment = {

        options.image = docker_image
        options.worker_memory = 80  
        options.worker_cores = 8 
        cluster = gateway.new_cluster(options)
        cluster.adapt(minimum=min_workers, maximum=max_workers)
        return cluster
Nothing stands out here. I think you’d need to try using the Dask cluster without Prefect to see the behavior. I haven’t been able to dig anything around this Dask issue where the workers are not being used by the scheduler
I tried using Dask on its own and its behaving as expected with mapping or distribution. Also, only one of the mapped Prefect tasks is using single worker, all other mapped tasks in the flow are using multiple workers.
Ah okay that sounds like it may have been treated as a reduce. Do you have a minimal example?
It like has to do with chaining mapped tasks right?