dask/dask

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

aavbsouza

@aavbsouza

@aamirshafi maybe it is useful https://arxiv.org/pdf/1908.05790.pdf

Scott Sievert

@stsievert

Thanks for that paper @aavbsouza

aavbsouza

@aavbsouza

Hi, I have been playing with the dask array transpose and einsum. But to truly transpose
the data it was necessary to follow the tranpose/einsum operation with a rechunk operation.
Is this behavior expected? thanks =)

Scott Sievert

@stsievert

Does “truly transpose” mean “perform the computation for the transpose”?

aavbsouza

@aavbsouza

yes, as in for instance distributed 3d fft

Paul B

@pborgen

Anyone run into a issue of when you open the dask gui to look at the status it locks up? Seems like the only screen that causes the gui to lock up is the status screen

@Jacob-Barhak I ran into the same issue....I wanted to organize my code better so I started by creating a class then I referenced that class in my task that I submitted...I am not sure how I make that module available to all the workers.

Scott Sievert

@stsievert

@aavbsouza that might be expected. Dask is declarative, so no computation is performed until “dask.compute” is called. Rechunking might trigger the same operation.

What does the Dask Dashboard look like? That’s a really useful tool for diagnosing these performance issues.

Paul B

@pborgen

Anyone know how to lower the refresh rate on the dask gui status page....I am running my app at a customer that has one of those web page vpn access setups...The rdp is already laggy and when I go to the status page it locks up....My computations still finish those....Runs fine on my laptop :-)

aavbsouza

@aavbsouza

@stsievert I was trying to trigger the computation using the persist function. I cannot use the compute since, as far I am understand it causes the result to be send to the main program.

aamirshafi

@aamirshafi

@aavbsouza Thanks for the paper.

Scott Sievert

@stsievert

@aavbsouza “send to the main program”? “Persist” stores an object in memory. I often persist DataFrames so I don’t read from disk repeatedly. What do you mean “causes the result to be send to the main program”?

odovad

@odovad

Hi all, I just raised an issue on github; if you can have a look.
Not sure why it does what it does lol
dask/dask#6357
Thanks ;)

Tom Augspurger

@TomAugspurger

@odovad typically you won’t need to ping in multiple channels. Just raising an issue is sufficient.

Responding there now.

aavbsouza

@aavbsouza

@stsievert I was using persist to keep the data in the distributed memory instead of compute, sorry for the confusion

@stsievert However even after the persist the data is not tranposed only after the rechunk

odovad

@odovad

@TomAugspurger Ok, thanks.

Scott Sievert

@stsievert

@aavbsouza this is a question for SO I think. I need more context. After you post on SO could you put the link here?

aavbsouza

@aavbsouza

