Как правильно использовать client.scatter и когда в Dask - PullRequest
Новая
       29

Как правильно использовать client.scatter и когда в Dask

1 голос
/ 20 марта 2019

При выполнении «большого» количества задач я получаю эту ошибку:

Рассмотрите возможность рассеивания больших объектов заранее с помощью client.scatter to уменьшить нагрузку на планировщик и сохранить данные о работниках

И я также получаю кучу сообщений, подобных этим: 

tornado.application - ERROR - Exception in callback <bound method BokehTornado._keep_alive of <bokeh.server.tornado.BokehTornado object at 0x7f20d25e10b8>>
Traceback (most recent call last):
  File "/home/muammar/.local/lib/python3.7/site-packages/tornado/ioloop.py", line 907, in _run
    return self.callback()
  File "/home/muammar/.local/lib/python3.7/site-packages/bokeh/server/tornado.py", line 542, in _keep_alive
    c.send_ping()
  File "/home/muammar/.local/lib/python3.7/site-packages/bokeh/server/connection.py", line 80, in send_ping
    self._socket.ping(codecs.encode(str(self._ping_count), "utf-8"))
  File "/home/muammar/.local/lib/python3.7/site-packages/tornado/websocket.py", line 447, in ping
    raise WebSocketClosedError()
tornado.websocket.WebSocketClosedError
tornado.application - ERROR - Exception in callback <bound method BokehTornado._keep_alive of <bokeh.server.tornado.BokehTornado object at 0x7f20d25e10b8>>
Traceback (most recent call last):
  File "/home/muammar/.local/lib/python3.7/site-packages/tornado/ioloop.py", line 907, in _run
    return self.callback()
  File "/home/muammar/.local/lib/python3.7/site-packages/bokeh/server/tornado.py", line 542, in _keep_alive
    c.send_ping()
  File "/home/muammar/.local/lib/python3.7/site-packages/bokeh/server/connection.py", line 80, in send_ping
    self._socket.ping(codecs.encode(str(self._ping_count), "utf-8"))
  File "/home/muammar/.local/lib/python3.7/site-packages/tornado/websocket.py", line 447, in ping
    raise WebSocketClosedError()
tornado.websocket.WebSocketClosedError
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:52950 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:52964 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:52970 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:52984 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:52986 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53002 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53016 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53018 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53038 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53042 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53048 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53060 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53068 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53072 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53146 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53156 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53170 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53178 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53186 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53188 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53192 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53194 remote=tcp://127.0.0.1:37945>
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://127.0.0.1:53196 remote=tcp://127.0.0.1:37945>

Эти задачи выполняются внутри ClassCreatingTheIssue, где у меня нет доступа (я думаю) к client. Просто у вас есть идея, я вставляю ниже сценарий, который вызывает эти вещи: 

from dask.distributed import Client, LocalCluster
import sys
sys.path.append('../../')
from mypackage import SomeClass
from mypackage.module2 import SomeClass2
from mypackage.module3 import ClassCreatingTheIssue


def train():

    calc = SomeClass(something=SomeClass2(**stuff),
                     something2=ClassCreatingTheIssue())

    calc.train(training_set=images)


if __name__ == '__main__':
    cluster = LocalCluster(n_workers=8, threads_per_worker=2)
    client = Client(cluster, asyncronous=True)
    train()

Мне удалось сузить, какая функция вызвала эту ошибку, и она выглядит следующим образом: 

def get_lt(self, index):
    """Return LT vectors

    Parameters
    ----------
    index : int
        Index of image.

    Returns
    -------
    _LT : list
        Returns a list that maps atomic fingerprints in the images.
      """
    _LT = []

    for i, group in enumerate(self.fingerprint_map):                                                                                                                                                         
        if i == index:
            for _ in group:
                _LT.append(1.)
        else:
            for _ in group:
                _LT.append(0.)
    return _LT

Эта отложенная функция в основном возвращает очень большой список. Какой способ использовать client.scatter в этом контексте? Буду очень признателен за любую помощь!

Примечание: иногда в этот момент все приложение не работает и все терпит неудачу. Я подтвердлю позже, потому что сейчас я запускаю еще один тест.

1 Ответ

1 голос
/ 04 апреля 2019

Какую версию Dask Distributed вы используете?Я пользуюсь 1.26, в котором есть предупреждающее сообщение: 

/Users/scott/anaconda3/lib/python3.6/site-packages/distributed/worker.py:2791: UserWarning: Large object of size 8.00 MB detected in task graph: 
  (array([[ 0.02152672,  0.09287627, -0.32135721, .. ... 1.25601994]]),)
Consider scattering large objects ahead of time
with client.scatter to reduce scheduler burden and 
keep data on workers

    future = client.submit(func, big_data)    # bad

    big_future = client.scatter(big_data)     # good
    future = client.submit(func, big_future)  # good
  % (format_bytes(len(b)), s))

Это предупреждающее сообщение уже давно (хотя никаких жестких цифр; инструмент обвинения GitHub здесь не очень полезен).

Вот фрагмент кода, чтобы проиллюстрировать это: 

import numpy as np
from distributed import Client
client = Client()

def f(x):
    return x.sum()

N = 1_000
x = np.random.randn(N, N)

r1 = client.submit(f, x).result()

x_scattered = client.scatter(x)
r2 = client.submit(f, x_scattered).result()

assert r1 == r2
...