R Language
Parallelverarbeitung

Bemerkungen

Parallelverarbeitung mit foreach-Paket

Mit dem foreach Paket wird die Leistung der parallelen Verarbeitung erweitert. Bevor Sie jedoch Multi-Core-CPUs verwenden möchten, müssen Sie einen Multi-Core-Cluster zuweisen. Das doSNOW Paket ist eine Möglichkeit.

Eine einfache Anwendung der foreach-Schleife besteht darin, die Summe der Quadratwurzel und des Quadrats aller Zahlen von 1 bis 100000 zu berechnen.

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
} 

Die Struktur der Ausgabe von foreach wird durch das .combine Argument gesteuert. Die Standardausgabestruktur ist eine Liste. Im obigen Code wird stattdessen c verwendet, um einen Vektor zurückzugeben. Es ist zu beachten, dass eine Berechnungsfunktion (oder ein Operator) wie "+" auch verwendet werden kann, um eine Berechnung durchzuführen und ein weiterverarbeitetes Objekt zurückzugeben.

Es ist wichtig zu erwähnen, dass das Ergebnis jeder foreach-Schleife der letzte Aufruf ist. In diesem Beispiel wird also k zum Ergebnis addiert.

Parallelverarbeitung mit Parallelpaket

Das Basispaket parallel ermöglicht die parallele Berechnung durch Gabelung, Sockets und Erzeugung von Zufallszahlen.

Ermitteln Sie die Anzahl der auf dem localhost vorhandenen Kerne:

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

Erstellen Sie einen Cluster der Kerne auf dem localhost:

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

Zunächst muss eine für die Parallelisierung geeignete Funktion erstellt werden. Betrachten Sie das mtcars Dataset. Eine Regression bei mpg könnte durch Erstellen eines separaten Regressionsmodells für jede cyl Ebene verbessert werden.

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])
}

Erstellen Sie eine Funktion, die alle möglichen Iterationen von zlevels . Dies ist immer noch in Serie, aber es ist ein wichtiger Schritt, da es den genauen Prozess bestimmt, der parallelisiert wird.

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

Curry diese Funktion:

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

Paralleles Computing mit parallel kann nicht auf die globale Umgebung zugreifen. Glücklicherweise erstellt jede Funktion eine lokale Umgebung, auf die parallel kann. Die Erstellung einer Wrapper-Funktion ermöglicht die Parallelisierung. Die anzuwendende Funktion muss auch innerhalb der Umgebung platziert werden.

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

Erstellen Sie nun einen Cluster und führen Sie die Wrapper-Funktion aus.

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

Stoppen Sie den Cluster immer, wenn Sie fertig sind.

parallel::stopCluster(parallelcluster)

Das parallel - Paket enthält die gesamte apply() Familie, mit dem Präfix par .

Zufallszahlengenerierung

Ein Hauptproblem bei der Parallelisierung ist die Verwendung von RNG als Saatgut. Zufallszahlen nach Anzahl werden durch die Anzahl der Operationen vom Beginn der Sitzung oder vom letzten set.seed() . Da parallele Prozesse aus derselben Funktion stammen, kann derselbe Samen verwendet werden, was möglicherweise zu identischen Ergebnissen führt! Anrufe werden seriell auf den verschiedenen Kernen ausgeführt, bieten jedoch keinen Vorteil.

Ein Satz von Startwerten muss generiert und an jeden parallelen Prozess gesendet werden. Dies geschieht in einigen Paketen automatisch ( parallel , snow usw.), muss aber in anderen explizit angesprochen werden.

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

Samen können auch auf Reproduzierbarkeit eingestellt werden.

clusterSetRNGStream(cl = parallelcluster, iseed)

mcparallelDo

Das Paket mcparallelDo ermöglicht die asynchrone Bewertung von R-Code unter Unix-ähnlichen Betriebssystemen (z. B. Linux und MacOSX). Die zugrunde liegende Philosophie des Pakets orientiert sich an den Erfordernissen der explorativen Datenanalyse und nicht an der Codierung. Berücksichtigen Sie für die Codierung der Asynchronität das future Paket.

Beispiel

Daten erstellen

data(ToothGrowth)

Löst mcparallelDo aus, um eine Analyse an einer Gabel durchzuführen

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

Andere Dinge tun, z

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

Das Ergebnis von mcparallelDo wird in Ihrer targetEnvironment zurückgegeben, z. B. .GlobalEnv, wenn eine Nachricht vollständig ist (standardmäßig)

summary(interactionPredictorModel)

Andere Beispiele

# 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)
}