在 R 中并行化向量化函数的最简单方法是什么？

Question

我有一个非常大的列表X和一个向量化函数f。我想计算f(X)，但是如果我用单核来计算，这将需要很长时间。我有（访问）一台 48 核服务器。并行计算 f(X) 的最简单方法是什么？以下不是正确答案：

library(foreach)
library(doMC)
registerDoMC()

foreach(x=X, .combine=c) %dopar% f(x)

上面的代码确实会并行计算f(X)，但它是通过应用f来实现的> 分别到 X 的每个元素。这忽略了 f 的矢量化性质，并且可能会导致速度变慢，而不是更快。我不想将 f 按元素应用于 X，而是希望将 X 拆分为合理大小的块，并将 f 应用于那些。

那么，我是否应该手动将 X 拆分为 48 个大小相等的子列表，然后然后并行地将 f 应用于每个子列表，然后手动将结果放在一起？或者有专门为此设计的包吗？

如果有人想知道，我的具体用例是此处。

Answer 1

虽然这是一个较老的问题，但对于通过谷歌偶然发现这个问题的每个人（像我一样）来说，这可能会很有趣：看看 multicore 包中的 pvec 函数。我认为它正是你想要的。

Answer 2

这是我的实现。这是一个函数chunkmap，它需要一个
向量化函数，应向量化的参数列表，
以及不应向量化的参数列表（即
常量），并返回与调用该函数相同的结果
直接参数，但结果是并行计算的。
对于函数 f，向量参数 v1、v2、v3 和标量
参数 s1、s2，以下内容应返回相同的结果：

f(a=v1, b=v2, c=v3, d=s1, e=s2)
f(c=v3, b=v2, e=s2, a=v1, d=s1)
chunkapply(FUN=f, VECTOR.ARGS=list(a=v1, b=v2, c=v3), SCALAR.ARGS=list(d=s1, e=s2))
chunkapply(FUN=f, SCALAR.ARGS=list(e=s2, d=s1), VECTOR.ARGS=list(a=v1, c=v3, b=v2))

因为 chunkapply 函数不可能知道哪个
f 的参数是向量化的，哪些不是，这取决于你
调用时指定，否则会得到错误的结果。你
通常应该命名你的参数以确保它们受到约束
正确。

library(foreach)
library(iterators)
# Use your favorite doPar backend here
library(doMC)
registerDoMC()

get.chunk.size <- function(vec.length,
                           min.chunk.size=NULL, max.chunk.size=NULL,
                           max.chunks=NULL) {
  if (is.null(max.chunks)) {
    max.chunks <- getDoParWorkers()
  }
  size <- vec.length / max.chunks
  if (!is.null(max.chunk.size)) {
    size <- min(size, max.chunk.size)
  }
  if (!is.null(min.chunk.size)) {
    size <- max(size, min.chunk.size)
  }
  num.chunks <- ceiling(vec.length / size)
  actual.size <- ceiling(vec.length / num.chunks)
  return(actual.size)
}

ichunk.vectors <- function(vectors=NULL,
                           min.chunk.size=NULL,
                           max.chunk.size=NULL,
                           max.chunks=NULL) {
  ## Calculate number of chunks
  recycle.length <- max(sapply(vectors, length))
  actual.chunk.size <- get.chunk.size(recycle.length, min.chunk.size, max.chunk.size, max.chunks)
  num.chunks <- ceiling(recycle.length / actual.chunk.size)

  ## Make the chunk iterator
  i <- 1
  it <- idiv(recycle.length, chunks=num.chunks)
  nextEl <- function() {
    n <- nextElem(it)
    ix <- seq(i, length = n)
    i <<- i + n
    vchunks <- foreach(v=vectors) %do% v[1+ (ix-1) %% length(v)]
    names(vchunks) <- names(vectors)
    vchunks
  }
  obj <- list(nextElem = nextEl)
  class(obj) <- c("ichunk", "abstractiter", "iter")
  obj
}

chunkapply <- function(FUN, VECTOR.ARGS, SCALAR.ARGS=list(), MERGE=TRUE, ...) {
  ## Check that the arguments make sense
  stopifnot(is.list(VECTOR.ARGS))
  stopifnot(length(VECTOR.ARGS) >= 1)
  stopifnot(is.list(SCALAR.ARGS))
  ## Choose appropriate combine function
  if (MERGE) {
    combine.fun <- append
  } else {
    combine.fun <- foreach:::defcombine
  }
  ## Chunk and apply, and maybe merge
  foreach(vchunk=ichunk.vectors(vectors=VECTOR.ARGS, ...),
          .combine=combine.fun,
          .options.multicore = mcoptions) %dopar%
  {
    do.call(FUN, args=append(vchunk, SCALAR.ARGS))
  }
}

## Only do chunkapply if it will run in parallel
maybe.chunkapply <- function(FUN, VECTOR.ARGS, SCALAR.ARGS=list(), ...) {
  if (getDoParWorkers() > 1) {
    chunkapply(FUN, VECTOR.ARGS, SCALAR.ARGS, ...)
  } else {
    do.call(FUN, append(VECTOR.ARGS, SCALAR.ARGS))
  }
}

以下一些示例显示 chunkapply(f,list(x)) 生成与 f(x) 相同的结果。我已将 max.chunk.size 设置得非常小，以确保实际使用分块算法。

> # Generate all even integers from 2 to 100 inclusive
> identical(chunkapply(function(x,y) x*y, list(1:50), list(2), max.chunk.size=10), 1:50 * 2)
[1] TRUE

> ## Sample from a standard normal distribution, then discard values greater than 1
> a <- rnorm(n=100)
> cutoff <- 1
> identical(chunkapply(function(x,limit) x[x<=limit], list(x=a), list(limit=cutoff), max.chunk.size=10), a[a<cutoff])
[1] TRUE

如果有人有比“chunkapply”更好的名字，请推荐。

编辑：

正如另一个答案指出的那样，多核包中有一个名为 pvec 的函数，它的功能与我编写的功能非常相似。对于简单的情况，您应该这样做，并且您应该投票赞成 Jonas Rauch 的答案。然而，我的函数有点更通用，所以如果以下任何一个适用于您，您可能需要考虑使用我的函数：

• 您需要使用多核以外的并行后端（例如 MPI）。我的函数使用 foreach，因此您可以使用任何为 foreach 提供后端的并行化框架。
• 您需要传递多个向量化参数。 pvec 仅对单个参数进行矢量化，因此您无法轻松地使用 pvec 实现并行矢量化加法。我的函数允许您指定任意参数。

Answer 3

itertools 包就是为了解决此类问题而设计的。在这种情况下，我将使用 isplitVector：

n <- getDoParWorkers()
foreach(x=isplitVector(X, chunks=n), .combine='c') %dopar% f(x)

对于此示例，pvec 无疑更快、更简单，但是这可以在 Windows 上与 doParallel 包一起使用。

Answer 4

Map-Reduce 可能就是您正在寻找的；它已移植到R

Answer 5

像这样的事情怎么样？ R 将利用所有可用内存，而多核将在所有可用核心上并行化。

library(multicore)
result = mclapply(X, function,mc.preschedule=FALSE, mc.set.seed=FALSE)