Parallelo * all'interno delle funzioni

Question

Desidero utilizzare la funzionalità parallela del pacchetto plyr all'interno delle funzioni.

avrei pensato che il modo corretto per esportare gli oggetti che sono stati creati all'interno del corpo della funzione (in questo esempio, l'oggetto è df_2) è il seguente

# rm(list=ls()) 
library(plyr) 
library(doParallel) 

workers=makeCluster(2) 
registerDoParallel(workers,core=2) 

plyr_test=function() { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    df_2=data.frame(type=c("a","b"),x=3:4) 

    #export df_2 via .paropts 
    ddply(df_1,"type",.parallel=TRUE,.paropts=list(.export="df_2"),.fun=function(y) { 
    merge(y,df_2,all=FALSE,by="type") 
    }) 
} 
plyr_test() 
stopCluster(workers) 

Tuttavia, questo genera un errore di

Error in e$fun(obj, substitute(ex), parent.frame(), e$data) : 
    unable to find variable "df_2" 

così ho fatto qualche ricerca e ha scoperto che funziona, se posso esportare manualmente df_2

workers=makeCluster(2) 
registerDoParallel(workers,core=2) 

plyr_test_2=function() { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    df_2=data.frame(type=c("a","b"),x=3:4) 

    #manually export df_2 
    clusterExport(cl=workers,varlist=list("df_2"),envir=environment()) 

    ddply(df_1,"type",.parallel=TRUE,.fun=function(y) { 
    merge(y,df_2,all=FALSE,by="type") 
    }) 
} 
plyr_test_2() 
stopCluster(workers) 

Dà il risultato corretto

type x.x x.y 
1 a 1 3 
2 b 2 4 

ma ho anche scoperto che il seguente codice funziona

workers=makeCluster(2) 
registerDoParallel(workers,core=2) 

plyr_test_3=function() { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    df_2=data.frame(type=c("a","b"),x=3:4) 

    #no export at all! 
    ddply(df_1,"type",.parallel=TRUE,.fun=function(y) { 
    merge(y,df_2,all=FALSE,by="type") 
    }) 
} 
plyr_test_3() 
stopCluster(workers) 

plyr_test_3() dà anche il risultato corretto e non capisco perché. Avrei pensato di dover esportare df_2 ...

La mia domanda è: qual è il modo giusto per gestire il parallelo *ply all'interno delle funzioni? Ovviamente, plyr_test() non è corretto. In qualche modo ho la sensazione che l'esportazione manuale in plyr_test_2() sia inutile. Ma penso anche che lo plyr_test_3() sia una specie di cattivo stile di programmazione. Qualcuno potrebbe spiegarci per favore? Grazie ragazzi!

Answer 1

Il problema è che plyr_testdf_2 è definita plyr_test non accessibile dal pacchetto doParallel, e quindi blocca quando cerca di esportare df_2. Quindi questo è un problema di scoping. plyr_test2 evita questo problema perché non tenta di utilizzare l'opzione .export, ma come hai intuito, la chiamata a clusterExport non è necessaria.

La ragione per cui sia plyr_test2 e plyr_test3 avere successo è che df_2 viene serializzato insieme con la funzione anonima che viene passato alla funzione ddply tramite l'argomento .fun. Infatti, sia df_1 sia df_2 vengono serializzati insieme alla funzione anonima poiché tale funzione è definita all'interno di plyr_test2 e plyr_test3.È utile che df_2 sia incluso in questo caso, ma l'inclusione di df_1 non è necessaria e potrebbe danneggiare le prestazioni.

Finché lo standard df_2 viene catturato nell'ambiente della funzione anonima, non verrà mai utilizzato nessun altro valore di df_2, indipendentemente da ciò che viene esportato. A meno che non si possa impedire che venga catturato, è inutile esportarlo con .export o clusterExport perché verrà utilizzato il valore acquisito. Puoi solo metterti nei guai (come hai fatto con lo) cercando di esportarlo ai lavoratori.

Si noti che in questo caso, foreach non esegue l'esportazione automatica df_2 perché non è in grado di analizzare il corpo della funzione anonima per vedere quali simboli sono referenziati. Se si chiama foreach direttamente senza utilizzare una funzione anonima, allora vedrà il riferimento e lo esporterà automaticamente, rendendo inutile l'esportazione esplicita usando.

Si potrebbe evitare che l'ambiente di plyr_test dall'essere serializzato insieme con la funzione anonima modificando suo ambiente prima di passarlo al ddply:

plyr_test=function() { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    df_2=data.frame(type=c("a","b"),x=3:4) 
    clusterExport(cl=workers,varlist=list("df_2"),envir=environment()) 
    fun=function(y) merge(y, df_2, all=FALSE, by="type") 
    environment(fun)=globalenv() 
    ddply(df_1,"type",.parallel=TRUE,.fun=fun) 
} 

Uno dei vantaggi del pacchetto foreach è che non è così incoraggiarti a creare una funzione all'interno di un'altra funzione che potrebbe catturare accidentalmente un gruppo di variabili.

---

Questo problema mi fa pensare che foreach dovrebbe includere un'opzione chiamata .exportenv che è simile all'opzione clusterExportenvir. Sarebbe molto utile per plyr, poiché consentirebbe di esportare correttamente df_2 utilizzando. Tuttavia, il valore esportato non sarebbe stato ancora utilizzato a meno che l'ambiente contenente df_2 non fosse stato rimosso dalla funzione .fun.

Answer 2

Sembra un problema di ambito.

Ecco la mia "suite di test" che mi consente di esportare variabili diverse o evitare di creare df_2 all'interno della funzione. Aggiungo e rimuovo un dummy df_2 e df_3 al di fuori della funzione e confronto.

