使用 R 中的并行函数最小化开销 [关闭]答案

【问题标题】：Minimizing overhead with parallel functions in R [closed]使用 R 中的并行函数最小化开销 [关闭]
【发布时间】：2017-01-11 00:58:40
【问题描述】：

我尝试报告我在mclapply 中遇到的关于不允许大返回值的错误。

显然该错误已在开发版本中修复，但我对响应者的评论更感兴趣：

序列化对象的大小有 2GB 的限制，例如mclapply 可以从分叉的进程返回，此示例正在尝试 16GB。这在 R-devel 中已被取消（对于 64 位构建），但这种用法非常不寻常且效率很低（该示例需要 ca 150GB，因为（取消）序列化涉及的所有副本）

如果使用 mclapply 对大数据进行并行计算效率低下，那么有什么更好的方法呢？我做这种事情的需求只会越来越大，而且我肯定到处都遇到瓶颈。我看到的教程是关于如何使用函数的非常基本的介绍，但不一定是如何有效地使用函数来管理权衡。该文档对这种权衡有一个小小的介绍：

mc.preschedule：如果设置为“TRUE”，则计算首先被划分到（最多）有多少工作有核心，然后作业开始，每个作业可能涵盖多个价值。如果设置为“FALSE”，则为每个作业派生一个作业 “X”的值。前者更适合短期计算或 ‘X’中的大量值，后者更适合工作完成时间差异较大且数量不多 “X”的值与“mc.cores”的比较

和

默认情况下（‘mc.preschedule = TRUE’）输入‘X’被分割成与核心一样多的部分（当前值已传播依次跨核心，即核心 1 的第一个值，第二个到核心 2，...（核心 + 1）-th 值到核心 1 等），然后是一个进程被分叉到每个核心并收集结果。

如果没有预先安排，则为每个值派生一个单独的作业 'X'。确保运行的作业不超过“mc.cores” 一次，一旦这个数字被分叉，主进程就会等待让孩子在下一次分叉前完成

可靠地对这些事物进行基准测试需要花费大量时间，因为有些问题只会在规模上体现出来，然后很难弄清楚到底发生了什么。因此，更好地了解函数的行为会很有帮助。

编辑：

我没有具体的示例，因为我经常使用 mclapply，并且想更好地了解如何考虑性能影响。虽然写入磁盘可以解决错误，但我认为这对于必须发生的（反）序列化没有帮助，这也必须通过磁盘 IO。

一个工作流程如下：取一个大的稀疏矩阵M，并将其分块写入磁盘（比如M1-M100），因为M 本身不适合内存。

现在说，对于I 中的每个用户i，M 中有Ci 列，我想在用户级别添加和聚合。对于较小的数据，这将是相对微不足道的：

m = matrix(runif(25), ncol=5)
df = data.frame(I=sample(1:6, 20, replace=T), C=sample(1:5, 20, replace=T))
somefun = function(m) rowSums(m)
res = sapply(sort(unique(df$I)), function(i) somefun(m[,df[df$I == i,]$C]))

但是对于更大的数据，我的方法是根据列所在的矩阵M1-M100 将用户/列的 data.frame 拆分为不同的 data.frames，对这些 data.frames 执行并行循环，读取在关联的矩阵中，然后循环遍历用户，提取列并应用我的函数，然后获取输出列表，然后再次循环并重新聚合。

如果我有一个不能像那样重新聚合的函数（到目前为止，这不是问题），这并不理想，但我显然用这种方法洗牌了太多数据。

【问题讨论】：

在不了解问题的结构的情况下，很难说出如何才能做得更好。我在错误报告中查看了您的示例，这可能只是为了证明您无法返回大对象。您可能想查看共享内存包。您需要发布一个最小的可重现示例，该示例可以捕获您的问题的结构，但又足够简单易懂。我认为工作安排（如mc.preschedule）不会让你有任何收获。
我已经发布了一个简单的例子来说明我正在做的事情，但复杂性来自于磁盘 IO、序列化和此类问题所发生的一切。我想要做的事情清楚吗？

标签： r parallel-processing overhead mclapply

