Assignment of a value from a foreach loop

Question

I would like to parallelize a loop like

td        <- data.frame(cbind(c(rep(1,4),2,rep(1,5)),rep(1:10,2)))
names(td) <- c("val","id")

res <- rep(NA,NROW(td))
for(i in levels(interaction(td$id))){
res[td$id==i] <- mean(td$val[td$id!=i])
}  

with the help of foreach() of the library(doParallel) in order to speed up computations. Unfortunately foreach doesn't seem to support direct assignments, at least

registerDoParallel(4)
res <- rep(NA,NROW(td))
foreach(i=levels(interaction(td$id))) %dopar%{
res[td$id==i] <- mean(td$val[td$id!=i])}

不做我想要的(给出与上面的正常循环相同的结果).任何想法我做错了什么或我怎么能以某种方式"破解" foreach中的.combine选项以便做我想要的？请注意,id变量的顺序在原始数据集中并不总是相同.任何提示都将非常感谢!

Answer 1

要有效地并行执行这些计算,您需要使用分块,因为单独的平均计算不会花费太多时间.使用时foreach,我经常使用itertools包中的函数进行分块.在这种情况下,我使用该isplitVector函数以便为每个工人生成一个任务.结果是向量,因此只需将它们加在一起即可将它们组合在一起,这就是为什么r向量必须初始化为零向量的原因.

vadd <- function(a, ...) {
  for (v in list(...))
    a <- a + v
  a
}

res <- foreach(ids=isplitVector(unique(td$id), chunks=workers),
               .combine='vadd',
               .multicombine=TRUE,
               .inorder=FALSE) %dopar% {
  r <- rep(0, NROW(td))
  for (i in ids)
    r[td$id == i] <- mean(td$val[td$id != i])
  r
}

这是将原始顺序版本放入foreach循环中的典型示例,但仅对数据的子集进行操作.由于只有一个结果可以为每个工人组合,因此后处理非常少,因此它可以非常有效地运行.

为了了解这是如何执行的,我使用以下数据集对照顺序版本和Rolands的数据表版本进行基准测试:

set.seed(107)
n <- 1000000
m <- 10000
td <- data.frame(val=rnorm(n), id=sample(m, n, replace=TRUE))

我包括这个因为性能非常依赖于数据.您甚至可以通过使用不同的随机种子获得不同的性能结果.

以下是我的Linux机箱带有Xeon CPU X5650和12 GB RAM的一些基准测试结果:

• 循环顺序:359秒
• 顺序数据表版本:208秒
• foreach/doParallel/PSOCK与4名工人:104秒

因此,对于至少一个数据集,值得并行执行该计算.这不是一个完美的加速,但它并不太糟糕.要在您自己的计算机上运行任何这些基准测试,或使用不同的数据集,您可以通过上面的链接从pastebin下载它们.

更新

在完成这些基准测试后,我有兴趣使用data.table它foreach来获得更快的版本.这就是我提出的建议(来自Matthew Dowle的建议):

cmean <- function(v, mine) if (mine) mean(v) else 0
nuniq <- length(unique(td$id))
res <- foreach(grps=isplitIndices(nuniq, chunks=workers),
               .combine='vadd',
               .multicombine=TRUE,
               .inorder=FALSE,
               .packages='data.table') %dopar% {
  td[, means := cmean(td$val[-.I], .GRP %in% grps), by=id]
  td$means
}

td现在是一个data.table对象.我isplitIndices从itertools包中使用生成与每个任务块相关联的组号的向量.该cmean函数是一个包装器mean,对于不应在该任务块中计算的组返回零.它使用与非数据表版本相同的组合功能,因为任务结果是相同的.

有四个工作人员和相同的数据集,这个版本运行56.4秒,与顺序数据表版本相比,速度提高了3.7,使其成为明显的赢家,比顺序for循环快6.4倍.基准测试可以从pastebin下载到这里.

Answer 2

如果您使用data.table而不是并行化循环,那么您的性能提升将会提高几个数量级:

library(data.table)
DT <- data.table(td)

DT[, means := mean(DT[-.I, val]), by = id]

identical(DT$means, res)
#[1] TRUE

如果你想使用foreach你需要将它与一个merge:

library(foreach)
res2 <- foreach(i=levels(interaction(td$id)), .combine=rbind) %do% {
  data.frame(level = i, means = mean(td$val[td$id!=i]))}

res2 <- merge(res2, td, by.x = "level", by.y = "id", sort = FALSE)

#    level    means val
# 1      1 1.111111   1
# 2      1 1.111111   1
# 3      2 1.111111   1
# 4      2 1.111111   1
# 5      3 1.111111   1
# 6      3 1.111111   1
# 7      4 1.111111   1
# 8      4 1.111111   1
# 9      5 1.000000   2
# 10     5 1.000000   2
# 11     6 1.111111   1
# 12     6 1.111111   1
# 13     7 1.111111   1
# 14     7 1.111111   1
# 15     8 1.111111   1
# 16     8 1.111111   1
# 17     9 1.111111   1
# 18     9 1.111111   1
# 19    10 1.111111   1
# 20    10 1.111111   1