r - 使用 doparallel 在 foreach 循环内循环

Question

我有一个包含循环的函数

myfun = function(z1.d, r, rs){
  x = z1.d[,r]
  or.d = order(as.vector(x), decreasing=TRUE)[rs]
  zz1.d = as.vector(x)
  r.l = zz1.d[or.d]

  y=vector()
  for (i in 1:9)
  {
    if(i<9) y[i]=mean( x[(x[,r] >= r.l[i] & x[,r] < r.l[i+1]),r] ) else{
      y[i] =  mean( z1.d[(x >= r.l[9]),r] )}
  }
  return(y)
}

rs 是一个数值向量，z1.d 是一个动物园，y 也是一个数值向量。

当我尝试在并行循环内运行该函数时:

cls = makePSOCKcluster(8)
registerDoParallel(cls)

rlarger.d.1  = foreach(r=1:dim(z1.d)[2], .combine = "cbind") %dopar% {    
  myfun(z1.d, r, rs)}

stopCluster(cls)

我收到以下错误:

Error in { : task 1 failed - "incorrect number of dimensions"

我不知道为什么，但我意识到如果我将循环从函数中取出，它不会给出错误。

此外，如果我使用 %do% 而不是 %dopar% 运行完全相同的代码(因此不是并行运行)，它可以正常工作(缓慢但没有错误)。

编辑:根据要求，这里是参数示例:

dim(z1.d)
[1] 8766  107
> z1.d[1:4,1:6]
                    AU_10092 AU_10622 AU_12038 AU_12046 AU_13017 AU_14015
1966-01-01 23:00:00       NA       NA       NA    1.816        0    4.573
1966-01-02 23:00:00       NA       NA       NA    9.614        0    4.064
1966-01-03 23:00:00        0       NA       NA    0.000        0    0.000
1966-01-04 23:00:00        0       NA       NA    0.000        0    0.000

> rs
[1] 300 250 200 150 100  75  50  30  10

r 在 foreach 循环中定义

Answer 1

弹出错误是因为您未能在工作线程上启动 zoo。因此，工作人员不知道如何正确处理动物园对象，而是将它们作为矩阵处理，而在子集化时，矩阵的行为方式不同!因此，快速解决您所说的问题的方法是将 .packages="zoo" 添加到您的 foreach 调用中。

在我看来，你甚至不需要进行并行计算。如果您使用数字向量而不是动物园对象，您可以显着增强您的功能:

# sample time series to match your object's size
set.seed(1234)
z.test <- as.zoo(replicate(107,sample(c(NA,runif(1000,0,10)),size = 8766, replace = TRUE)))

myfun_new <-  function(z, r, rs){
  x <-  as.numeric(z[,r])
  r.l <- x[order(x, decreasing=TRUE)[rs]]
  res_dim <- length(rs)
  y=numeric(res_dim)
  for (i in 1:res_dim){
    if(i< res_dim){ 
      y[i] <- mean( x[(x >= r.l[i] & x < r.l[i+1])], na.rm = TRUE ) 
    }else{
      y[i] <-   mean( x[(x >= r.l[res_dim])] , na.rm = TRUE)
    }
  }
  return(y)
}

简单的计时显示了改进:

system.time({
  cls = makePSOCKcluster(4)
  registerDoParallel(cls)
  rlarger.d.1 = foreach(r=1:dim(z.test)[2],.packages = "zoo", .combine = "cbind") %dopar% { 
    myfun(z.test, r, rs)}
  stopCluster(cls)
})
##  User      System verstrichen 
##  0.08        0.10       10.93
system.time({
  res <-sapply(1:dim(z.test)[2], function(r){myfun_new(z.test, r, rs)})
})
##  User      System verstrichen 
##  0.48        0.21        0.68

虽然结果相同(仅列名称不同)

all.equal(res, rlarger.d.1, check.attributes = FALSE)
## [1] TRUE