Équilibrage de charge pour le traitement parallèle

Question

Je cours une fonction similaire au calcul de l'écart type... mais qui prend beaucoup plus de temps à s'exécuter.

Je souhaite que la fonction soit utilisée pour calculer la valeur cumulée de l'écart type, c'est-à-dire pour les jours 1 à n la fonction de type écart type pour cela.

Cependant, en raison de la longue période de calcul nécessaire, je voulais exécuter cela sur un cluster.

Donc, je voulais diviser les données pour que chaque nœud du cluster termine à peu près en même temps. par exemple si ma fonction était la suivante, la méthode sur une seule machine fonctionnerait de la manière suivante:

vec <- xts(rnorm(1000),Sys.Date()-(1:1000)
lapply(1:length(vec), function(x){
    Sys.sleep(30)
    sd(as.numeric(vec[1:x]))
}

(N.B Le sys.sleep est ajouté là pour représenter le temps supplémentaire nécessaire pour traiter ma fonction personnalisée)

cependant, disons que je voulais diviser cela sur deux machines et au lieu de 1, comment pourrais-je diviser le vecteur 1:length(vec) de sorte que je puisse donner à chaque machine une liste de c(1:y) à la machine 1 et c((y+1):length(vec)) à la machine 2, pour que les deux machines terminent à temps. c'est-à-dire quelle serait la valeur de y pour que les deux processus se terminent à peu près en même temps... et que se passerait-il si nous le faisions sur 10 machines... comment pourrait-on trouver les coupures dans le vecteur original c(1:length(vec)) pour que cela fonctionne...

c'est-à-dire. J'aurais

y <- 750 # C'est juste une estimation de l'endroit potentiellement où cela pourrait être.
vec <- xts(rnorm(1000),Sys.Date()-(1:1000)
# sur la machine 1 j'aurais
lapply(1:y, function(x){
    Sys.sleep(30)
    sd(as.numeric(vec[1:x]))
}

# et sur la machine 2 j'aurais

lapply(y+1:length(vec), function(x){
    Sys.sleep(30)
    sd(as.numeric(vec[1:x]))
}

Answer 1

Le package parallel fait désormais partie de base R et peut aider à exécuter R sur des clusters de taille modérée, y compris sur Amazon EC2. La fonction parLapplyLB répartira le travail d'un vecteur d'entrée sur les nœuds de travail d'un cluster.

Une chose à savoir est que makePSOCKcluster est (actuellement à partir de R 2.15.2) limité à 128 travailleurs par la constante NCONNECTIONS dans connections.c.

Voici un exemple rapide d'une session utilisant le package parallel que vous pouvez essayer sur votre propre machine :

library(parallel)
help(package=parallel)

## create the cluster passing an IP address for the head node
## hostname -i works on Linux, but not on BSD
## descendants (like OS X)
# cl <- makePSOCKcluster(hosts, master=system("hostname -i", intern=TRUE))

## for testing, start a cluster on your local machine
cl <- makePSOCKcluster(rep("localhost", 3))

## do something once on each worker
ans <- clusterEvalQ(cl, { mean(rnorm(1000)) })

## push data to the workers
myBigData <- rnorm(10000)
moreData <- c("foo", "bar", "blabber")
clusterExport(cl, c('myBigData', 'moreData'))

## test a time consuming job
## (~30 seconds on a 4 core machine)
system.time(ans <- parLapplyLB(cl, 1:100, function(i) {
  ## summarize a bunch of random sample means
  summary(
    sapply(1:runif(1, 100, 2000),
           function(j) { mean(rnorm(10000)) }))
}))

## shut down worker processes
stopCluster(cl)

Le groupe Bioconductor a mis en place un moyen vraiment facile de commencer : Utilisation d'un cluster parallèle dans le cloud

Pour en savoir plus sur l'utilisation du package parallel sur EC2, consultez : R in the Cloud et pour R sur des clusters en général, consultez : CRAN Task View: High-Performance and Parallel Computing with R.

Enfin, une autre option bien établie externe à R est Starcluster.

Answer 2

Regardez le paquet snow -- en particulier la fonction clusterApplyLB pour gérer une fonction d'application équilibrée de charge.

Cela va en fait gérer la distribution du travail vers les nœuds/cores de manière plus intelligente qu'une simple partition égale.

Answer 3

Envisagez d'utiliser Hadoop (alias MapReduce) via RHIPE.