escribiendo en un entorno global cuando se ejecuta en paralelo

Question

Tengo un data.frame de celdas, valores y coordenadas. Reside en el ambiente global.

> head(cont.values) 
    cell value x y 
1 11117 NA -34 322 
2 11118 NA -30 322 
3 11119 NA -26 322 
4 11120 NA -22 322 
5 11121 NA -18 322 
6 11122 NA -14 322 

Debido a que mi función personalizada lleva casi un segundo para calcular célula individual (y tengo decenas de miles de células para calcular) No quiero duplicar los cálculos para las células que ya tienen un valor. Mi siguiente solución intenta evitar eso. Cada celda se puede calcular de forma independiente, gritando para la ejecución en paralelo.

Lo que mi función realmente hace es verificar si hay un valor para un número de celda específico y si es NA, lo calcula y lo inserta en lugar de NA.

puedo correr mi función mágica (resultado es value para una correspondiente cell) usando aplica familia de funciones y desde dentro apply, puedo leer y escribir cont.values sin ningún problema (que es en el entorno global).

Ahora, quiero ejecutar esto en paralelo (usando snowfall) y no puedo leer o escribir desde/a esta variable desde el núcleo individual.

Pregunta: ¿Qué solución sería capaz de leer/escribir desde/hacia una variable dinámica que reside en el entorno global desde el trabajador (núcleo) al ejecutar una función en paralelo. ¿Hay un mejor enfoque para hacer esto?

Answer 1

el patrón de un almacén central que los trabajadores consultan los valores se implementa en el paquete rredis en CRAN. La idea es que el servidor Redis mantenga un almacén de pares clave-valor (su marco de datos global, reimplementado).Los trabajadores consultan al servidor para ver si se ha calculado el valor (redisGet) y, si no, lo calculan y lo almacenan (redisSet) para que otros trabajadores puedan volver a utilizarlo. Los trabajadores pueden ser guiones R, por lo que es fácil ampliar la fuerza de trabajo. Es un paradigma paralelo alternativo muy bueno. Aquí hay un ejemplo que usa la noción de 'memorizar' cada resultado. Tenemos una función que es lento (para un segundo)

fun <- function(x) { Sys.sleep(1); x } 

Escribimos un 'memoizer' que devuelve una variante de fun que comprueba primero para ver si ya se ha calculado el valor de x, y si es así usos que

memoize <- 
    function(FUN) 
{ 
    force(FUN) # circumvent lazy evaluation 
    require(rredis) 
    redisConnect() 
    function(x) 
    { 
     key <- as.character(x) 
     val <- redisGet(key) 
     if (is.null(val)) { 
      val <- FUN(x) 
      redisSet(key, val) 
     } 
     val 
    } 
} 

A continuación, memoize nuestra función

funmem <- memoize(fun) 

e ir

> system.time(res <- funmem(10)); res 
    user system elapsed 
    0.003 0.000 1.082 
[1] 10 
> system.time(res <- funmem(10)); res 
    user system elapsed 
    0.001 0.001 0.040 
[1] 10 

Esto requiere un servidor redis ejecutándose fuera de R pero muy fácil de instalar; vea la documentación que viene con el paquete rredis.

A dentro-R versión paralela podría ser

library(snow) 
cl <- makeCluster(c("localhost","localhost"), type = "SOCK") 
clusterEvalQ(cl, { require(rredis); redisConnect() }) 
tasks <- sample(1:5, 100, TRUE) 
system.time(res <- parSapply(cl, tasks, funmem)) 

Answer 2

Depende de cuál sea la función en cuestión, por supuesto, pero me temo que snowfall no será de mucha ayuda allí. La cosa es que tendrás que exportar el marco de datos completo a los diferentes núcleos (ver ?sfExport) y aún así encontrar la manera de combinarlo. Ese tipo supera al objetivo principal de cambiar el valor en el entorno global, ya que es probable que desee mantener el uso de la memoria lo más bajo posible.

Puede sumergirse en las funciones de bajo nivel de snow para lograr que esto funcione. Véase el siguiente ejemplo:

#Some data 
Data <- data.frame(
    cell = 1:10, 
    value = sample(c(100,NA),10,TRUE), 
    x = 1:10, 
    y = 1:10 
) 
# A sample function 
sample.func <- function(){ 
    id <- which(is.na(Data$value)) # get the NA values 

    # this splits up the values from the dataframe in a list 
    # which will be passed to clusterApply later on. 
    parts <- lapply(clusterSplit(cl,id),function(i)Data[i,c("x","y")]) 

    # Here happens the magic 
    Data$value[id] <<- 
    unlist(clusterApply(cl,parts,function(x){ 
     x$x+x$y 
     } 
    )) 
} 
#now we run it 
require(snow) 
cl <- makeCluster(c("localhost","localhost"), type = "SOCK") 
sample.func() 
stopCluster(cl) 
> Data 
    cell value x y 
1  1 100 1 1 
2  2 100 2 2 
3  3  6 3 3 
4  4  8 4 4 
5  5 10 5 5 
6  6 12 6 6 
7  7 100 7 7 
8  8 100 8 8 
9  9 18 9 9 
10 10 20 10 10 

Usted todavía tendrá que copiar (parte de) sus datos, aunque para conseguir que los núcleos. Pero eso ocurrirá de todos modos cuando llame al snowfall funciones de alto nivel en marcos de datos, ya que snowfall usa la función de bajo nivel snow de todos modos.

Además, no se debe olvidar que si cambia un valor en un marco de datos, toda la estructura de datos se copiará también en la memoria. Por lo tanto, no ganará tanto agregando los valores uno a uno cuando regresen del clúster. Es posible que desee probar algunos enfoques diferentes y también hacer algunos perfiles de memoria.

Answer 3

Estoy de acuerdo con Joris en que tendrá que copiar sus datos a los otros núcleos. En el lado positivo, no tiene que preocuparse de que NA esté en los datos o no, dentro de los núcleos. Si el original se llama data.framecont.values: (errores tipográficos de restricción)

nnaidx<-is.na(cont.values$value) #where is missing data originally 
dfrnna<-cont.values[nnaidx,] #subset for copying to other cores 
calcValForDfrRow<-function(dfrRow){return(dfrRow$x+dfrRow$y)}#or whatever pleases you 
sfExport(dfrnna, calcValForDfrRow) #export what is needed to other cores 
cont.values$value[nnaidx]<-sfSapply(seq(dim(dfrnna)[1]), function(i){calcValForDfrRow(dfrnna[i,])}) #sfSapply handles 'reordering', so works exactly as if you had called sapply 

debería funcionar muy bien