ghislainp
commented
1 year ago the get_dask_client function seems not to work when using an external dask cluster declared with DaskTaskRunner(address=data_cluster_address). I get an error "No global client found and no address provided". Below is one of the examples in README.md slightly adapted to run on an external dask cluster.
I investigated a bit and neither get_worker nor get_client work. It seems that the task 'process_data' is unable to know that it's running in a dask worker. I suspect this is related to other problems with external dask cluster, such as PrefectHQ/prefect-dask#47
I hacked prefect_dask.utils._generate_client_kwargs to use client_kwargs['address'] instead of calling get_client().scheduler.address, and it works now. I can make a PR. However, this is still not satisfactorily, because I need to secede and rejoin my task as explained here https://distributed.dask.org/en/stable/task-launch.html because I deadlock my cluster very easily.
I think that scheduling task with prefect_dask on a cluster, and running calculations with the same cluster is a useful pattern to deal with memory problems, and to monitor what happens on the cluster with the dask dashboard. Though, I'm open to other options, because I've been stuck on this for months...
import dask.dataframe
import dask.distributed
from prefect import flow, task
from prefect_dask import DaskTaskRunner, get_dask_client
# client = dask.distributed.Client()
@task
def read_data(start: str, end: str) -> dask.dataframe.DataFrame:
df = dask.datasets.timeseries(start, end, partition_freq="4w")
return df
@task
def process_data(df: dask.dataframe.DataFrame) -> dask.dataframe.DataFrame:
with get_dask_client():
df_yearly_avg = df.groupby(df.index.year).mean()
return df_yearly_avg.compute()
@flow(task_runner=DaskTaskRunner(address="tcp://127.0.0.1:8786"))
def dask_pipeline():
df = read_data.submit("1988", "2022")
df_yearly_average = process_data.submit(df)
return df_yearly_average
# dask_pipeline()
fjetter
zanieb
dchudz
jrbourbeau
giorgiobasile
Hi! I noticed recently that
get_dask_client(added in https://github.com/PrefectHQ/prefect-dask/pull/33) is not working with a Coiled cluster. I'm using Prefect Cloud.Problem
The following snippet does not work using a Coiled cluster (but does work with
dask.distributed.LocalCluster):When I run the above snippet I get a
TypeError: TLS expects assl_contextargument of type ssl.SSLContext (perhaps check your TLS configuration?) Instead got None.Full traceback
```python-traceback Traceback (most recent call last): File "/Users/sarahj/Downloads/prefect-dask-client.py", line 31, inLooking at the full traceback, it seems
_expect_tls_contextin https://github.com/dask/distributed/blob/main/distributed/comm/tcp.py expects TLS arguments, but is not getting the information needed.Since
get_dask_clientis a utility aroundworker_client, I checked to see if usingworker_clientworks with a Coiled cluster, and it does:Potential solution?
If I get the
clientthe waydask.distributed.worker_clientdoes, then the aforementioned failing snippet works. ie changing (lines 103-104 in get_dask_client) from:to:
It seems like there are other arguments
get_dask_clientmay need, so not saying this exactly will work, but thought it'd be a helpful start.