拆分数据集并将子集并行传递给函数，然后重新组合结果

Question

这是我尝试使用 foreach 包所做的事情。 我有 600 行和 58000 列的数据集，其中有很多缺失值。

我们需要使用名为“missForest”的包来估算缺失值，其中它不是并行的，一次运行这些数据需要很长时间。 所以，我正在考虑将数据分成 7 个数据集（我有 7 个核心），它们具有相同的行数（我的行）和不同数量的 col（标记）。 然后使用%dopar%将数据集并行传递给missForest？

我不知道如何将数据划分为更小的数据集并将这些数据集传递给missForest，然后重新组合输出！

如果你能告诉我怎么做，我会非常感激？

这是一个小例子，形成 BLR 包，演示我的问题：

library(BLR)
library(missForest)
data(wheat)
X2<- prodNA(X, 0.1)
dim(X2)                 ## i need to divide X2 to several 7 data frames (ii)

X3<- missForest(X2)

X3$Ximp  ## combine ii data frames

Answer 1

在并行处理大型矩阵时，仅传递每个集群工作程序所需的数据可能非常重要。在 Linux 上直接或间接使用 mclapply 时，这不是问题。但是在 Windows 上，输入数据是通过套接字连接发送到集群工作人员的，因此它可能非常重要。

对于这种情况，我使用itertools 包中的isplitCol 函数。它在矩阵的列块上创建一个迭代器。使用chunks 参数，您可以拆分矩阵，以便每个集群工作人员获得一个子矩阵。

这是您的示例到 foreach 的翻译，它使用 isplitCol 将输入矩阵拆分为 7 个子矩阵，因此与将 X2 自动导出到每个工作人员相比，发送给每个工作人员的数据减少了七倍工人：

library(doParallel)
library(itertools)
library(BLR)
library(missForest)
ncores <- 7
cl <- makePSOCKcluster(ncores)
registerDoParallel(cl)
data(wheat)
X2 <- prodNA(X, 0.1)
X3 <- foreach(m=isplitCols(X2, chunks=ncores), .combine='cbind',
              .packages='missForest') %dopar% {
  missForest(m)$ximp
}
print(X3)
stopCluster(cl)

Answer 2

library(multicore)
n.cores <- 7
cuts <- cut(1:ncol(X2), n.cores)
X3 <- mclapply(levels(cuts), function(x) 
  missForest(X2[, cuts == x])$ximp , mc.cores=n.cores)
X3 <- do.call(cbind, X3)

cut 将列拆分为 7 个间隔，然后 mclapply 发送到您的 7 个核心。最后将它们全部绑定在一起

编辑：添加我的 foreach 实现。注意：我以前从未使用过这个包，但它似乎在做我期望的事情

library(doParallel)
library(foreach)
n.cores <- 7
cuts <- cut(1:ncol(X2), n.cores)

cl <- makeCluster(n.cores)
registerDoParallel(cl)

X3 <- foreach(x=levels(cuts), .combine=cbind, .multicombine=TRUE) %dopar% { 
  library(missForest); missForest(X2[, cuts == x])$ximp 
}

Answer 3

您必须将矩阵拆分为ncores 部分，然后再次组合它们。由于您使用的是随机森林，因此您可以随机拆分数据（并执行多次以检查和验证结果）。

ncores = 7

split = sample(seq(ncores), size=ncol(X2), replace=TRUE) # random partitioning

X3 = foreach(i=seq_len(ncores), .combine=cbind, .inorder=FALSE) %dopar% {
  ind = which(split==i) # the selected rows for this core
  X = rbind(ind, missForest(X2[,ind])$ximp) # add the index as first row!
}

ind = X3[1,] # get all the index back
ind = sort(ind, index.return=TRUE)$ix # sort the index to recover the original row order
X3 = X3[-1,ind] # remove the index