【解决方案1】：

我希望我的回答还不算晚，但我认为您的示例可以通过bigmemory 包使用共享内存/文件来处理。

让我们创建数据

library(bigmemory)
library(parallel)

#your large file-backed matrix (all values initialized to 0)
#it can hold more than your RAM as it is written to a file
m=filebacked.big.matrix(nrow=5,
                        ncol=5,
                        type="double",
                        descriptorfile="file_backed_matrix.desc",
                        backingfile="file_backed_matrix",
                        backingpath="~")

#be careful how to fill the large matrix with data
set.seed(1234)
m[]=c(matrix(runif(25), ncol=5))
#print the data to the console
m[]

#your user-col mapping
#I have added a unique idx that will be used below
df = data.frame(unique_idx=1:20,
                I=sample(1:6, 20, replace=T),
                C=sample(1:5, 20, replace=T))

#the file-backed matrix that will hold the results
resm=filebacked.big.matrix(nrow=nrow(df),
                           ncol=2,
                           type="double",init = NA_real_,
                           descriptorfile="res_matrix.desc",
                           backingfile="res_backed_matrix",
                           backingpath="~")

#the first column of resm will hold the unique idx of df
resm[,1]=df$unique_idx
resm[]

现在，让我们转到您要执行的功能。你写了rowSums，但从你的文字推断你的意思是colSums。我相应地改变了它。

somefun = function(x) {
  #attach the file-backed big.matrix
  #it makes the matrix "known" to the R process (no copying involved!)
  #input
  tmp=attach.big.matrix("~/file_backed_matrix.desc")
  #output
  tmp_out=attach.big.matrix("~/res_matrix.desc")

  #store the output in the file-backed matrix resm
  tmp_out[x$unique_idx,2]=c(colSums(tmp[,x$C,drop=FALSE]))
  #return a little more than the colSum result
  list(pid=Sys.getpid(),
       user=x$I[1],
       col_idx=x$C)
}

在所有核上进行并行计算

#perform colSums using different threads
res=mclapply(split(df,df$I),somefun,mc.cores = detectCores())

检查结果

#processes IDs
unname(sapply(res,function(x) x$pid))
#28231 28232 28233 28234 28231 28232

#users
unname(sapply(res,function(x) x$user))
#1 2 3 4 5 6 

#column indexes
identical(sort(unname(unlist(sapply(res,function(x) x$col_idx)))),sort(df$C))
#[1] TRUE

#check result of colSums
identical(lapply(split(df,df$I),function(x) resm[x$unique_idx,2]),
          lapply(split(df,df$I),function(x) colSums(m[,x$C,drop=FALSE])))
#[1] TRUE

编辑：我已经在我的编辑中处理了您的评论。将结果存储在文件支持的输出矩阵 resm 中按预期工作。

【讨论】：

我尝试使用 Rdsm 和 bigmemory 包来做到这一点，将每个进程的结果写入矩阵的一列。我已将其设置为每个进程只会写入一次列。但是，在小玩具示例中结果最终不一样，所以通过共享内存在同一个矩阵中写入不同的列似乎不是线程安全的操作？我想我的代码中可能存在错误。稍后将尝试发布该尝试的记录。谢谢！
@James 我已经在我的编辑中处理了您的评论。我现在的答案还包括输出到共享矩阵。代码按预期工作。

【解决方案2】：

为了限制适度大的 N 的开销，使用mc.preschedule = TRUE 几乎总是更好（即将工作分成与内核一样多的块）。

看来您的主要权衡是在内存使用和 CPU 之间。也就是说，您只能并行化，直到正在进行的进程用尽您的 RAM。要考虑的一件事是，不同的工作人员可以在 R 会话中读取相同的对象而不会重复。因此，只有在并行函数调用中修改/创建的对象才会为每个内核添加它们的内存占用量。

如果您最大限度地使用内存，我的建议是将您的整个计算分成多个子作业并依次循环（例如使用 lapply），在该循环中调用 mclapply 以并行化每个子作业，并可能节省子作业的输出到磁盘以避免将其全部保存在内存中。

【讨论】：