使用 mclapply 并行化 R 代码不会产生正确的结果答案

【问题标题】：Parallelizing the R code using mclapply does not generate the correct results使用 mclapply 并行化 R 代码不会产生正确的结果
【发布时间】：2020-03-26 02:39:22
【问题描述】：

我有一个dataframedf，需要应用一个函数来为每一列（calc.fitness）打分：

    df
  #         ch1     ch2   ch3    ch4   ch5    ch6   ch7   ch8
  # g1       5      2      7     10     7     10    10    6
  # g2       1      4      5      4     1     2     5     4
  # g3      16      14     7      4     2     2     8     7
  # g4       7      5      5      3     2     5     1     6
  # g5       7      2      1      3     7     2     4     1
  # g6       4      7      11     4     9     3     9     14
  # g7      12      8      6      7     5     9     7     4
  # g8       4      2      3      2     2     4     1     1
  # g9       1      2      1      1     2     1     2     1

使用sapply，我会得到以下结果，这是正确的，但随着df 的大小增加，会非常耗时：

sapply(as.list(df), calc.fitness,filterTable=my.df)
#     ch1          ch2            ch3            ch4            ch5            ch6            ch7             ch8 
# 8.481359e-02  6.419552e-01   5.847587e-02   6.713477e-02   1.552056e-01   1.305787e+34   2.805074e-01    2.039931e+00