@stsievert Actually there is a question on SO (https://stackoverflow.com/questions/56736851/how-if-to-use-dask-to-transpose-distributed-3d-numpy-arrays)

aavbsouza

@aavbsouza

it not mine but is the same problem. Actually the link suggested by the author have some good diagrams (1D slab decomposition for mine problem). As far I understand when I call the tranpose or the einsum routines and after that the persist() function the dask runtime operate at chunk level (if necessary tranposing the chunks) and change the logical shape (chunk shapes, "global shape") but the original number of chunks of the array is preserved. However the data is not actually moved across chunks . I believe that is the expected behavior for people coming for instance from MPI backgrounds (where the MPI_alltoall* functions would do that).

Paul B

@pborgen

Trying to get a list of unique index in a simple array....anyone know how to do this?

device_irn_list = list(ddf.index.unique()) does not do it

Leo Singer

@lpsinger

Is there some documentation of the lifecycle of dask-jobqueue clusters? I am looking for answers to questions like, "Are workers automatically killed when the process in which the PBSCluster is declared exits?"

Leo Singer

@lpsinger

And if I submit multiple dask-jobqueue applications from the same user account, is there any risk of crosstalk between them?

Scott Sievert

@stsievert

@aavbsouza so does that answer your question?

aavbsouza

@aavbsouza

Hi @stsievert . If he rechunk is the correct approach there are no questions anymore. But I believe that should be useful to add to the documentation a note about this behaviour. Since it could be unexpected for some people. Thanks for your help

Scott Sievert

@stsievert

@aavbsouza I think that’d be a great addition to the documentation too! You’re probably the best person to make that change because you’re closest to the issue. The Dask project will gladly take a PR!

Tamas Petkovich

@tpetkovich

Hi! Any idea why my distributed scheduler (with Localcluster) starts to sync compute with a client.persist([list of delayed objects]), instead of providing a future? And it is also not using more cores than 1.

TimoRoth

@TimoRoth

Looking at this script: http://dpaste.com/1XXD46D
The documentation at https://distributed.dask.org/en/latest/manage-computation.html#client-compute very clearly says "returns a single future", but it returns a list of futures for me.

Am I doing something wrong, looking at the wrong docs or something?

Paul B

@pborgen

Anyone else ever have the dask status page lock up the gui in their vm?

seertaak

@seertaak

Hi all, I'm using dask-kubernetes to scale a cluster of - for now - 10 pods. However, I am never able to create more than 5 pods: after the fifth, I get the error message

kubernetes_asyncio.client.exceptions.ApiException: (403)
Reason: Forbidden
HTTP response headers: <CIMultiDictProxy('Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'Date': 'Mon, 06 Jul 2020 14:25:58 GMT', 'Content-Length': '301')>
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"pods \"dask-root-b0b73888-d4cd4m\" is forbidden: exceeded quota: pods-high, requested: pods=1, used: pods=10, limited: pods=10","reason":"Forbidden","details":{"name":"dask-root-b0b73888-d4cd4m","kind":"pods"},"code":403}

Note:

I can create a deployment of 10 machines without issues.
I can create 10 pods by applying a single pod (with spec matching that provided to dask kubernetes) ten times.
If I type:

$ kubectl -n some=namespace describe resourcequota
Name:       pods-high
Namespace:  asf-forecast-parameters
Resource    Used  Hard
--------    ----  ----
pods        5     10

I see the string "pods-high" mentioned in the error message above, and it is associated with a "Used" limit of 5 pods. I wonder if that's the reason I can only get 5 dask nodes?

Has anyone seen this kind of error before? What is the right way to get past it?

Many thanks in advance

Paul B

@pborgen

Calling all dask wizards. I am having a performance issue and wanted to see if someone can do a 60min screen share with me to review my approach. I am in EST. I can paypal of venmo 100 USD. My email is pborg99@yahoo.com if someone can help

ldacey

@ldacey_gitlab

would dask be reading all files ultimately in this scenario? I have partitioned data by created_date but some end users want to have an aggregated report per ticket ID. A ticket ID can exist on many of the individual partition files, and I would need to capture some min and max timestamps etc.

Fabio Luporini

@FabioLuporini

Hi everybody, since a few days we're seeing "random" failures in our CI due to distributed emitting:

tornado.application - ERROR - Exception in callback <bound method Nanny.memory_monitor of <Nanny: None, threads: 1>>
Traceback (most recent call last):
  File "/opt/hostedtoolcache/Python/3.7.7/x64/lib/python3.7/site-packages/tornado/ioloop.py", line 907, in _run
    return self.callback()
  File "/opt/hostedtoolcache/Python/3.7.7/x64/lib/python3.7/site-packages/distributed/nanny.py", line 414, in memory_monitor
    process = self.process.process
AttributeError: 'NoneType' object has no attribute 'process'

Now, I'm not sure whether this is a bug on our side, your side, or simply our side doing something which makes latest distributed unhappy, causing this error. Said that the AttributeError... doesn't make much sense to me.
I'd welcome any sort of feedback. If you think it's an actual distributed issue, I'll be happy to file a bug report on github. If you think it's our fault, well I'm all ears for guidance.
Some more context: the error is triggered in a jupyter notebook by a cell which calls scipy.minimize from a DASK worker. Link here but probably not very informative...
Thanks!

ldacey

@ldacey_gitlab

                instance_count=pd.NamedAgg(column='instance_id', aggfunc='nunique'),
                met_count=pd.NamedAgg(column='met_count', aggfunc='sum'),
                breach_count=pd.NamedAgg(column='breach_count', aggfunc='sum'),
                sla_count=pd.NamedAgg(column='sla_count', aggfunc='sum'),
                avg_met_seconds=pd.NamedAgg(column='met_seconds', aggfunc='mean'),
                total_met_seconds=pd.NamedAgg(column='met_seconds', aggfunc='sum'),
                avg_breach_seconds=pd.NamedAgg(column='breach_seconds', aggfunc='mean'),
                total_breach_seconds=pd.NamedAgg(column='breach_seconds', aggfunc='sum'),
                avg_seconds=pd.NamedAgg(column='status_calendar', aggfunc='mean'),
                total_seconds=pd.NamedAgg(column='status_calendar', aggfunc='sum'),
            )

is something like that possible in dask? I need to aggregate the same column in multiple ways

Paul B

@pborgen

I have a large dasks data frame. and have set a index and divisions. When I do a ddf.loc[value_in_the_index].compute() it still goes pretty slow....Wanted to see if anyone have any ideas on how to improve the performance of this operation?

Leo Singer

@lpsinger

I have several command line tools that have a -j option (inspired by make's option) which tells them to use Python multiprocessing. I am thinking of using dask-jobqueue so that these tools can use local jobs, PBS etc. for launching workers to handle their parallel sections. Do you know of established conventions for how to present this through the command line arguments for my tool? Some possibilities that occurred to me include foo -j 10 --cluster pbs or foo -j pbs:10.

Talley Lambert

@tlambert03

small question: is it possible to store compute kwargs on a particular dask object? for instance, instead of x.compute(scheduler="single-threaded"), somehow specify that tasks on x should always be single-threaded (but only on x if that's not the global.config)?

Benjamin Zaitlen

@quasiben

@tlambert03 I don't know of a away to do what you are asking but you could use a context manager if you don't want to mess with a global config: with dask.config.set(scheduler='single-thread'):

Talley Lambert

@tlambert03

thanks. yeah I saw the context manager. The (probably rare) use case here is when you're passing around dask objects (e.g. with other libaries) and you're not sure when/where they're going to call compute on it. I'm sure it's probably still a sign of a bad design 😅 was just curious if it was a supported scenario

Matthew Rocklin

@mrocklin

You could make your own class and define the __dask_scheduler__ method.

@tlambert03

But otherwise, no

Talley Lambert

@tlambert03

thanks @mrocklin! that'll do it!

Leo Singer

@lpsinger

When is the next releases of dask-jobqueue coming? I need dask/dask-jobqueue#405 in a release

Matthew Rocklin

@mrocklin

I recommend asking on the dask-jobqueue issue tracker. Very few of the dask maintainers track this chat room.

rmsrms1987

@rmsrms1987

Hey Everyone, is there a way to share a variable between dask delayed processes? For example, within each process I want to append to a list, and each subsequent process would have access to this list.

Where communities thrive

People

Repo info

Activity