Параллельное программирование в pyRserve с использованием сельдерея

Question

У меня есть кадр данных, состоящий из Time Series.

Индекс даты | Временная серия 1 | Временная серия 2 | ... и так далее

Я использовал pyRserve запустить функцию прогнозирования с помощью R.

Я хочу осуществить параллельную обработку с использованием сельдерея. Я написал код рабочего в следующем контексте.

def pipeR(k #input variable): 
    conn = pyRserve.connect(host = 'localhost', port = 6311) 
    # OPENING THE CONNECTION TO R 

    conn.r.i = k 
    # ASSIGNING THE PYTHON VARIABLE TO THAT OF IN THE R ENVIRONMENT 

    conn.voideval\(''' 
    WKR_Func <- forecst(a) 
    { 
    ...# FORECASTS THE TIMESERIES IN COLUMN a OF THE DATAFRAME 
    } 
    ''') 

    conn.eval('forecst(i)') 
    # CALLING THE FUNCTION IN R 

group(pipeR.s(k) for k in [...list of column headers...])() 

Для реализации параллельной обработки, можно иметь один порт для всех рабочих процессов (как я сделал в приведенном выше коде, порт: 6311) или я должен иметь различные порты для различных рабочих процессов ??

В настоящее время я получаю сообщение об ошибке

Ошибка в SocketConnection ("локальный", порт = порт, сервер = TRUE, блокирующие = TRUE,: не может открыть соединение

в R .

Answer 1

проблема решена, когда получил я открыл различные порты для каждого рабочего процесса ...

def pipeR(k, Frequency, Horizon, Split, wd_path): 
    # GENERATING A RANDOM PORT 
    port = randint(1000,9999) 

    # OPENING THE PORT IN THE R ENVIRONMENT 
    conn0 = pyRserve.connect(host = 'localhost', port = 6311) 
    conn0.r.port = port 
    conn0.voidEval\ 
    (''' 
     library(Rserve) 
     Rserve(port = port, args = '--no-save') 
    ''') 

    # OPENING THE PORT IN THE PYTHON ENVIRONMENT 
    conn = pyRserve.connect(host = 'localhost', port = port) 

    # ASSIGNING THE PYTHON VARIABLE TO THAT OF IN THE R ENVIRONMENT 
    conn.r.i = k 

    conn.voideval\ 
    (''' 
    WKR_Func <- forecst(a) 
    { 
    ...# FORECASTS THE TIMESERIES IN COLUMN a OF THE DATAFRAME 
    } 
    ''') 

    conn.eval/('forecst(i)') 
    conn0.close()