我使用 [Tag:mclapply` 使其更快，如下所示：

numCores <- detectCores()
result <- unlist(mclapply(1:8, function(x) {
  return(calc.fitness(df[,x], filterTable=my.df))}, mc.preschedule = TRUE, mc.cores = numCores))

# result
# [1] 8.481359e-02 8.481359e-02 8.481359e-02 8.481359e-02 1.305787e+34 1.305787e+34 1.305787e+34 1.305787e+34

但结果显示，mclapply 无法正常工作，我不知道是什么问题以及如何解决。非常感谢任何帮助！

PS：calc.fitness 是一个很长的方法，我在这里尝试使它更短：

calc.fitness <- function(df.val, filterTable = my.df) {
  input.path <- "/home/Nikki/Desktop/v2017.0/exec/Input_2017.txt"
  filterTable$xe <-  df.val[1]
  filterTable$xth <- df.val[2]
  filterTable$xfi <- df.val[3]
  filterTable$xfw <- df.val[4]
  filterTable$xfm <- df.val[5]
  filterTable$xls <- df.val[6]
  filterTable$xhls <- df.val[7]
  filterTable$xvt <- df.val[8]
  filterTable$xvd <- df.val[9]
  write.fwf(filterTable,append = TRUE,file = paste("Input_2017", ".txt", sep = ""),width = 25, rownames = F,colnames = F,quote = F)
  command <- "wine  /home/Nikki/Desktop/v2017.0/exec/2017File.exe"
  system(command)
  output.file <-read.table("/home/Nikki/Desktop/v2017.0/exec/Output_2017.txt",header = TRUE,fill = TRUE)
  output.pgt <- as.numeric(levels(output.file$pgt))[output.file$pgt]
  calc.sol <- output.pgt[!is.na(output.pgt)]
  opt.sol <- filterTable$PressureDropGL
  n <- length(calc.sol)
  subtract.val <- calc.sol - opt.sol
  denominator <- opt.sol
  sq.output <-  (subtract.val / denominator) ^ 2
  fitness.val <- sum(sq.output) / n
  return(fitness.val)
}# end of function

我的.df：

感谢您的帮助。

【问题讨论】：

您好 Nikki，如果您提供函数 calc.fitness 的代码或它来自的包，会更容易提供帮助。
嗨，Ian，calc.fitness 是我编写的一个很长的函数，我试图让它更短且易于理解！我编辑了帖子，你可以看到这个功能！提前非常感谢您！
看到你的函数后，你正在将其中生成的数据附加到一个文件中。如果按顺序执行，这可能会很好，但是当您在并行进程中执行此操作时，您一定会遇到麻烦。查看我编辑的评论。

标签： dataframe sapply r parallel-processing data-manipulation sapply mclapply

【解决方案1】：

sapply 简化为一个矩阵，而取消列出列列表会依次返回每列的向量。考虑使用cumsum 函数作为说明：

df <-
    structure(
        list(
            ch1 = c(5L, 1L, 16L, 7L, 7L, 4L, 12L, 4L, 1L),
            ch2 = c(2L, 4L, 14L, 5L, 2L, 7L, 8L, 2L, 2L),
            ch3 = c(7L, 5L, 7L, 5L, 1L, 11L, 6L, 3L, 1L),
            ch4 = c(10L, 4L, 4L, 3L, 3L, 4L, 7L, 2L, 1L),
            ch5 = c(7L, 1L, 2L, 2L, 7L, 9L, 5L, 2L, 2L),
            ch6 = c(10L, 2L, 2L, 5L, 2L, 3L, 9L, 4L, 1L),
            ch7 = c(10L, 5L, 8L, 1L, 4L, 9L, 7L, 1L, 2L),
            ch8 = c(6L, 4L, 7L, 6L, 1L, 14L, 4L, 1L, 1L)
        ),
        class = "data.frame",
        row.names = c("g1", "g2", "g3", "g4", "g5", "g6", "g7", "g8", "g9")
    )

sapply(as.list(df), cumsum)
#>       ch1 ch2 ch3 ch4 ch5 ch6 ch7 ch8
#>  [1,]   5   2   7  10   7  10  10   6
#>  [2,]   6   6  12  14   8  12  15  10
#>  [3,]  22  20  19  18  10  14  23  17
#>  [4,]  29  25  24  21  12  19  24  23
#>  [5,]  36  27  25  24  19  21  28  24
#>  [6,]  40  34  36  28  28  24  37  38
#>  [7,]  52  42  42  35  33  33  44  42
#>  [8,]  56  44  45  37  35  37  45  43
#>  [9,]  57  46  46  38  37  38  47  44

unlist(parallel::mclapply(1:8, function(x) {
    return(cumsum(df[,x]))}, mc.preschedule = TRUE, mc.cores = 4L))
#>  [1]  5  6 22 29 36 40 52 56 57  2  6 20 25 27 34 42 44 46  7 12 19 24 25 36 42
#> [26] 45 46 10 14 18 21 24 28 35 37 38  7  8 10 12 19 28 33 35 37 10 12 14 19 21
#> [51] 24 33 37 38 10 15 23 24 28 37 44 45 47  6 10 17 23 24 38 42 43 44

do.call(cbind, parallel::mclapply(1:8, function(x) {
    return(cumsum(df[,x]))}, mc.preschedule = TRUE, mc.cores = 4L))
#>       [,1] [,2] [,3] [,4] [,5] [,6] [,7] [,8]
#>  [1,]    5    2    7   10    7   10   10    6
#>  [2,]    6    6   12   14    8   12   15   10
#>  [3,]   22   20   19   18   10   14   23   17
#>  [4,]   29   25   24   21   12   19   24   23
#>  [5,]   36   27   25   24   19   21   28   24
#>  [6,]   40   34   36   28   28   24   37   38
#>  [7,]   52   42   42   35   33   33   44   42
#>  [8,]   56   44   45   37   35   37   45   43
#>  [9,]   57   4618   46   38   37   38   47   44

^{由reprex package (v0.3.0) 于 2020 年 3 月 25 日创建}

编辑： 看到您的函数后，您正在将其中生成的数据附加到一个文件中。如果按顺序执行，这可能会很好，但是当您在并行进程中执行此操作时，您一定会遇到麻烦。并行生成多个 wine 进程本身也可能不是最有效的过程，即使它产生了正确的结果（使用 profvis 包分析你的（线性）代码会显示瓶颈）。 2017File.exe 有什么替代方法来计算fitness.val？

如果您的计划确实是按顺序附加列中的结果，那么要使用您的 exe 文件正确启动结果的并行生成，您可能必须保存按顺序增长的文件的唯一实例（您的 write.fwf 命令），然后将这些并行传递给 exe 命令，为每个连续步骤生成唯一的 output.txt 文件，然后以正确的顺序加载结果。

【讨论】：