使用 foreach 循环和并行处理生成矩阵

Question

我正在尝试将我当前使用的 for 循环转换为运行跨大型矩阵的进程。当前的 for 循环在 30 x 30 的部分中找到最大值，并创建一个具有最大值的新矩阵。

for 循环的当前代码如下所示：

mat <- as.matrix(CHM) # CHM is the original raster image
maxm <- matrix(nrow=nrow(mat)/30, ncol=ncol(mat)/30) # create new matrix with new dimensions

for(i in 1:dim(maxm)[1]) {
  for(j in 1:dim(maxm)[2]) {
    row <- 30 * (i - 1) + 1
    col <- 30 * (j - 1) + 1
    maxm[i,j] <- max(CHM[row:(row + 29), col:(col + 29)])
   }
 }

我想将其转换为 foreach 循环以使用并行处理。我已经生成了以下代码，但是这个 dosen 工作。我不确定如何在 foreach 循环中生成新矩阵：

ro<-nrow(mat)/30
co<-ncol(mat)/30
maxm <- matrix(nrow=nrow(mat)/30, ncol=ncol(mat)/30)

foreach(i=ro, .combine='cbind') %:%
  foreach(j=co, .combine='c') %dopar% {
    row <- 30 * (i - 1) + 1
    col <- 30 * (j - 1) + 1
    maxm[i,j]<-(max(CHM[row:(row + 29), col:(col + 29)]))
  }

有什么建议吗！

Answer 1

让我试着在这里得到答案。

据我所知，R使用集群系统进行并行计算，每个节点都有自己的环境。因此，foreach-%dopar%，首先，将所有当前的 .globalEnv 复制到每个集群节点，然后尝试运行编写在循环体中的代码。代码执行后没有备份。您只会得到 result = foreach(...) { } 的结果。因此，每个节点中的代码 maxm[i,j]<-(max(CHM[row:(row + 29), col:(col + 29)])) 仅更改矩阵的本地副本，仅此而已。所以，"correct" 代码可能是这样的：

mat <- as.matrix(CHM);
ro<-nrow(mat)/30;
co<-ncol(mat)/30;

maxm = foreach(i=1:ro, .combine='cbind') %:% 
{ 
   result = foreach(j = 1:co, .combine='c') %dopar% 
            { 
                row <- 30 * (i - 1) + 1; 
                col <- 30 * (j - 1) + 1; 
                max(CHM[row:(row + 29), col:(col + 29)]); 
            } 
   result; 
}

也许还需要对 maxm 使用 as.matrix。

Answer 2

在并行执行任何操作之前，应该尝试查看是否可以进行矢量化。一旦完成问题 'is parallelization reasonable?'

在这个具体示例中，并行化不太可能像您预期的那样快，因为在每次迭代中您都将输出保存到一个公共对象中。 R 在并行化中通常不支持这一点，相反，人们应该在所谓的 'embarrassingly parallel-able' 问题中寻求并行化，直到人们更好地理解并行问题的工作原理。简而言之：不要对 R 中的数据执行并行更改，除非您知道自己在做什么。它不太可能更快。

也就是说在你的情况下它实际上变得非常棘手。您似乎在执行 'rolling-max window'，输出应保存在组合矩阵中。将数据直接保存到其他矩阵中的另一种方法是 return 具有 3 列的矩阵 x、i、j，其中后两个是指示的索引其中 row/column x 的值应该放在其中。

为了使其工作，正如 Dmitriy 在他的回答中指出的那样，需要将数据导出到每个 cluster（并行会话），以便我们可以使用它。之后，以下示例显示了如何执行并行化

首先：创建集群并导出数据集

set.seed(1)
#Generate test example
n <- 3000
dat <- matrix(runif(n^2), ncol = n)
library(foreach)
library(doParallel)
#Create cluster
cl <- parallel::makeCluster(parallel::detectCores())
#Register it for the foreach loop
doParallel::registerDoParallel(cl)
#Export the dataset (could be done directly in the foreach, but this is more explicit)
parallel::clusterExport(cl, "dat")

接下来我们进入 foreach 循环。请注意，根据文档，嵌套的 foreach 循环应使用 %:% 标记分隔，如下面的示例所示：

output <- foreach(i = 1:(nrow(dat)/30), .combine = rbind, .inorder = FALSE) %:% 
    foreach(j = 1:(ncol(dat)/30), .combine = rbind, .inorder = FALSE) %dopar%{
        row <- 30 * (i - 1) + 1
        col <- 30 * (j - 1) + 1
        c(x = max(dat[row:(row + 29), col:(col + 29)]), i = i, j = j)
    }

注.inorder = FALSE。因为我 return 我不关心索引，只关心速度。最后但同样重要的是，我们需要创建矩阵。 Matrix 包函数 Matrix::SparseMatrix 允许指定值和索引。

output <- Matrix::sparseMatrix(output[,"i"], output[,"j"], x = output[,"x"])

这还是比较慢的。对于 n = 3000，执行计算大约需要 6 秒 + 导出数据的 not-insignificant 开销。但它可能比使用顺序循环的相同方法更快。

使用 foreach 循环和并行处理生成矩阵

Produce a matrix using a foreach loop and parallel processing

foreach

for-loop

r

matrix

doparallel