Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hi ,

I am having issues with DaskTaskRunner not not submitting tasks correctly. After a task is submitted it stays in pending state . The only way to get it to work is to set
```task_runner=DaskTaskRunner(cluster_kwargs={"processes": False})```
but then I am only using multithreading in one single process.

This used to work for me in the lower version of prefect but after upgrading to 2.19.19 this started happening

I have managed to solve this.

The solution seems to be importing the dask package and creating the dask cluster object before registering the flow.

```import dask.distributed
from prefect_dask import DaskTaskRunner

client = dask.distributed.Client()

@flow(
    name="xyz",
  task_runner=DaskTaskRunner(address=client.scheduler.address),
)
...```