dask/dask

Please use Stack Overflow with the #dask tag for usage questions and github issues for bug reports

Michał Jastrzębski

@inc0

or is there better way to achieve this result?

Apostolos Vlachopoulos

@avlahop

Interesting paper: https://legion.stanford.edu/pdfs/legate-preprint.pdf

Davis Bennett

@d-v-b

@avlahop is the code on github? I couldn't find a link in the paper

Isaiah Norton

@ihnorton

is it possible to show all of the annotations in the dashboard at once?

Apostolos Vlachopoulos

@avlahop

@d-v-b I'm sorry not sure. I couldn't find it either from a first brief reading

Dustin van Weersel

@dvweersel

I'm configuring Dask on a single node together with a JupyterHub instance. I would like to specify a default LocalCluster when people create one. I found the distributed.yaml in /etc/dask. Can I specify stuff like n_workers, threads_per_worker anywhere there? I've been looking for a complete list of configuration options, but I'm unable to find it

Bence Nagy

@underyx

heya, I had a pretty large graph (around 20k tasks) and when I clicked on the Tasks tab on the web UI, the scheduler seems to have crashed and I lost a couple hours of processing effort

is this expected or did I misconfigure things somehow?

Martin Durant

@martindurant

I’m not sure what the largest output we’ve tested with that view is. It’s conceivable that yours is the largest. Was there no error message? Probably you should raise an issue on distributed, with all the information you can gather.

Bence Nagy

@underyx

yep, that's fair enough

there wasn't any error message as I think the container just ran out of memory

mainly I just wanted to confirm whether there's any way to be resilient against such failures

Martin Durant

@martindurant

You should be able to find the relevant parts of the graphical output quite easily. It ought to be possible to make aggregate graphics (like the memory and processing sections on the status/ dashboard) or just refuse to work with datasets that have become too large, or even to monitor internal memory usage of the dashboard and halt when it gets too big.

Bence Nagy

@underyx

I think that would be great indeed, I worry about people unknowingly crashing production workloads by just randomly clicking around on the dashboard

