R Language
Параллельная обработка

замечания

Параллельная обработка с помощью пакета foreach

Пакет foreach обеспечивает параллельную обработку R. Но прежде чем вы захотите использовать многоядерные процессоры, вам необходимо назначить многоядерный кластер. Пакет doSNOW - одна из возможностей.

Простым использованием цикла foreach является вычисление суммы квадратного корня и квадрата всех чисел от 1 до 100000.

library(foreach)
library(doSNOW)

cl <- makeCluster(5, type = "SOCK")
registerDoSNOW(cl)

f <- foreach(i = 1:100000, .combine = c, .inorder = F) %dopar% {
    k <- i ** 2 + sqrt(i)
    k
} 

Структура вывода foreach контролируется аргументом .combine . Структура вывода по умолчанию - это список. В приведенном выше коде c вместо этого используется c . Обратите внимание, что функция вычисления (или оператор), такая как "+" также может использоваться для выполнения вычисления и возврата еще одного обработанного объекта.

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

Параллельная обработка с параллельным пакетом

parallel базовый пакет позволяет осуществлять параллельное вычисление с помощью разветвления, сокетов и генерации случайных чисел.

Определите количество ядер, присутствующих на локальном хосте:

parallel::detectCores(all.tests = FALSE, logical = TRUE)

Создайте кластер ядер на локальном хосте:

parallelCluster <- parallel::makeCluster(parallel::detectCores())

Во-первых, необходимо создать функцию, подходящую для распараллеливания. Рассмотрим набор данных mtcars . Регрессия на mpg может быть улучшена путем создания отдельной модели регрессии для каждого уровня cyl .

data <- mtcars
yfactor <- 'cyl'
zlevels <- sort(unique(data[[yfactor]]))
datay <- data[,1]
dataz <- data[,2]
datax <- data[,3:11]


fitmodel <- function(zlevel, datax, datay, dataz) {
  glm.fit(x = datax[dataz == zlevel,], y = datay[dataz == zlevel])
}

Создайте функцию, которая может проходить через все возможные итерации zlevels . Это все еще в серийном режиме, но является важным шагом, поскольку он определяет точный процесс, который будет распараллелен.

fitmodel <- function(zlevel, datax, datay, dataz) {
  glm.fit(x = datax[dataz == zlevel,], y = datay[dataz == zlevel])
}


for (zlevel in zlevels) {
  print("*****")
  print(zlevel)
  print(fitmodel(zlevel, datax, datay, dataz))
}

Выполните эту функцию:

worker <- function(zlevel) {
    fitmodel(zlevel,datax, datay, dataz)
  }

Параллельные вычисления с использованием parallel не могут получить доступ к глобальной среде. К счастью, каждая функция создает локальное окружение parallel может получить доступ. Создание функции обертки позволяет распараллеливать. Функция, которая должна применяться, также должна быть помещена в окружающую среду.

wrapper <- function(datax, datay, dataz) {
  # force evaluation of all paramters not supplied by parallelization apply
  force(datax)
  force(datay)
  force(dataz)
  # these variables are now in an enviroment accessible by parallel function
  
  # function to be applied also in the environment
  fitmodel <- function(zlevel, datax, datay, dataz) {
    glm.fit(x = datax[dataz == zlevel,], y = datay[dataz == zlevel])
  }
  
  # calling in this environment iterating over single parameter zlevel
  worker <- function(zlevel) {
    fitmodel(zlevel,datax, datay, dataz)
  }
  return(worker) 
}

Теперь создайте кластер и запустите функцию обертки.

parallelcluster <- parallel::makeCluster(parallel::detectCores())
models <- parallel::parLapply(parallelcluster,zlevels,
                              wrapper(datax, datay, dataz))

Всегда завершайте кластер, когда закончите.

parallel::stopCluster(parallelcluster)

parallel пакет включает в себя весь семейство apply() префиксом par .

Генерация случайных чисел

Основной проблемой с распараллеливанием является использование RNG в качестве семян. Случайные числа по числу повторяются числом операций либо с начала сеанса, либо с самого последнего set.seed() . Поскольку параллельные процессы возникают из одной и той же функции, он может использовать одно и то же семя, что может привести к одинаковым результатам! Вызовы будут выполняться в последовательном порядке на разных ядрах, не давая преимущества.

Набор семян должен быть сгенерирован и отправлен каждому параллельному процессу. Это делается автоматически в некоторых пакетах ( parallel , snow и т. Д.), Но они должны быть явно адресованы другим.

s <- seed
for (i in 1:numofcores) {
    s <- nextRNGStream(s)
    # send s to worker i as .Random.seed
}

Семена также могут быть установлены для воспроизводимости.

clusterSetRNGStream(cl = parallelcluster, iseed)

mcparallelDo

Пакет mcparallelDo позволяет асинхронно оценивать R-код на Unix-подобных (например, Linux и MacOSX) операционных системах. Основополагающая философия пакета согласовывается с потребностями анализа разведочных данных, а не с кодированием. Для кодирования асинхронности рассмотрим future пакет.

пример

Создание данных

data(ToothGrowth)

Триггер mcparallelДля выполнения анализа на развилке

mcparallelDo({glm(len ~ supp * dose, data=ToothGrowth)},"interactionPredictorModel")

Делайте другие вещи, например

binaryPredictorModel <- glm(len ~ supp, data=ToothGrowth)
gaussianPredictorModel <- glm(len ~ dose, data=ToothGrowth)

Результат mcparallelDo возвращается в вашу целевую среду, например .GlobalEnv, когда она заполнена сообщением (по умолчанию)

summary(interactionPredictorModel)

Другие примеры

# Example of not returning a value until we return to the top level
for (i in 1:10) {
  if (i == 1) {
    mcparallelDo({2+2}, targetValue = "output")
  }
  if (exists("output")) print(i)
}

# Example of getting a value without returning to the top level
for (i in 1:10) {
  if (i == 1) {
    mcparallelDo({2+2}, targetValue = "output")
  }
  mcparallelDoCheck()
  if (exists("output")) print(i)
}