Как распараллелить цикл для генерации новых данных на основе исходных?

Question

Можно ли распараллелить цикл генерации признаков так, чтобы на один проход цикла приходилось одно ядро/поток?

По времени каждый поток может отличаться, но задача параллельно посчитать и всё красиво в один фрейм данных соединить.

%%time

from random import choices
stats = pd.DataFrame({'object':choices(range(1, 4), k=200000),
                  'data1':choices(range(1, 100), k=200000),
                  'data2':choices(range(500, 2000), k=200000)})

data_cols = ['data1','data2']
obj = 'object'

for ro in [4,5,6,7,8,9,10,11]:

    temp = stats[[obj] + data_cols]
    for s in data_cols:
    
        st = temp[[obj,s]].groupby(obj)[s].rolling(ro,min_period=0).agg(['std','mean','median']).reset_index()
        st[['std','mean','median']] = st.groupby([obj])['std','mean','median'].shift(+1)
        st = st.sort_values('level_1').set_index('level_1')
        temp[['{}_{}_{}'.format(ro,method,s) for method in ['std','mean','median']]] = st[['std','mean','median']]

cola = [c for c in temp.columns if c not in stats.columns]
stats = stats.join(temp[cola])

Для параллельного расчёта сейчас запускаю ноутбуки под каждый ro -> вычисляю и сохраняю в pickle, а в основном ноутбуке жду/ищу сохранённый pickle каждого ro и соединяю:

import os
for file in ['{}_stats{}'.format(x,w) for w in range(4,12)]:
    while os.path.isfile(file) == False:
        time.sleep(5)
        print('WHERE is MY FILES o_O',x,file)
    else:
        time.sleep(5)
        temp_stats = pd.read_pickle(file)
        cola = [c for c in temp_stats.columns if c not in stats.columns]
        stats = stats.join(temp_stats[cola]) 
for file in ['{}_stats{}'.format(x,w) for w in range(4,12)]:
    os.remove(file)

Answer 1

Упакуйте всю логику обработки для одного значения ro в функцию для удобства и воспользуйтесь joblib.parallel:

from joblib import Parallel, delayed

def f(stats, ro, data_cols=data_cols, obj='object'):
    temp = stats[[obj] + data_cols]
    for s in data_cols:
        st = temp[[obj,s]].groupby(obj)[s].rolling(ro,min_period=0).agg(['std','mean','median']).reset_index()
        st[['std','mean','median']] = st.groupby([obj])['std','mean','median'].shift(+1)
        st = st.sort_values('level_1').set_index('level_1')
        temp[['{}_{}_{}'.format(ro,method,s) for method in ['std','mean','median']]] = st[['std','mean','median']]
    return temp.loc[:, temp.columns.difference(stats.columns)]

ro_vals = [4,5,6,7,8,9,10]
n_jobs = 5


# распараллеливание
res = pd.concat(
    Parallel(n_jobs=n_jobs, require='sharedmem')(delayed(f)(stats, ro=ro, data_cols=data_cols, obj='object') for ro in ro_vals),
    axis=1)
# NOTE: (chunks):-->^^^^^^--------------- (function): -->^
res = stats.join(res)

Сравнение скорости выполнения:

In [76]: %%timeit
    ...: for ro in [4,5,6,7,8,9,10,11]:
    ...:     temp = stats[[obj] + data_cols]
    ...:     for s in data_cols:
    ...:         st = temp[[obj,s]].groupby(obj)[s].rolling(ro,min_period=0).agg(['std','mean','median']).reset_index()
    ...:         st[['std','mean','median']] = st.groupby([obj])['std','mean','median'].shift(+1)
    ...:         st = st.sort_values('level_1').set_index('level_1')
    ...:         temp[['{}_{}_{}'.format(ro,method,s) for method in ['std','mean','median']]] = st[['std','mean','median']]
    ...: cola = [c for c in temp.columns if c not in stats.columns]
    ...: res = stats.join(temp[cola])
4.56 s ± 271 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [77]: %%timeit
    ...: res = pd.concat(
    ...:     Parallel(n_jobs=5, require='sharedmem')(delayed(f)(stats, ro=ro, data_cols=data_cols, obj='object') for ro in ro_vals),
    ...:     axis=1)
    ...: res = stats.join(res)
1.75 s ± 47.7 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

БЛОГ НА HUSL

Как распараллелить цикл для генерации новых данных на основе исходных?

Ответы (1 шт):