如何并行化 R 中包的函数

Posted 2023-02-15

【中文标题】如何并行化 R 中包的函数

【英文标题】：How to parallelize a function for a package in R

【发布时间】：2019-05-07 03:42:57

【问题描述】：

我想并行化我正在处理的包的一部分。我应该使用哪些包和什么语法来使包灵活且可用于不同的架构？我的问题在于单个 sapply() 调用，如此模拟代码所示：

.heavyStuff <- function(x)  
   # do a lot of work
   Sys.sleep(1)


listOfX <- 1:20

userFunc1 <- function(listOfX)     
  res <- sapply(listOfX, .heavyStuff)
  return(res)

根据不同的指南，我炮制了以下内容：

userFunc2 <- function(listOfX, dopar.arg=2) 
  if(requireNamespace("doParallel")) 
    doParallel::registerDoParallel(dopar.arg)
    res <- foreach(i=1:length(listOfX)) %dopar% 
       .heavyStuff(listOfX[[i]])
    
    names(res) <- names(listOfX)
   else 
    res <- sapply(listOfX, .heavyStuff)
  
  return(res)

问题：

我可以安全地在包中使用这样的代码吗？它能否在各种平台上正常运行？ 有没有办法避免foreach() 构造？我更喜欢使用类似 sapply 或 lapply 的功能。但是，并行库中的结构似乎更加特定于平台。 如果dopar.arg==NULL，上面的代码不起作用，即使the introduction to doParallel says that没有任何参数“你会得到三个工人和类Unix系统 您将获得大约相当于系统内核数量一半的工作人员数量。”

【参考方案1】：

作为future框架的作者，我建议你看看future.apply包，例如

library(future.apply)
userFunc2 <- function(listOfX)     
  res <- future_sapply(listOfX, .heavyStuff)
  return(res)

默认情况是按顺序运行，但如果用户愿意，他们可以使用他们想要的任何并行的未来后端，例如

library(future)
plan(multiprocess)    # parallel on local machine - all cores by default

library(future.batchtools)
plan(batchtools_sge)  # parallel on an SGE compute cluster

library(future)
plan(sequential)      # sequentially

设计模式是您决定什么并行化，而用户如何并行化。

【讨论】：

这是一个非常有趣的解决方案。它看起来非常直观且非常灵活。我看到的唯一缺点是允许用户向plan() 提供额外的参数将需要额外的代码（我已经将... 用于其他内容）。运行顺序 future_sapply 而不是 sapply 时是否有很多开销？

让用户来控制plan() 的想法主要是因为作为一个包开发者/维护者，你无法真正知道用户有哪些可用的计算资源。此外，您无法预测未来可能会有哪些新的并行后端可用。 （这个想法不是一成不变的，可以解决，但它是一个很好的设计目标）。

future_sapply() 与 sapply() plan(sequential) 时的开销：首先，目标是将其保持在最低限度。话虽如此，可能还有改进的空间（还没有把优化推到那么远）。全局变量的自动识别会产生一些开销，因此如果您知道作为开发人员可以手动指定这些变量（参数future.globals）。此外，根据经验，如果存在并行工作的代码，则意味着涉及大量处理时间，在这种情况下增加的开销应该相对较小。

不，我完全明白为什么您需要将计划留给用户；这与我原来的解决方案（dopar.arg）没有什么不同。唯一的问题是我需要多个参数（并且它们的名称未知，因为它们取决于计划）才能传递给我的函数（而在 foreach %dopar% 解决方案中，只传递一个参数就足够了)。

因此，您的 foreach 示例仅假定本地计算机上的并行化。如果这就是您想要支持的全部内容，您可以随时通过nworkers 致电plan(multiprocess, workers = nworkers)。但是，如果您这样做，您需要确保撤消您的 plan() 设置，这样您就不会弄乱/覆盖用户已经为其他目的设置的内容。另见github.com/HenrikBengtsson/future/issues/263