r - 函数内的并行 *ply

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

它看起来像一个范围问题。

这是我的“测试套件”，它允许我 .export 不同的变量或避免在函数内创建 df_2。我在函数之外添加和删除了一个虚拟的 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 的情况下失败，但其他人可以尝试。