2

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

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

%%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)

1 ответ 1

3

Упакуйте всю логику обработки для одного значения 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)
2
  • Это было элегантно! Бесподобно! Как всегда на высоте!) Спасибо большущее! 6 июн 2021 в 17:46
  • @AlexandrKosolapov, всегда рад помочь) 6 июн 2021 в 17:52

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.