Как создать кластеры внутри кластера при параллельной обработке Python?

Question

Я работаю над задачей, которая требует параллельного процесса на нескольких этапах.

Мой вопрос:

У меня четыре набора данных.Я написал функцию для обработки этих 4 наборов данных.Для параллельной обработки этих 4 наборов данных я создал 4 кластера, а затем отправил 4 набора данных в 4 кластера.Опять же, я хочу разделить каждый кластер на 4 кластера, поскольку я должен выполнить операцию group by.Я написал код для обработки вышеуказанной проблемы.Но это бросало мне ошибку.Вот мой пример кода.

def applyParallel(dfGrouped, func):
    retLst = Parallel(n_jobs=2, verbose=10)(delayed(func)(group) for name, group in dfGrouped)
    return pd.concat(retLst)

# This is the function to do group by parallel processing.
def new(x):
    tsc_out = applyParallel(x.groupby(gbcols), custum_func) 

def f1():
    from joblib import Parallel, delayed
    Parallel(n_jobs=4)(delayed(new)(i) for i in range(4)) #4 data sets

Эта функция работает правильно, если мы создаем кластеры только за один раз.Это не работает, если мы пытаемся разделить каждый кластер на несколько кластеров.В моем случае мне нужно создать всего 20 кластер.

Заранее спасибо.

Answer 1

Если я правильно понял ваш случай, вы можете сделать что-то подобное, используя пул работников charm4py (см. https://charm4py.readthedocs.io/en/latest/pool.html для получения дополнительной информации).Например:

from charm4py import charm

def applyParallel(dfGrouped, func):
    retLst = charm.pool.map(func, [group for name, group in dfGrouped], ncores=2)
    return pd.concat(retLst)

# This is the function to do group by parallel processing
def new(x):
    tsc_out = applyParallel(x.groupby(gbcols), custom_func)

def f1(args):
    charm.pool.map(new, [i for i in range(4)], ncores=4, allow_nested=True)
    exit()

charm.start(f1)

Используя пул работников, вы можете запускать задачи (которые, я думаю, совпадают с тем, что вы называете кластерами), определяя, сколько ядер вы хотите запустить для данного набора задач.Хотя я думаю, что в вашем случае вы могли бы просто оставить ncores=-1 и позволить charm4py планировать задачи, используя доступные ядра.

Это также будет работать, если у вас есть задачи, запускающие другие задачи (если вы используете allow_nested=True).

Следует отметить, что при запуске программы с помощью charm4py вы указываете, сколько процессов нужно запускать (вы даже можете запускать процессы на нескольких хостах, и приложение может использовать их все одновременно).Так, например, если у вас 8 ядер и вы хотите использовать все из них, запустите программу с 8 процессами.