compute () в dask не работает

Я пробую простое параллельное вычисление в Dask. Это мой код.

  import time
  import dask as dask
  import dask.distributed as distributed
  import dask.dataframe as dd
  import dask.delayed as delayed
  from dask.distributed import Client,progress

  client = Client('localhost:8786')
  df = dd.read_csv('file.csv')
  ddf = df.groupby(['col1'])[['col2']].sum() 
  ddf = ddf.compute()
  print ddf

Кажется, что это нормально из документации, но при запуске я получаю следующее:

    Traceback (most recent call last):
    File "dask_prg1.py", line 17, in <module>
    ddf = ddf.compute()
    File "/usr/local/lib/python2.7/site-packages/dask/base.py", line 156, in compute
   (result,) = compute(self, traverse=False, **kwargs)
    File "/usr/local/lib/python2.7/site-packages/dask/base.py", line 402, in compute
   results = schedule(dsk, keys, **kwargs)
   File "/usr/local/lib/python2.7/site-packages/distributed/client.py", line 2159, in get
direct=direct)
  File "/usr/local/lib/python2.7/site-packages/distributed/client.py", line 1562, in gather
asynchronous=asynchronous)
 File "/usr/local/lib/python2.7/site-packages/distributed/client.py", line 652, in sync
return sync(self.loop, func, *args, **kwargs)
 File "/usr/local/lib/python2.7/site-packages/distributed/utils.py", line 275, in sync
six.reraise(*error[0])
 File "/usr/local/lib/python2.7/site-packages/distributed/utils.py", line 260, in f
result[0] = yield make_coro()
   File "/usr/local/lib/python2.7/site-packages/tornado/gen.py", line 1099, in run
value = future.result()
 File "/usr/local/lib/python2.7/site-packages/tornado/concurrent.py", line 260, in result
raise_exc_info(self._exc_info)
 File "/usr/local/lib/python2.7/site-packages/tornado/gen.py", line 1107, in run
yielded = self.gen.throw(*exc_info)
 File "/usr/local/lib/python2.7/site-packages/distributed/client.py", line 1439, in _gather
traceback)
File "/usr/local/lib/python2.7/site-packages/dask/bytes/core.py", line 122, in read_block_from_file
with lazy_file as f:
File "/usr/local/lib/python2.7/site-packages/dask/bytes/core.py", line 166, in __enter__
f = SeekableFile(self.fs.open(self.path, mode=mode))
 File "/usr/local/lib/python2.7/site-packages/dask/bytes/local.py", line 58, in open
return open(self._normalize_path(path), mode=mode)
 IOError: [Errno 2] No such file or directory: 'file.csv'

Я не понимаю, что не так. Пожалуйста, помогите мне с этим. Заранее спасибо.


dataframe dask dask-distributed
person Sweta    schedule 19.07.2018    source источник

Ответы (1)


arrow_upward
2
arrow_downward

Вы можете передать абсолютный путь к файлу read_csv. Причина в том, что вы поручаете работу по открытию и чтению файла работнику dask, и, возможно, вы не запустили эту работу с тем же рабочим каталогом, что и ваш скрипт / сеанс.

person mdurant    schedule 19.07.2018
comment
Проблема не в этом. Пробовал. Я также попытался опустить оператор compute (), а затем запустить его. Он работает нормально. Итак, я думаю, что проблема в выражении compute (). - person Sweta; 19.07.2018
comment
Находятся ли ваши сотрудники на одном компьютере и есть ли у них разрешение на просмотр одних и тех же файлов? - person mdurant; 19.07.2018
comment
Спасибо. Вот в чем проблема. Рабочие находятся на разных машинах, и, вероятно, один из них не может получить доступ к файлу csv. - person Sweta; 19.07.2018