R并行：rbind并行到单独的data.frames

Question

以下代码在 Windows 和 Ubuntu 平台上产生不同的结果。我理解这是因为处理并行处理的方法不同。

总结：
我不能 insert / rbind 在 Linux 上并行处理数据（mclapply，mcmapply），而 我可以在 Windows 上完成 .

感谢@Hong Ooi 指出mclapply 不能在Windows 上并行工作，但以下问题仍然有效。

当然，同一个data.frame没有多次插入，每次插入都在单独的data.frame中执行。

library(R6)
library(parallel)

# storage objects generator
cl <- R6Class(
    classname = "cl",
    public = list(
        data = data.frame(NULL),
        initialize = function() invisible(self),
        insert = function(x) self$data <- rbind(self$data, x)
    )
)

N <- 4L # number of entities
i <- setNames(seq_len(N),paste0("n",seq_len(N)))

# random data.frames
set.seed(1)
ldt <- lapply(i, function(i) data.frame(replicate(sample(3:10,1),sample(letters,1e5,rep=TRUE))))

# entity storage
lcl1 <- lapply(i, function(i) cl$new())
lcl2 <- lapply(i, function(i) cl$new())
lcl3 <- lapply(i, function(i) cl$new())

# insert data
invisible({
    mclapply(names(i), FUN = function(n) lcl1[[n]]$insert(ldt[[n]]))
    mcmapply(FUN = function(dt, cl) cl$insert(dt), ldt, lcl2, SIMPLIFY=FALSE)
    lapply(names(i), FUN = function(n) lcl3[[n]]$insert(ldt[[n]]))
})

### Windows

sapply(lcl1, function(cl) nrow(cl$data)) # mclapply
#     n1     n2     n3     n4
# 100000 100000 100000 100000
sapply(lcl2, function(cl) nrow(cl$data)) # mcmapply
#     n1     n2     n3     n4
# 100000 100000 100000 100000
sapply(lcl3, function(cl) nrow(cl$data)) # lapply
#     n1     n2     n3     n4
# 100000 100000 100000 100000

### Unix

sapply(lcl1, function(cl) nrow(cl$data)) # mclapply
#n1 n2 n3 n4
# 0  0  0  0
sapply(lcl2, function(cl) nrow(cl$data)) # mcmapply
#n1 n2 n3 n4
# 0  0  0  0
sapply(lcl3, function(cl) nrow(cl$data)) # lapply
#     n1     n2     n3     n4
# 100000 100000 100000 100000

还有问题：

如何在 Linux 平台上实现 rbind 到单独的 data.frames 中？

附：在我的情况下，像SQLite 这样的内存外存储不能被视为解决方案。

Answer 1

问题是mclapply 和mcmapply 不打算与有副作用的函数一起使用。您的函数正在修改列表中的对象，但mclapply 不会将修改后的对象发送回主进程：它只返回函数显式返回的值。这意味着当工作人员在mclapply 返回时退出时，您的结果将丢失。

通常我会将代码更改为不依赖于副作用，并返回已修改的对象。这是使用clusterApply 的一种方法，它也可以在 Windows 上并行工作：

library(R6)
library(parallel)
cl <- R6Class(
    classname = "cl",
    public = list(
        data = data.frame(NULL),
        initialize = function() invisible(self),
        insert = function(x) self$data <- rbind(self$data, x)))

N <- 4L # number of entities
i <- setNames(seq_len(N),paste0("n",seq_len(N)))
set.seed(1)
ldt <- lapply(i, function(i)
  data.frame(replicate(sample(3:10,1),sample(letters,1e5,rep=TRUE))))
nw <- 3  # number of workers
clust <- makePSOCKcluster(nw)
idx <- splitIndices(length(i), nw)
nameslist <- lapply(idx, function(iv) names(i)[iv])

lcl4 <- do.call('c', clusterApply(clust, nameslist, 
  function(nms, cl, ldt) {
    library(R6)
    lcl4 <- lapply(nms, function(n) cl$new())
    names(lcl4) <- nms
    lapply(nms, FUN = function(n) lcl4[[n]]$insert(ldt[[n]]))
    lcl4
  }, cl, ldt))

如果您想创建一次对象列表，然后并行修改多个对象，则此方法不起作用。这也是可能的，但你必须有持久的工人。在这种情况下，您会在所有任务完成后从工作人员那里获取修改后的对象。不幸的是，mclapply 没有使用持久化的worker，所以在这种情况下你必须使用基于集群的函数，例如clusterApply。这是一种方法：

# Initialize the cluster workers
clusterEvalQ(clust, library(R6))
clusterExport(clust, c('cl', 'ldt'))
clusterApply(clust, nameslist, function(nms) {
  x <- lapply(nms, function(n) cl$new())
  names(x) <- nms
  assign('lcl4', x, pos=.GlobalEnv)
  NULL
})

# Insert data into lcl4 on each worker
clusterApply(clust, nameslist, function(nms) {
  lapply(nms, FUN = function(n) lcl4[[n]]$insert(ldt[[n]]))
  NULL
})

# Concatenate lcl4 from each worker
lcl4 <- do.call('c', clusterEvalQ(clust, lcl4))

这和之前的方法很相似，只是将流程分为三个阶段：worker初始化、任务执行和结果检索。我还使用clusterExport 和clusterEvalQ 以更传统的方式初始化了worker。

Answer 2

我认为mclapply 的Windows 版本正在运行，因为它将其工作委托给lapply。检查时序或 CPU 核心使用情况可以验证这一点。根据R source，Windows 的mclapply 和mcmapply 被替换为顺序版本。

看来，代码的并行化方式有问题，目前看不出具体是什么。