多集群并行引导中的可变范围

Question

在运行并行计算时，我试图弄清楚如何将函数和包传递给boot()函数。在循环中加载包或定义函数似乎非常昂贵。我经常用于其他并行任务的foreach()函数有一个.packages和.export参数，可以很好地处理这个问题(请参阅这个SO question)，但是我不知道如何使用引导包来完成这个任务。

下面是一个毫无意义的示例，它展示了切换到并行时发生了什么：

library(boot)
myMean <- function(x) mean(x)
meaninglessTest <- function(x, i){
  return(myMean(x[i]))
}

x <- runif(1000)

bootTest <- function(){
  out <- boot(data=x, statistic=meaninglessTest, R=10000, parallel="snow", ncpus=4)
  return(boot.ci(out, type="perc"))
}

bootTest()

抱怨(如预期的)它找不到myMean。

Sidenote:当运行这个示例时，它的运行速度比一个内核慢，可能是因为将这个简单的任务拆分到多个核上比实际的任务更耗时。为什么默认不将R/ncpus拆分成作业批次--这不是默认行为的原因吗？

更新:正如Steve所指出的，引导()使用的parLapply实际上将作业拆分为甚至批/块。该函数是clusterApply的整洁包装器：

docall(c, clusterApply(cl, splitList(x, length(cl)), lapply, 
    fun, ...))

当我放大重复次数的时候，这并没有更好的表现，这让我有点惊讶：

> library(boot)
> set.seed(10)
> x <- runif(1000)
> 
> Reps <- 10^4
> start_time <- Sys.time()
> res <- boot(data=x, statistic=function(x, i) mean(x[i]), 
+             R=Reps, parallel="no")
> Sys.time()-start_time
Time difference of 0.52335 secs
> 
> start_time <- Sys.time()
> res <- boot(data=x, statistic=function(x, i) mean(x[i]), 
+             R=Reps, parallel="snow", ncpus=4)
> Sys.time()-start_time
Time difference of 3.539357 secs
> 
> Reps <- 10^5
> start_time <- Sys.time()
> res <- boot(data=x, statistic=function(x, i) mean(x[i]), 
+             R=Reps, parallel="no")
> Sys.time()-start_time
Time difference of 5.749831 secs
> 
> start_time <- Sys.time()
> res <- boot(data=x, statistic=function(x, i) mean(x[i]), 
+             R=Reps, parallel="snow", ncpus=4)
> Sys.time()-start_time
Time difference of 23.06837 secs

我希望这只是由于非常简单的平均函数，而更复杂的情况表现得更好。我必须承认，我觉得这有点令人不安，因为在10.000 & 100.000的情况下，集群初始化时间应该是相同的，但是绝对时间差增加了，4核版本花费了5倍的时间。我想这一定是列表合并的结果，因为我找不到其他解释。

Answer 1

如果要并行执行的函数(本例中为meaninglessTest)有额外的依赖项(如myMean)，标准解决方案是通过clusterExport函数将这些依赖项导出到集群。这需要创建一个集群对象并通过"cl“参数将其传递给boot：

library(boot)
library(parallel)
myMean <- function(x) mean(x)
meaninglessTest <- function(x, i){
  return(myMean(x[i]))
}
cl <- makePSOCKcluster(4)
clusterExport(cl, 'myMean')

x <- runif(1000)

bootTest <- function() {
  out <- boot(data=x, statistic=meaninglessTest, R=10000,
              parallel="snow", ncpus=4, cl=cl)
  return(boot.ci(out, type="perc"))
}

bootTest()
stopCluster(cl)

请注意，一旦集群工作人员被初始化，它们可以被boot多次使用，并且不需要重新初始化，所以它并不昂贵。

要在群集工作人员上加载包，可以使用clusterEvalQ

clusterEvalQ(cl, library(randomForest))

这很好，也很简单，但是对于更复杂的工作人员初始化，我通常创建一个"worker init“函数，并通过clusterCall执行它，这对于在每个工作人员上执行一次函数是完美的。

至于您的侧注，性能很差，因为统计函数做的工作太少了，正如您所说的，但是我不知道为什么您认为工作没有在工人之间平均分配。在这种情况下，parLapply函数用于并行地执行工作，并且它确实均衡和有效地分配了工作，但这并不能保证比使用lapply顺序运行更好的性能。但也许我误解了你的问题。