(as I've just done myself :D)

ended up commenting here https://github.com/dask/distributed/issues/2979#issuecomment-530915742

Martin Durant

@martindurant

It is, of course, in a separate thread, as opposed to separate process, so it will always be a little tricky to know when some memory limit is approached because of the graphics; however, I can imagine safety safeguards which turn off some graphics known to be resource-hungry above some threshold

TomMaullin

@TomMaullin

Hi all, I'm pretty new to Dask and just had a quick question I was wondering if anyone could help me with - I have two 3D dask arrays and want to broadcast the solve operation across the last 2 dimensions, i.e. given A and B I want to do C[i,:,:]=solve(A[i,:,:], B[i,:,:]) for all i in the leading dimension... I tried the below but it seems to be really slow (slower than numpy) - does anyone know what I'm doing wrong/what I could do better? Sorry if this was the wrong place to ask.

C = da.apply_gufunc(np.linalg.solve,  "(i,j),(i,k)->(j,k)", A, B, vectorize=True,output_dtypes=A.dtype)

mattchan-tencent

@mattchan-tencent

Hi, I've got a question about the scale function in the cluster class. From the source, it doesn't look like scale_up does anything.

Does it just set scale and then read it back then scale up/down from that?

David Hoese

@djhoese

@TomMaullin You may want to post on stackoverflow with the dask tag. You'll get a wider audience there and a better answer.

Apostolos Vlachopoulos

@avlahop

is it possible to scatter a small dataframe and join it with a larger one kind like Spark uses the Broadcast Join? Will we see a performance gain on that?

TomMaullin

@TomMaullin

ah thank you! I believe the question is already on overflow with no reply - seeing it on there is what made me realise this would be an issue for my own code also

TomMaullin

@TomMaullin

https://stackoverflow.com/questions/57912503/broadcasting-solve-in-dask

Arpit Solanki

@arpit1997

@mrocklin Do you think a maximum limit of workers which can be used by a submitted job would be useful, Although I see that we already have a param for providing a set of workers.

On a broader idea worker pools, something like this could be rather useful.
Can these two behaviours be achieved currently?

Arpit Solanki

@arpit1997

@mrocklin I can open an issue in dask/distributed if this should/can be implemented.

Matthew Rocklin

@mrocklin

Hi @arpit1997 ! Yes, I think that raising an issue for this kind of discussion would be useful. Most of the development conversation happens on GitHub rather than Gitter. It's good to raise issues there even for things about which you are unsure.

evalparse

@xiaodaigh

Does anyone has a good book recommendation for learning Dask?

Jim Crist-Harif

@jcrist

I think there's a few books people have written on using dask, but AFAIK none of the authors are affiliated with the project. We generally recommend people walk through the documentation. There are also a few tutorials around, this is the one we usually give: https://github.com/dask/dask-tutorial

evalparse

@xiaodaigh

thanks. after reading through the documentation I don't feel i get a good grasp of the innards. like how many workers are there etc.

Davis Bennett

@d-v-b

does anyone else get frustrated with map_blocks

Davis Bennett

@d-v-b

specifically, I often find myself trying to apply an arbitrary function on chunks of an array, but the output of that function is not an array (image registration is usually where this comes up for me, second most common is trying debug chunked operations, e.g. map_blocks(func_that_returns_a_debug_string))

but map_blocks assumes that I'm mapping an array on to another array, which doesn't match these use cases well

Davis Bennett

@d-v-b

i wonder if there would be an appetite for something like da.apply_blocks() that would not be expected to return a dask array, to enable generalized chunkwise operations

RuiLoureiro

@RuiLoureiro

I use pytest to test my application. Since it relies on dask, I need to have a local cluster as a fixture.

I tried to do it myself and close the cluster on the fixture cleanup. Even though it works, I get a warning with more than 100 lines because dask tries to communicate with the cluster after it has been closed.

I use pytest to test my application. Since it relies on dask, I need to have a local cluster as a fixture.

I've also tried to use the client fixture from distributed.utils_test, but that gives me the following error

file /Users/ruiloureiro/.virtualenvs/tarsier/lib/python3.7/site-packages/distributed/utils_test.py, line 561
@pytest.fixture
  def client(loop, cluster_fixture):
E       fixture 'loop' not found
>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, client, cov, doctest_namespace, monkeypatch, no_cover, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory
>       use 'pytest --fixtures [testpath]' for help on them.`

Scott Sievert

@stsievert

Would it be possible to mock Dask’s response in the tests? https://github.com/pytest-dev/pytest-mock/

Alessio Civitillo

@acivitillo

Hello, wondering if the retries argument inside dask.compute() retries only failed tasks or all tasks? https://distributed.dask.org/en/latest/api.html#distributed.Client.compute

Martin Durant

@martindurant

^ should be failed tasks. If the dependencies went away in the meantime, some may need to be recalculated.

David Brochart

@davidbrochart

When I have a computation launched in the background with persist(), followed by a call to progress to have a progress bar, is there a way to wait for the computation to finish and still have the bar rendering? compute or wait will block including the bar rendering.

David Brochart

@davidbrochart

Oh maybe by using client.persist and then checking status?

David Brochart

@davidbrochart

But that needs event loop integration

joeschmid

@joeschmid

Hi Dask community, thanks for a great project -- we're shifting a lot of our data science work onto Dask (+ Prefect, potentially) and we've had a good experience. Question on heterogeneous workers: we have need for different types of workers, e.g. high memory, GPU, etc. We use dask-kubernetes (on AWS EKS) which currently just supports a single k8s pod spec for all workers. I'm about to start on a potential PR in dask-kubernetes that would allow for different worker types, but only in non-adaptive mode. Before I dive in, I thought I'd ask if anyone knows of existing work in this area. I see this old issue: dask/distributed#2118 but haven't found anything else. Thanks in advance!

RuiLoureiro

@RuiLoureiro

Hey all, my application requires me to launch tasks from within other tasks, like the following

def a():
    # ... some computation ..

def b():
    # ... some computation ..

def c():
    client = get_client()
    a = client.submit(a)
    b = client.submit(b)

    [a,b] = client.gather([a,b])

    return a+b

client = get_client()
res = client.submit(c)

However, I would like to have access to the intermediate results a and b, but only c shows up in client.futures.

Is there a way to tell dask to keep the results for a and b?

Thank you

joeschmid

@joeschmid

@RuiLoureiro I think you want to use a worker_client in your task c like this:

from dask.distributed import worker_client


def c(n):
    with worker_client() as client:
        client = get_client()
        a = client.submit(a)
        b = client.submit(b)
        [a,b] = client.gather([a,b])
    return a + b

RuiLoureiro

@RuiLoureiro

@joeschmid How should I access a and b? client.futures still only shows c.

joeschmid

@joeschmid

@RuiLoureiro how about just returning a and b explicitly in c(), e.g.

from dask.distributed import worker_client


def c(n):
    with worker_client() as client:
        client = get_client()
        a = client.submit(a)
        b = client.submit(b)
        [a,b] = client.gather([a,b])
    return [a + b, a, b]

RuiLoureiro

@RuiLoureiro

@joeschmid This example is a simplification of my application and making every task return its dependencies would be overly complicated, even more so because I have several levels of dependency (not only one level as like the example).

joeschmid

@joeschmid

@RuiLoureiro got it, that makes sense. It might be overkill, but I suppose you could build up the entire tree of intermediate results. One open source project that I'm familiar with runs on Dask and does that: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/engine/flow_runner.py The key line:

task_states[task] = executor.submit(
                        self.run_task,
                        task=task,
                        state=task_state,
                       ... snipped ...
                    )

We run large Flows with Prefect on Dask. When our Flow Run is finished, we get back a results object that we can then examine for the state and return value of the various tasks. Depending on what you're doing, it might be easier to just use Prefect to build your flow and run it on Dask. It makes it very easy to return data from tasks and build up multi-step flows that run on Dask. We've had a good experience with it.

RuiLoureiro

@RuiLoureiro

@joeschmid Very interesting, will look into it. Even if we don't end up using the library, it's always interesting to see how other people build on top of dask. Thank you very much!

garanews

@garanews

is it possible to have 2 different dask cluster running on same machine?

joeschmid

@joeschmid

@RuiLoureiro You bet. Also, Prefect has a really helpful public Slack community available at prefect-community.slack.com . You can post questions there and get helpful, fast responses from their team.

bgoodman44

@bgoodman44

Anyone here?

Where communities thrive

People

Repo info

Activity