doParallel 而不是 apply答案

【问题标题】：doParallel instead of applydoParallel 而不是 apply
【发布时间】：2021-09-20 17:27:42
【问题描述】：

我一直在使用apply 使用Windows 10 将data.frame 的每一行提交给R 中的function。这种方法有效。然而，随着function 变得更加复杂并且数据集更大，这种方法变得不可行。因此，我希望使用并行处理将data.frame 的每一行提交并运行到function。

我之前在我的Windows 10 笔记本电脑和Unix 集群上都使用过doParallel，但仅用于运行多个外部R 文件。永远不要在一个R 文件中将data.frame 的单独行提交给function。有人可以告诉我如何实现后者吗？如果必须，我想我可以在下面的代码中为data.frame master.iter 的每一行创建单独的R 文件，但必须有更简单的方法。

这是一个简单的示例，它与 apply 一起运行并获得所需的结果：

master.iter <- read.table(text = '
    scenario  aaa   bbb   ccc   ddd   eee
       1       1     5     0     20    10
       2       1    10     0   2000  1000
', header = TRUE, stringsAsFactors = FALSE)

master.function <- function(scenario, aaa, bbb, ccc, ddd, eee) {

  scenario <- as.numeric(c(scenario))
  aaa <- as.numeric(c(aaa))
  bbb <- as.numeric(c(bbb))
  ccc <- as.numeric(c(ccc))
  ddd <- as.numeric(c(ddd))
  eee <- as.numeric(c(eee))
  AAA <- seq(aaa,bbb,1)
  BBB <- AAA * ddd
  CCC <- AAA * eee

  my.table <- data.frame(AAA = AAA,
                         BBB = BBB,
                         CCC = CCC)

  output.list <- list(scenario = scenario,
                      aaa = aaa, bbb = bbb, ccc = ccc, ddd = ddd, eee = eee,
                      my.table = my.table)

  master_output <- do.call(cbind, output.list)

  return = list(master_output = master_output)
}

function.output <- apply(master.iter, 1, function(x) {master.function( x[1],  x[2],  x[3],  x[4],  x[5],  x[6])})

master.df <- do.call("rbind", lapply(function.output, as.data.frame))
colnames(master.df) <- names(function.output[[1]]$master_output)

desired.result <- read.table(text = '
   scenario aaa bbb ccc  ddd  eee my.table.AAA my.table.BBB my.table.CCC
          1   1   5   0   20   10            1           20           10
          1   1   5   0   20   10            2           40           20
          1   1   5   0   20   10            3           60           30
          1   1   5   0   20   10            4           80           40
          1   1   5   0   20   10            5          100           50
          2   1  10   0 2000 1000            1         2000         1000
          2   1  10   0 2000 1000            2         4000         2000
          2   1  10   0 2000 1000            3         6000         3000
          2   1  10   0 2000 1000            4         8000         4000
          2   1  10   0 2000 1000            5        10000         5000
          2   1  10   0 2000 1000            6        12000         6000
          2   1  10   0 2000 1000            7        14000         7000
          2   1  10   0 2000 1000            8        16000         8000
          2   1  10   0 2000 1000            9        18000         9000
          2   1  10   0 2000 1000           10        20000        10000
', header = TRUE)

这是我通常用来将单独的R 文件提交到Ubuntu cluster 的R 代码。针对上述问题，我尝试修改下面的R 代码。但是，我还没有找到解决方案。

setwd('/home/ubuntu/')
library(doParallel)
detectCores()
my.AWS.n.cores <- detectCores()
registerDoParallel(my.cluster <- makeCluster(my.AWS.n.cores))
folderName <- 'R_files_a'
files <- list.files(folderName, full.names=TRUE)
start.time <- Sys.time()
foreach(file = files, .errorhandling = "remove") %dopar% {
  source(file)
}
stopCluster(my.cluster)
end.time <- Sys.time()
total.time.c <- end.time-start.time
total.time.c

【问题讨论】：

你可以试试library(collapse);dapply(master.iter, MARGIN = 1, function(x) {master.function( x[1], x[2], x[3], x[4], x[5], x[6])}, paralle = TRUE)
该错误与代码无关。这与缺少包或 Rcpp 包的旧版本有关
仅供参考，如果您有一个有效的apply() 解决方案，那么您可以将其替换为future.apply::future_apply()。要并行化，请设置，例如 future::plan("multisession")。有关详细信息，请参阅 future.apply.futureverse.org>。

标签： r parallel-processing apply

【解决方案1】：

我们可能会使用collapse

library(collapse)
dapply(master.iter, MARGIN = 1, function(x) {
    master.function( x[1],  x[2],  x[3],  x[4],  x[5],  x[6])
      }, parallel = TRUE)

-输出

 [1]     1     1     1     1     1     1     1     1     1     1     5     5     5     5     5     0     0     0     0     0    20    20    20    20    20
 [26]    10    10    10    10    10     1     2     3     4     5    20    40    60    80   100    10    20    30    40    50     2     2     2     2     2
 [51]     2     2     2     2     2     1     1     1     1     1     1     1     1     1     1    10    10    10    10    10    10    10    10    10    10
 [76]     0     0     0     0     0     0     0     0     0     0  2000  2000  2000  2000  2000  2000  2000  2000  2000  2000  1000  1000  1000  1000  1000
[101]  1000  1000  1000  1000  1000     1     2     3     4     5     6     7     8     9    10  2000  4000  6000  8000 10000 12000 14000 16000 18000 20000
[126]  1000  2000  3000  4000  5000  6000  7000  8000  9000 10000

【讨论】：

【解决方案2】：

df <- master.iter

library(doParallel)

ncores <- detectCores()-1
cl <- parallel::makeCluster(ncores)
registerDoParallel(cl)

v <- foreach(i = 1:nrow(df)) %dopar% {
  master.function(df[i,1], df[i,2], df[i,3], df[i,4], df[i,5], df[i,6])
}
stopCluster(cl)

【讨论】：