В чем разница между dask.bag / dask.delayed для l oop, выберите лучший способ для python параллельных заданий в dask

Question

Env info:

Python==3.7
Win7 i5-2310:2-cores-4-threads 4GB-memory

Dask Ini:

from dask.distributed import Client

Client()

# Client Scheduler: tcp://127.0.0.1:50723 Dashboard:
# http://127.0.0.1:8787/status Cluster Workers: 4 Cores: 4 Memory: 4.20 GB

Первый: отложенная функция с For-l oop:

@dask.delayed
def teststr(x,y):
    return x+y


rt=[]
for i in range(1000):
    rt.append(teststr(str(i),str(i+1)) )
rt = dask.compute(*rt)

It показывает Wall time: 3.23 s

Второй: Сумка на основе списка с функцией карты:

import dask.bag as db

b=db.from_sequence([(str(i),str(i+1)) for i in range(1000)])

def teststr2(x,y):
    return x+y

%%time
rt2=b.map(lambda x:teststr2(x[0],x[1])).compute()

Показывает Wall time: 899 ms Forl oop один занимает гораздо больше времени, чем сумка, и это стоит дискового ввода-вывода много (возможно, virtual-mem).

Я не проверял, что произойдет, если я изменю конфигурацию dask.client (поток против процесса), различные задачи (функция Integer против функции строки против функции без GIL против ввода-вывода базы данных против дискового ввода-вывода) или объекта большего размера (работа с большим DataFrame или Numpy при выполнении некоторого задания анализа данных.).

Как выбрать лучший способ для python параллельных заданий в Даск. Хотя я получаю тот же вывод, иногда это занимает слишком много времени на вычисления. Должны быть некоторые правила, помогающие мне.

Я надеюсь, что это могут быть шпаргалки, которые сообщают мне лучший способ организовать мой код на основе System-Env, виды задач или что-то еще.

Answer 1

Одним из основных различий между этими двумя вариантами является количество задач. Вы можете сделать len(thing.dask), чтобы быстро взглянуть на график, необходимый для вычисления заданного объекта dask, отложенного или мешкового.

>>> rt2=b.map(lambda x:teststr2(x[0],x[1]))
>>> len(rt2.dask)
200

>>> rt=[]
>>> for i in range(1000):
...     rt.append(teststr(str(i),str(i+1)) )
>>> sum(len(t.dask) for t in rt)
1000

Таким образом, количество заданий в пакете в пять раз меньше, и это это потому, что bag делает свои вычисления в пакетном режиме на входе (также называемом разделами). Поскольку запрашиваемая вами вычислительная функция происходит очень быстро, общее время полностью определяется накладными расходами для каждой задачи и временем, затрачиваемым на отправку графика задач в планировщик.