Cannot call compute() on a dask dataframe when inside a task

[pborgen]

Below is the code to reproduce. I am running this with the latest version. Specified below.

I have a dedicated vm for the scheduler. I have other vm's for the workers. I have seen this pass before when I only have one worker running. Seems to always fail when I have 2 workers.

Related Issue:
#2336

Version:
dask==2.16.0
distributed==2.16.0

How I start the scheduler:
dask-scheduler

How I start my workers:
dask-worker tcp://x.x.x.x:8786 --nanny-port 8001 --worker-port 8002 --no-dashboard

from distributed import Client, get_client
import dask

if __name__ == '__main__':

    def getCompute(id):
        ddf = get_client().get_dataset("foo")

        #filter
        return ddf[ddf.id == id].compute()

    daskClient = Client('x.x.x.x:8786')
    daskClient.restart()

    ddf = dask.datasets.timeseries()
    ddf = ddf.persist()
    daskClient.publish_dataset(foo=ddf)

    #get id's
    id_list = ddf['id'].unique().compute()

    futures = []
    try:
        for id in id_list:
            futures.append(daskClient.submit(getCompute, id))
    finally:
        daskClient.unpublish_dataset("foo")

    results = daskClient.gather(futures)

Traceback (most recent call last):
  File "D:/dev/code/netsense.support/facts_ingester/facts_ingester/test.py", line 29, in <module>
    results = daskClient.gather(futures)
  File "D:\dev\code\netsense.support\facts_ingester\facts_ingester\venv\lib\site-packages\distributed\client.py", line 1961, in gather
    return self.sync(
  File "D:\dev\code\netsense.support\facts_ingester\facts_ingester\venv\lib\site-packages\distributed\client.py", line 815, in sync
    return sync(
  File "D:\dev\code\netsense.support\facts_ingester\facts_ingester\venv\lib\site-packages\distributed\utils.py", line 347, in sync
    raise exc.with_traceback(tb)
  File "D:\dev\code\netsense.support\facts_ingester\facts_ingester\venv\lib\site-packages\distributed\utils.py", line 331, in f
    result[0] = yield future
  File "D:\dev\code\netsense.support\facts_ingester\facts_ingester\venv\lib\site-packages\tornado\gen.py", line 735, in run
    value = future.result()
  File "D:\dev\code\netsense.support\facts_ingester\facts_ingester\venv\lib\site-packages\distributed\client.py", line 1826, in _gather
    raise exception.with_traceback(traceback)
  File "D:/dev/code/netsense.support/facts_ingester/facts_ingester/test.py", line 10, in getCompute
    return ddf[ddf.id == id].compute()
  File "c:\python38\lib\site-packages\dask\base.py", line 166, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "c:\python38\lib\site-packages\dask\base.py", line 444, in compute
    results = schedule(dsk, keys, **kwargs)
  File "c:\python38\lib\site-packages\distributed\client.py", line 2646, in get
    used as an optimization to avoid recomputation.
  File "c:\python38\lib\site-packages\distributed\client.py", line 2543, in _graph_to_futures
    
ValueError: Inputs contain futures that were created by another client.

[jcrist]

Hmmm, I'm unable to reproduce with the following, but I may be missing what's causing the issue:

from distributed import Client, get_client
import dask.datasets


def do_compute(id):
    ddf = get_client().get_dataset("foo")

    #filter
    return ddf[ddf.id == id].compute()


def main():
    with Client() as client:
        ddf = dask.datasets.timeseries()
        ddf = ddf.persist()

        #get id's
        id_list = ddf['id'].unique().compute()

        client.publish_dataset(foo=ddf)
        try:
            futures = client.map(do_compute, id_list)
            results = client.gather(futures)
        finally:
            client.unpublish_dataset("foo")


if __name__ == '__main__':
    main()

Note that I did have to update your test code to avoid a race condition with the dataset being unregistered before the do_compute task starts (you need to gather the results first before unregistering), but even with the race condition I didn't see the same issue.

[pborgen]

@jcrist Yeap thanks. This is nicer...Are you running a couple workers on different vm's?

[pborgen]

I can reproduce the error with the below

from distributed import Client, get_client
import dask



def getCompute(id):
    ddf = get_client().get_dataset("foo")

    #filter
    return ddf[ddf.id == id].compute()


def main():
    with Client('x.x.x.x:8786') as client:
        client.restart()
        ddf = dask.datasets.timeseries()
        ddf = ddf.persist()

        # get id's
        id_list = ddf['id'].unique().compute()

        client.publish_dataset(foo=ddf)
        try:
            futures = client.map(getCompute, id_list)
            results = client.gather(futures)
        finally:
            client.unpublish_dataset("foo")
            
if __name__ == '__main__':
    main()

[crusaderky]

Just ran your code with 2 workers, 1 thread per worker - can't reproduce.
Are you sure you're using distributed 2.16.0 or higher?

[crusaderky]

Just retested on dask-docker, with 2 workers on different hosts (1 thread, 1 process per worker) and the scheduler on a third, separate host. Still can't reproduce.

[pborgen]

Darn.....I am just running on some azure vm's.....With I do a "dask-worker --version" I get 2.16.0. For the scheduler also "dask-scheduler --version" 2.16.0....

Setup
1 vm - scheduler
2 vm's workers
1 dev box running the script

Maybe there is some type of dependency that needs to be updated. Or maybe I just run docker....The cli way was so easy to get going...I have used docker in the past.

I run all my workers and my scheduler with the below commands via the cli
dask-worker tcp://10.1.14.5:8786 --nanny-port 8001 --worker-port 8002 --no-dashboard
dask-scheduler

"pip freeze" on my dev box

altgraph==0.16.1
atomicwrites==1.3.0
attrs==19.1.0
bokeh==2.0.2
boto3==1.9.176
botocore==1.12.176
click==7.1.2
cloudpickle==1.4.1
colorama==0.4.1
cx-Oracle==7.1.3
cycler==0.10.0
dask==2.16.0
distributed==2.16.0
docutils==0.14
envparse==0.2.0
findspark==1.3.0
fsspec==0.7.3
future==0.17.1
gitdb==4.0.2
GitPython==3.1.0
HeapDict==1.0.1
importlib-metadata==0.18
Jinja2==2.11.2
jmespath==0.9.4
kiwisolver==1.2.0
locket==0.2.0
MarkupSafe==1.1.1
matplotlib==3.2.1
more-itertools==7.1.0
msgpack==1.0.0
numpy==1.18.4
packaging==19.0
pandas==1.0.3
partd==1.1.0
pefile==2019.4.18
Pillow==7.1.2
pluggy==0.12.0
psutil==5.7.0
psycopg2==2.8.4
py==1.8.0
py4j==0.10.7
pyarrow==0.17.0
PyInstaller==3.5
pyparsing==2.4.0
pypiwin32==223
pyspark==2.4.5
pyspark-stubs==2.4.0.post7
pytest==4.6.3
python-dateutil==2.8.0
pytz==2019.1
pywin32==227
pywin32-ctypes==0.2.0
PyYAML==5.3
ruamel.yaml==0.16.10
ruamel.yaml.clib==0.2.0
s3transfer==0.2.1
scipy==1.4.1
six==1.12.0
smmap==3.0.1
sortedcontainers==2.1.0
tblib==1.6.0
toolz==0.10.0
tornado==6.0.4
typing-extensions==3.7.4.2
urllib3==1.25.3
wcwidth==0.1.7
xmltodict==0.12.0
zict==2.0.0
zipp==0.5.1

[crusaderky]

What's the OS on the scheduler, workers, and the client?
Is it 32 or 64 bit python?
How exactly did you install your stack?

[crusaderky]

I strongly suggest you

1. completely destroy the VMs
2. rebuild your environments from scratch, recording every step in detail
3. retry

[pborgen]

What's the OS on the scheduler, workers, and the client?
Is it 32 or 64 bit python?
How exactly did you install your stack?

Below are most of the steps I took. I think I am going to start using dask docker. And see what happens. Thanks so much for your help.

Details for the scheduler and work vms:
Python 3.8.2 (tags/v3.8.2:7b3ab59, Feb 25 2020, 23:03:10) [MSC v.1916 64 bit (AMD64)] on win32
Windows server 2012 r2

Some of the steps:
choco install python -y
choco install pip -y
choco install googlechrome -y
python -m pip install --upgrade pip
pip install wheel
python -m pip install "dask[complete]"
pip install pyarrow
pip install --upgrade numpy

[pborgen]

Just setup a ubuntu vm. I am running dask-docker on it for the scheduler and worker.. My dev machine is windows. I am getting the below warning. I am also getting that same error

Error:

distributed.worker - WARNING -  Compute Failed
Function:  getCompute
args:      (1018)
kwargs:    {}
Exception: ValueError('Inputs contain futures that were created by another client.')

Warning:

blosc
+-----------------------+---------+
|                       | version |
+-----------------------+---------+
| client                | None    |
| scheduler             | 1.9.1   |
| tcp://127.0.0.1:38323 | 1.9.1   |
+-----------------------+---------+

docker run -it --network host daskdev/dask dask-scheduler # start scheduler
docker run -it --network host daskdev/dask dask-worker localhost:8786 # start worker

[crusaderky]

Could you try again, but this time using anaconda on the client?
conda create -n dask python=3.8 distributed

[pborgen]

Gave this a try..The error is labeled at the bottom. Thanks again.

Was really hoping to be able to run everything on windows. Currently most our other products run windows and the operates team mostly uses windows. But that is not a hard requirment.

Steps:

1. I downloaded conda for windows and installed.
2. added condabin to my path
3. Run command 'conda create -n dask python=3.8 distributed'
4. I then got the errors outlined below under 'Other Steps' and then took the actions under 'Action' below.

Environment:
All Servers windows: One dev machine. Two Azure VM's all windows
My scheduler and workers are started via the cli
I run in pycharm with the 'Conda Environment' configured

Other steps:

Error:
Traceback (most recent call last):
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\__init__.py", line 2, in <module>
    from .core import (
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\core.py", line 9, in <module>
    import numpy as np
ModuleNotFoundError: No module named 'numpy'

Action:
conda install numpy

Error:
Traceback (most recent call last):
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\__init__.py", line 2, in <module>
    from .core import (
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\core.py", line 10, in <module>
    import pandas as pd
ModuleNotFoundError: No module named 'pandas'

Action:
conda install pandas


Traceback (most recent call last):
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\__init__.py", line 13, in <module>
    from .io import (
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\io\__init__.py", line 13, in <module>
    from .csv import read_csv, to_csv, read_table, read_fwf
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\dataframe\io\csv.py", line 21, in <module>
    from ...bytes import read_bytes, open_file, open_files
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\dask\bytes\__init__.py", line 9, in <module>
    raise ImportError(
ImportError: fsspec is required to use any file-system functionality. Please install using
conda install -c conda-forge 'fsspec>=0.3.3'
or
python -m pip install 'fsspec>=0.3.3'

Action: conda install -c conda-forge 'fsspec>=0.3.3'


Error:
ValueError: Mismatched versions found

numpy
+-----------------------+---------+
|                       | version |
+-----------------------+---------+
| client                | 1.18.1  |
| scheduler             | 1.18.4  |
| tcp://10.1.14.10:8002 | 1.18.4  |
| tcp://10.1.14.9:8002  | 1.18.4  |
+-----------------------+---------+

Action: 
conda install -c conda-forge 'numpy==1.18.4'

Error:

{'scheduler': {'host': {'python': '3.8.2.final.0', 'python-bits': 64, 'OS': 'Windows', 'OS-release': '10', 'machine': 'AMD64', 'processor': 'Intel64 Family 6 Model 79 Stepping 1, GenuineIntel', 'byteorder': 'little', 'LC_ALL': 'None', 'LANG': 'None'}, 'packages': {'python': '3.8.2.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.4', 'lz4': None, 'blosc': None}}, 'workers': {'tcp://10.1.14.10:8002': {'host': {'python': '3.8.2.final.0', 'python-bits': 64, 'OS': 'Windows', 'OS-release': '10', 'machine': 'AMD64', 'processor': 'Intel64 Family 6 Model 79 Stepping 1, GenuineIntel', 'byteorder': 'little', 'LC_ALL': 'None', 'LANG': 'None'}, 'packages': {'python': '3.8.2.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.4', 'lz4': None, 'blosc': None}}, 'tcp://10.1.14.9:8002': {'host': {'python': '3.8.2.final.0', 'python-bits': 64, 'OS': 'Windows', 'OS-release': '10', 'machine': 'AMD64', 'processor': 'Intel64 Family 6 Model 79 Stepping 1, GenuineIntel', 'byteorder': 'little', 'LC_ALL': 'None', 'LANG': 'None'}, 'packages': {'python': '3.8.2.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.4', 'lz4': None, 'blosc': None}}}, 'client': {'host': {'python': '3.8.2.final.0', 'python-bits': 64, 'OS': 'Windows', 'OS-release': '10', 'machine': 'AMD64', 'processor': 'Intel64 Family 6 Model 158 Stepping 9, GenuineIntel', 'byteorder': 'little', 'LC_ALL': 'None', 'LANG': 'None'}, 'packages': {'python': '3.8.2.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.4', 'lz4': None, 'blosc': None}}}
Traceback (most recent call last):
  File "D:/dev/code/netsense.support/datacrunch/test.py", line 33, in <module>
    main()
  File "D:/dev/code/netsense.support/datacrunch/test.py", line 28, in main
    results = client.gather(futures)
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\distributed\client.py", line 1961, in gather
    return self.sync(
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\distributed\client.py", line 815, in sync
    return sync(
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\distributed\utils.py", line 347, in sync
    raise exc.with_traceback(tb)
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\distributed\utils.py", line 331, in f
    result[0] = yield future
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\tornado\gen.py", line 735, in run
    value = future.result()
  File "E:\dev\Anaconda3\envs\dask\lib\site-packages\distributed\client.py", line 1826, in _gather
    raise exception.with_traceback(traceback)
  File "D:/dev/code/netsense.support/datacrunch/test.py", line 10, in getCompute
    return ddf[ddf.id == id].compute()
  File "c:\python38\lib\site-packages\dask\base.py", line 166, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "c:\python38\lib\site-packages\dask\base.py", line 444, in compute
    results = schedule(dsk, keys, **kwargs)
  File "c:\python38\lib\site-packages\distributed\client.py", line 2646, in get
    used as an optimization to avoid recomputation.
  File "c:\python38\lib\site-packages\distributed\client.py", line 2543, in _graph_to_futures
    
ValueError: Inputs contain futures that were created by another client.

[pborgen]

I also tested on my linux dask-docker server got the same error. I created a new conda environment to do this...

Side Note:
I am liking conda...I was able to blosc install with conda where pip would not let me the other day.

Extra steps after creating the conda environment.
conda install -c conda-forge 'python-blosc==1.9.1'
conda install -c conda-forge 'lz4==3.0.2'
conda install -c conda-forge 'numpy==1.18.1'
conda install dask

{'scheduler': {'host': {'python': '3.7.4.final.0', 'python-bits': 64, 'OS': 'Linux', 'OS-release': '4.15.0-1082-azure', 'machine': 'x86_64', 'processor': '', 'byteorder': 'little', 'LC_ALL': 'C.UTF-8', 'LANG': 'C.UTF-8'}, 'packages': {'python': '3.7.4.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.1', 'lz4': '3.0.2', 'blosc': '1.9.1'}}, 'workers': {'tcp://127.0.0.1:40043': {'host': {'python': '3.7.4.final.0', 'python-bits': 64, 'OS': 'Linux', 'OS-release': '4.15.0-1082-azure', 'machine': 'x86_64', 'processor': '', 'byteorder': 'little', 'LC_ALL': 'C.UTF-8', 'LANG': 'C.UTF-8'}, 'packages': {'python': '3.7.4.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.1', 'lz4': '3.0.2', 'blosc': '1.9.1'}}}, 'client': {'host': {'python': '3.7.4.final.0', 'python-bits': 64, 'OS': 'Windows', 'OS-release': '10', 'machine': 'AMD64', 'processor': 'Intel64 Family 6 Model 158 Stepping 9, GenuineIntel', 'byteorder': 'little', 'LC_ALL': 'None', 'LANG': 'None'}, 'packages': {'python': '3.7.4.final.0', 'dask': '2.16.0', 'distributed': '2.16.0', 'msgpack': '1.0.0', 'cloudpickle': '1.4.1', 'tornado': '6.0.4', 'toolz': '0.10.0', 'numpy': '1.18.1', 'lz4': '3.0.2', 'blosc': '1.9.1'}}}
Traceback (most recent call last):
  File "D:/dev/code/netsense.support/datacrunch/test.py", line 33, in <module>
    main()
  File "D:/dev/code/netsense.support/datacrunch/test.py", line 28, in main
    results = client.gather(futures)
  File "E:\dev\Anaconda3\envs\dask2\lib\site-packages\distributed\client.py", line 1967, in gather
    asynchronous=asynchronous,
  File "E:\dev\Anaconda3\envs\dask2\lib\site-packages\distributed\client.py", line 816, in sync
    self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
  File "E:\dev\Anaconda3\envs\dask2\lib\site-packages\distributed\utils.py", line 347, in sync
    raise exc.with_traceback(tb)
  File "E:\dev\Anaconda3\envs\dask2\lib\site-packages\distributed\utils.py", line 331, in f
    result[0] = yield future
  File "E:\dev\Anaconda3\envs\dask2\lib\site-packages\tornado\gen.py", line 735, in run
    value = future.result()
  File "E:\dev\Anaconda3\envs\dask2\lib\site-packages\distributed\client.py", line 1826, in _gather
    raise exception.with_traceback(traceback)
  File "D:/dev/code/netsense.support/datacrunch/test.py", line 10, in getCompute
    return ddf[ddf.id == id].compute()
  File "/opt/conda/lib/python3.7/site-packages/dask/base.py", line 166, in compute
  File "/opt/conda/lib/python3.7/site-packages/dask/base.py", line 444, in compute
  File "/opt/conda/lib/python3.7/site-packages/distributed/client.py", line 2655, in get
  File "/opt/conda/lib/python3.7/site-packages/distributed/client.py", line 2543, in _graph_to_futures
ValueError: Inputs contain futures that were created by another client.

[jacobtomlinson]

Just out of curiosity could you also install Dask from conda forge?