Параллельный * слой внутри функций

Question

Я хочу использовать параллельную функциональность пакета plyr внутри функций.

я бы подумал, что правильный способ экспортировать объекты, которые были созданы в теле функции (в данном примере, объект df_2) выглядит следующим образом

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

Однако, это бросает ошибка

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

Так что я сделал некоторые исследования и выяснили, что она работает, если я экспортировать 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) 

Это дает правильный результат

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

Но я также узнал, что следующий код работает

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() также дает правильный результат, и я не понимаю, почему. Я бы подумал, что мне нужно экспортировать df_2 ...

Мой вопрос: как правильно обращаться с параллельными *ply внутри функций? Очевидно, что plyr_test() неверен. Я как-то чувствую, что ручной экспорт в plyr_test_2() бесполезен. Но я также думаю, что plyr_test_3() - это своего рода плохой стиль кодирования. Может кто-нибудь прокомментировать это? Спасибо, парни!

Answer 1

Проблема в том, что plyr_testdf_2 определяется в plyr_test, который не доступен из doParallel пакета, и поэтому он терпит неудачу, когда он пытается экспортировать df_2. Так что это проблема. plyr_test2 избегает этой проблемы, потому что не пытается использовать опцию .export, но, как вы догадались, вызов clusterExport не нужен.

Причина, по которой обе plyr_test2 и plyr_test3 успеха является то, что df_2 сериализации вместе с анонимной функции, которая передается функции ddply.fun через аргумент. Фактически, и df_1, и df_2 сериализуются вместе с анонимной функцией, потому что эта функция определена внутри plyr_test2 и plyr_test3.Полезно, что в этом случае включен df_2, но включение df_1 не является необходимым и может повредить вашей работе.

Пока df_2 улавливается в среде анонимной функции, никакое другое значение df_2 никогда не будет использоваться, независимо от того, что вы экспортируете. Если вы не можете предотвратить его захват, бессмысленно экспортировать его либо с .export, либо clusterExport, потому что будет использовано захваченное значение. Вы можете только попасть в беду (как вы делали .export), пытаясь экспортировать его рабочим.

Обратите внимание, что в этом случае foreach не производит автоматический экспорт df_2, поскольку он не способен анализировать тело анонимной функции, чтобы увидеть, на какие символы ссылаются. Если вы вызываете foreach напрямую, не используя анонимную функцию, тогда он увидит ссылку и автоматически экспортирует ее, что делает ненужным явно экспортировать ее с помощью .export.

Вы могли бы предотвратить окружение plyr_test из сериализации вместе с анонимной функции, изменяя его окружение, прежде чем передать его 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) 
} 

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

---

Этот вопрос подсказывает мне, что foreach должен включать в себя опцию под названием .exportenv, который похож на вариант clusterExportenvir. Это было бы очень полезно для plyr, так как это позволило бы корректно экспортировать df_2 с использованием .export. Однако эта экспортированная стоимость все равно не будет использоваться, если только среда, содержащая df_2, была удалена из функции .fun.

Answer 2

Похоже, проблема с объемом.

Вот мой «набор тестов», который позволяет мне экспортировать различные переменные или не создавать df_2 внутри функции. Я добавляю и удаляю dumm_df_2 и df_3 вне функции и сравниваю.

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) 

Я не знаю, почему, но .export ищет df_2 в глобальной окружающей среды за пределами функции, (я видел parent.env() в коде, который может быть «более правильным», чем глобальное среда), в то время как для вычисления требуется, чтобы переменная находилась в той же среде, что и ddply, и автоматически экспортировала ее.

Использование фиктивной переменной для df_2 вне функции позволяет работать .export, в то время как вычисление использует df_2 внутри.

Когда .export не может найти переменную за пределами функции, он выводит:

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

С df_2 фиктивной переменной вне функции, но без одной внутренней стороны, .export это хорошо, но ddply выходы:

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

Вполне возможно, что так как это небольшой пример, или, может быть, не параллелизуемое, это на самом деле работает на одном ядре и избежать необходимости экспортировать что-либо. Более крупный пример может выйти из строя без .export, но кто-то другой может попробовать это.

Answer 3

Спасибо @ ARobertson за вашу помощь! Очень интересно, что plyr_test("df_2",T) работает, когда объект фиктивного объекта df_2 был определен вне тела функции.

Как представляется, ddply в конечном счете вызывает llply, который, в свою очередь, вызывает foreach(...) %dopar% {...}.

Я также попытался воспроизвести проблему с foreach, но foreach работает нормально.

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) 

Он бросает предупред

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

но он возвращает правильный результат

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

Так, foreach кажется, автоматически экспортировать df_2. Действительно, foreachvignette утверждает, что

...% dopar% функции заметили, что сослались эти переменные, и , что они были определены в текущей среде. В этом случае % dopar% будет автоматически экспортировать их параллельное выполнение рабочих один раз, и использовать их для всех оценок выражения для что Foreach исполнения ....

Поэтому мы можем опустить .export="df_2" и использовать

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) 

вместо этого. Это оценивается без предупреждения.

Примером примера фиктивной фигуры ARobertson и фактом, что foreach отлично работает, заставляет меня думать, что есть проблема в том, как * ply обрабатывает среды.

Мой вывод:

Обе функции plyr_test_3() и foreach_test_2() (которые явно не экспортировать df_2) пробег без ошибок и дают тот же результат. Поэтому ddply с parallel=TRUE в основном работает. НО использование более «подробного» стиля кодирования (т. Е. Явно экспорт df_2), например, в plyr_test(), вызывает ошибку, тогда как foreach(...) %dopar% {...} только выдает предупреждение.