Использование Dask LocalCluster () в модульной кодовой базе python - PullRequest
Новая
       65

Использование Dask LocalCluster () в модульной кодовой базе python

0 голосов
/ 09 апреля 2020

Я пытаюсь использовать Dask Distributed LocalCluster для параллельного запуска кода с использованием всех ядер одной машины.

Рассмотрим пример конвейера данных python со структурой папок ниже. 

sample_dask_program
├── main.py
├── parallel_process_1.py
├── parallel_process_2.py
├── process_1.py
├── process_2.py
└── process_3.py

main.py - это точка входа, которая выполняется при конвейерной передаче последовательно.

Например: 

def run_pipeline():
    stage_one_run_util()
    stage_two_run_util()

    ...

    stage_six_run_util()


if __name__ == '__main__':

    ...

    run_pipeline()

parallel_process_1.py и parallel_process_2.py - это модули, которые создают Client () и используют futures для достижения параллелизма. 

with Client() as client:
            # list to store futures after they are submitted
            futures = []

            for item in items:
                future = client.submit(
                    ...
                )
                futures.append(future)

            results = client.gather(futures)

process_1.py , process_2.py и process_3.py - это модули, которые выполняют простые вычисления, которые не должны выполняться параллельно с использованием всех ядер ЦП.

Traceback: 

  File "/sm/src/calculation/parallel.py", line 140, in convert_qty_to_float
    results = client.gather(futures)
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/client.py", line 1894, in gather
    asynchronous=asynchronous,
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/client.py", line 778, in sync
    self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/utils.py", line 348, in sync
    raise exc.with_traceback(tb)
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/utils.py", line 332, in f
    result[0] = yield future
  File "/home/iouser/.local/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
    value = future.result()
concurrent.futures._base.CancelledError

Это ошибка, выданная рабочими: 

distributed.worker - ERROR - failed during get data with tcp://127.0.0.1:33901 -> tcp://127.0.0.1:38821
Traceback (most recent call last):
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/comm/tcp.py", line 248, in write
    future = stream.write(frame)
  File "/home/iouser/.local/lib/python3.7/site-packages/tornado/iostream.py", line 546, in write
    self._check_closed()
  File "/home/iouser/.local/lib/python3.7/site-packages/tornado/iostream.py", line 1035, in _check_closed
    raise StreamClosedError(real_error=self.error)
tornado.iostream.StreamClosedError: Stream is closed
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/worker.py", line 1248, in get_data
    compressed = await comm.write(msg, serializers=serializers)
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/comm/tcp.py", line 255, in write
    convert_stream_closed_error(self, e)
  File "/home/iouser/.local/lib/python3.7/site-packages/distributed/comm/tcp.py", line 121, in convert_stream_closed_error
    raise CommClosedError("in %s: %s: %s" % (obj, exc.__class__.__name__, exc))
distributed.comm.core.CommClosedError: in <closed TCP>: BrokenPipeError: [Errno 32] Broken pipe

Я не могу локально воспроизвести эту ошибку или найти минимальный воспроизводимый пример, так как возникновение эта ошибка внезапная.

Это правильный способ использования Dask LocalClus тер в модульной python программе?

EDIT

Я заметил, что эти ошибки возникают, когда LocalCluster создается с относительно большим числом потоков и процессов. Я делаю вычисления, которые используют NumPy и Pandas, и это не очень хорошая практика, как описано здесь .

Иногда, когда LocalCluster создается с использованием 4 рабочих и 16 процессов , не выдается ошибка. Когда LocalCluster создается с использованием 8 рабочих и 40 процессов, выдается описанная выше ошибка.

Насколько я понимаю, dask случайным образом выбирает эту комбинацию (это проблема с dask?), Как я тестировал в том же экземпляре AWS Batch (с 8 ядрами (16 vCPU)).

Проблема не возникает, когда я принудительно создаю кластер только с потоками.

Например: 

cluster = LocalCluster(processes=False)
with Client(cluster) as client:
    client.submit(...)
    ...

Но создание LocalCluster с использованием только потоков замедляет выполнение в 2-3 раза.

Таким образом, решение проблемы заключается в поиске подходящего числа процессов / потоков, подходящих для программа?

1 Ответ

0 голосов
/ 09 апреля 2020

Чаще всего создается Dask Client один раз, а затем запускается много рабочих нагрузок на нем. 

with Client() as client:
    stage_one(client)
    stage_two(client)

При этом все, что вы делаете, должно быть в порядке. Если вы можете воспроизвести ошибку на минимальном примере, это было бы полезно (но без ожиданий).

...