library(plyr) 
library(doParallel) 

workers=makeCluster(2) 
registerDoParallel(workers,core=2) 

plyr_test=function(exportvar,makedf_2) { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    if(makedf_2){ 
    df_2=data.frame(type=c("a","b"),x=3:4) 
    } 
    print(ls()) 

    ddply(df_1,"type",.parallel=TRUE,.paropts=list(.export=exportvar,.verbose = TRUE),.fun=function(y) { 
    z <- merge(y,df_2,all=FALSE,by="type") 
    }) 
} 
ls() 
rm(df_2,df_3) 
plyr_test("df_2",T) 
plyr_test("df_2",F) 
plyr_test("df_3",T) 
plyr_test("df_3",F) 
plyr_test(NULL,T) #ok 
plyr_test(NULL,F) 
df_2='hi' 
ls() 
plyr_test("df_2",T) #ok 
plyr_test("df_2",F) 
plyr_test("df_3",T) 
plyr_test("df_3",F) 
plyr_test(NULL,T) #ok 
plyr_test(NULL,F) 
df_3 = 'hi' 
ls() 
plyr_test("df_2",T) #ok 
plyr_test("df_2",F) 
plyr_test("df_3",T) #ok 
plyr_test("df_3",F) 
plyr_test(NULL,T) #ok 
plyr_test(NULL,F) 
rm(df_2) 
ls() 
plyr_test("df_2",T) 
plyr_test("df_2",F) 
plyr_test("df_3",T) #ok 
plyr_test("df_3",F) 
plyr_test(NULL,T) #ok 
plyr_test(NULL,F) 

Non so perché, ma .export cerca df_2 nel contesto globale al di fuori della funzione, (ho visto parent.env() nel codice, che potrebbe essere "più corretta" rispetto globale ambiente) mentre il calcolo richiede che la variabile si trovi nello stesso ambiente di ddply e la esporti automaticamente.

L'utilizzo di una variabile fittizia per df_2 al di fuori della funzione consente a .export di funzionare, mentre il calcolo utilizza df_2 all'interno.

Quando .export non trovi la variabile fuori della funzione, emette:

Error in e$fun(obj, substitute(ex), parent.frame(), e$data) : 
    unable to find variable "df_2" 

Con una variabile dummy df_2 all'esterno della funzione ma senza uno interno, .export è uscite sottili ma ddply:

Error in do.ply(i) : task 1 failed - "object 'df_2' not found" 

e 'possibile che dal momento che questo è un piccolo esempio o forse no parallelizzabile, in realtà è in esecuzione su un core ed evitando la necessità di esportare nulla. Un esempio più grande potrebbe fallire senza .export, ma qualcun altro può provarlo.

Answer 3

Grazie @ARobertson per il vostro aiuto! È molto interessante il fatto che plyr_test("df_2",T) funzioni quando un oggetto fittizio df_2 è stato definito al di fuori del corpo della funzione.

Come sembra ddply in definitiva chiama llply che, a sua volta, chiama foreach(...) %dopar% {...}.

Ho anche provato a riprodurre il problema con foreach, ma foreach funziona correttamente.

library(plyr) 
library(doParallel) 

workers=makeCluster(2) 
registerDoParallel(workers,core=2) 

foreach_test=function() { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    df_2=data.frame(type=c("a","b"),x=3:4) 
    foreach(y=split(df_1,df_1$type),.combine="rbind",.export="df_2") %dopar% { 
    #also print process ID to be sure that we really use different R script processes 
    cbind(merge(y,df_2,all=FALSE,by="type"),Sys.getpid()) 
    } 
} 

foreach_test() 
stopCluster(workers) 

getta l'allarme

Warning message: 
In e$fun(obj, substitute(ex), parent.frame(), e$data) : 
    already exporting variable(s): df_2 

ma restituisce il risultato corretto

type x.x x.y Sys.getpid() 
1 a 1 3   216 
2 b 2 4   1336 

Quindi, foreach sembra esportare automaticamente df_2. Infatti, il foreachvignette afferma che

...% Dopar funzione% notato che queste variabili sono state utilizzate, e che sono state definite nel contesto attuale. In tal caso % Dopar% verrà esportarli automaticamente in esecuzione parallela lavoratori una volta, e li usa per tutte le valutazioni di espressione per che foreach esecuzione ....

Quindi possiamo omettere .export="df_2" e l'uso

library(plyr) 
library(doParallel) 

workers=makeCluster(2) 
registerDoParallel(workers,core=2) 

foreach_test_2=function() { 
    df_1=data.frame(type=c("a","b"),x=1:2) 
    df_2=data.frame(type=c("a","b"),x=3:4) 
    foreach(y=split(df_1,df_1$type),.combine="rbind") %dopar% { 
    #also print process ID to be sure that we really use different R script processes 
    cbind(merge(y,df_2,all=FALSE,by="type"),Sys.getpid()) 
    } 
} 

foreach_test_2() 
stopCluster(workers) 

invece. Questo valuta senza un avvertimento.

L'esempio di variabile fittizia di ARobertson e il fatto che foreach funzioni correttamente mi fanno ora pensare che ci sia un problema nel modo in cui * ply gestisce gli ambienti.

La mia conclusione è:

Entrambe le funzioni plyr_test_3() e foreach_test_2() (che non esportare esplicitamente df_2) corsa senza errori e dare lo stesso risultato. Pertanto, ddply con parallel=TRUE funziona fondamentalmente. MA utilizzando uno stile di codifica più "verboso" (ad esempio, esportando esplicitamente df_2) come in plyr_test() genera un errore mentre foreach(...) %dopar% {...} emette solo un avviso.