Please use Stack Overflow with the #dask tag for usage questions and github issues for bug reports
But I just get this in my logs over and over:
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/comm/core.py", line 204, in _raise
raise IOError(msg)
OSError: Timed out trying to connect to 'tcp://10.46.128.14:36210' after 10 s: in <distributed.comm.tcp.TCPConnector object at 0x7feb07069128>: ConnectionRefusedError: [Errno 111] Connection refused
Shouldn't that trigger the death timeout after six of those?
Whole stack trace, if it helps, is:
tornado.application - ERROR - Exception in callback functools.partial(<bound method IOLoop._discard_future_result of <tornado.platform.asyncio.AsyncIOLoop object at 0x7feb77f9b278>>, <Task finished coro=<Worker.heartbeat() done, defined at /opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/worker.py:866> exception=OSError("Timed out trying to connect to 'tcp://10.46.128.14:36210' after 10 s: in <distributed.comm.tcp.TCPConnector object at 0x7feb07444240>: ConnectionRefusedError: [Errno 111] Connection refused")>)
Traceback (most recent call last):
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/comm/core.py", line 215, in connect
quiet_exceptions=EnvironmentError,
tornado.util.TimeoutError: Timeout
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/tornado/ioloop.py", line 743, in _run_callback
ret = callback()
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/tornado/ioloop.py", line 767, in _discard_future_result
future.result()
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/worker.py", line 875, in heartbeat
metrics=await self.get_metrics(),
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/core.py", line 747, in send_recv_from_rpc
comm = await self.pool.connect(self.addr)
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/core.py", line 874, in connect
connection_args=self.connection_args,
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/comm/core.py", line 227, in connect
_raise(error)
File "/opt/lsst/software/stack/python/miniconda3-4.5.12/envs/lsst-scipipe-1172c30/lib/python3.7/site-packages/distributed/comm/core.py", line 204, in _raise
raise IOError(msg)
OSError: Timed out trying to connect to 'tcp://10.46.128.14:36210' after 10 s: in <distributed.comm.tcp.TCPConnector object at 0x7feb07444240>: ConnectionRefusedError: [Errno 111] Connection refused
Hey I just posted a question on SO: https://stackoverflow.com/questions/57760355/how-should-i-load-a-memory-intensive-helper-object-per-worker-in-dask-distribute
Summary: I have a big object that I need per-worker and don't want to reinit every function call. Can dask handle this?
That work has not been done. The project was to experiment and see if meaningful speedups could have been attained by cythonizing some algorithms used by the scheduler. Note that any work done on this may or may not be merged, but it may be an interesting experiment. I personally have no time to mentor/respond to this issue, others may.
We don't have any large projects we're actively working on. Depending on the size of project you need for your class, you may be interested in issues in our issue tracker: https://github.com/dask/dask/issues. Otherwise, adding dask support to a library in the python ecosystem would be an excellent project. Dask itself is fairly - we're mostly focused on performance and small changes at this point.
The Python ecosystem however still has plenty of work that could be done to add parallelism support. Libraries that have added dask support include xarray, scikit-image, hyperspy, etc... Finding a Python library that is looking to add dask support would likely be a good project, and is something the dask dev team is actively interested in.
Hello everyone. Just wanted to know whether I can perform Dask array computations inside a delayed function, something like this.
def dask_delayed_example(X):
mean = X.mean(axis=0)
std = X.std(axis=0)
return X - mean
X = df.to_dask_array(lengths=True)
res = dask_delayed_example(X)Is Dask delayed only meant for parallelizing computation or can it also be used to perform computation on larger-than-memory datasets in a distributed manner?
Hi @varunbpatil, I'm far from being an expert on dask but I think I have a similar use case. And I asked about it a couple days ago in here. @bnaul pointed me at this recipe https://distributed.dask.org/en/latest/task-launch.html - might be what you're after.
I also recall reading in the docs that nesting delays that return delayed is a antipattern of sorts. So I think you want that task scheduling other tasks approach
Hi there ! I'm trying to use dask-ssh as follows:
% dask-ssh --scheduler master1 node1
---------------------------------------------------------------
Dask.distributed v1.27.0
Worker nodes:
0: node1
scheduler node: master1:8786
---------------------------------------------------------------
[ scheduler master1:8786 ] : /home/applis/anaconda/envs/py3v19.04/bin/python -m distributed.cli.dask_scheduler --port 8786
[ worker node1 ] : /home/applis/anaconda/envs/py3v19.04/bin/python -m None master1:8786 --nthreads 0 --host node1 --memory-limit autobut I cannot connect a client to the scheduler and logging does not provide more information. Client('master1:8786') fails with OSError: Timed out trying to connect... and I can not access the web ui at master1:8787. However, running dask-scheduler and dask-workerwork fine. Any suggestion ?
i'm asking because I noticed that using
da.from_array(large_numpy_array) was blowing up workers, as (counterintuitively) the large_numpy_array is not actually partitioned along the expected chunks. Each worker appears to get a full copy of large_numpy_array regardless of the chunking.
i'm looking into the relevant code to see if there's a simple fix
No, the workers do not get a graph at all, they get tasks with inputs. Depending on the chunking scheme and the operation you are doing, tasks may require many pieces of the overall array. However, if you are doing from_array, it would seem it fits in memory anyway, so Dask is probably not helping you.
it just seems weird to me that
from_array(large_array, chunks=chunks)[0].compute(), for example, does not allocate data to workers the way one would expect from the chunking
in terms of tasks, i'm guessing that the first task (array creation) requires the entire array, so every subsequent task inherits that dependency.
I guess it shouldn't be too hard to kick off dask-jobs from within Flask. Was just wondering whether there are any examples and experiences doing that around.
Not having to setup message queues and the like seems like a huge plus to me.
but thanks for the documentation link
it looks like all these methods target optimal use of data that has been localized to the cluster using e.g.
scatter, is that right?
in my use case, there's no explicit management of data on the cluster -- I just locally define a dask array and try to compute its first element (for example) using the distributed scheduler
Yes, about manully moving things around in cluster memory. https://distributed.dask.org/en/latest/work-stealing.html describes when Dask might do this automatically
You should isolate the exact code, expectations and findings in an issue, then
hello everyone, I have question. I'm trying to run computation that would require large dataset as argument. Something like df.apply(navigate_over_array, args=(very_big_array,)). I'd like to pre-scatter very_big_array to all workers so function can reference it there. What's good way to pass it? If I scatter, I get future, is there a way that would allow me to access data behind this future from applied function in worker?
or is there better way to achieve this result?
