R：异步并行lapply

Question

到目前为止，我发现lapply在 R 中使用并行的最简单方法是通过以下示例代码：

library(parallel)
library(pbapply)

cl <- makeCluster(10)
clusterExport(cl = cl, {...})
clusterEvalQ(cl = cl, {...})

results <- pblapply(1:100, FUN = function(x){rnorm(x)}, cl = cl)

它有一个非常有用的功能，即为结果提供进度条，并且当​​不需要并行计算时，通过设置 可以很容易地重用相同的代码cl = NULL。

然而，我注意到的一个问题是它pblapply正在批量循环遍历列表。例如，如果一个工人在某项任务上停留了很长时间，那么剩下的工人将等待该任务完成，然后再开始一批新的工作。对于某些任务，这会为工作流程增加大量不必要的时间。

我的问题： 是否有任何类似的并行框架允许工作人员独立运行？进度条和重用代码的能力cl=NULL将是一个很大的优势。

也许可以修改现有代码pbapply来添加此选项/功能？

Answer 1

（免责声明：我是future框架和Progressr包的作者）

base::lapply()类似于和您的示例的一个接近的解决方案pbapply::pblapply()是将future.apply用作：

library(future.apply)

## The below is same as plan(multisession, workers=4)
cl <- parallel::makeCluster(4)
plan(cluster, workers=cl)

xs <- 1:100
results <- future_lapply(xs, FUN=function(x) {
  Sys.sleep(0.1)
  sqrt(x)
})

分块future.chunk.size：您可以通过参数或补充来控制分块量future.schedule。要禁用分块以便在唯一的并行任务中处理每个元素，请使用future.chunk.size=1。这样，如果有一个元素比其他元素花费的时间长得多，它就不会占用任何其他元素。

xs <- 1:100
results <- future_lapply(xs, FUN=function(x) {
  Sys.sleep(0.1)
  sqrt(x)
}, future.chunk.size=1)

并行进度更新： 如果您希望在进行并行处理时接收进度更新，可以使用progressr包并将其配置为使用progress包以进度条的形式报告更新（这里也带有ETA）。

library(future.apply)
plan(multisession, workers=4)

library(progressr)
handlers(handler_progress(format="[:bar] :percent :eta :message"))

with_progress({
  p <- progressor(along=xs)
  results <- future_lapply(xs, FUN=function(x) {
    p()  ## signal progress
    Sys.sleep(0.1)
    sqrt(x)
  }, future.chunk.size=1)
})

您可以将其包装到一个函数中，例如

my_fcn <- function(xs) {
  p <- progressor(along=xs)
  future_lapply(xs, FUN=function(x) {
    p()
    Sys.sleep(0.1)
    sqrt(x)
  }, future.chunk.size=1)
}

这样你就可以将它作为常规函数调用：

> result <- my_fcn(xs)

并用于plan()精确控制您希望它如何并行化。这不会报告进度。为此，您必须执行以下操作：

> with_progress(result <- my_fcn(xs))
[====>-----------------------------------------------------]   9%  1m

在后台运行所有内容：如果您的问题是如何在后台运行整个 shebang，请参阅“未来拓扑”小插图。这是另一个级别的并行化，但这是可能的。