通过使用多核和并行编程加速 data.table 组答案

【问题标题】：Speed-up data.table group by using multiple cores and parallel programming通过使用多核和并行编程加速 data.table 组
【发布时间】：2013-10-05 15:24:17
【问题描述】：

我的代码很大，聚合步骤是当前速度方面的瓶颈。

在我的代码中，我希望加快数据分组步骤的速度。我的数据的 SNOTE（简单的非平凡示例）如下所示：

library(data.table)
a = sample(1:10000000, 50000000, replace = TRUE)
b = sample(c("3m","2m2d2m","3m2d1i3s2d","5m","4m","9m","1m"), 50000000, replace = TRUE)
d = sample(c("3m","2m2d2m","3m2d1i3s2d","5m","4m","9m","1m"), 50000000, replace = TRUE)
e = a
dt = data.table(a = a, b = b, d = d, e = e)
system.time(c.dt <- dt[,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[1], by=a)])
   user  system elapsed 
 60.107   3.143  63.534

对于如此大的数据示例，这相当快，但就我而言，我仍在寻找进一步的加速。就我而言，我有多个内核，所以我几乎可以肯定一定有一种方法可以使用这种计算能力。

我愿意将我的数据类型更改为 data.frame 或 idata.frame 对象（理论上 idata.frame 应该比 data.frames 更快）。

我做了一些研究，似乎 plyr 包有一些并行功能可能会有所帮助，但我仍在努力为我正在尝试做的分组做这件事。在another SO post they discuss some of these ideas。由于它使用了 foreach 函数，我仍然不确定通过这种并行化可以实现多少。根据我的经验，foreach function 对于数百万个快速操作来说并不是一个好主意，因为内核之间的通信工作最终会减慢并行化工作。

【问题讨论】：

请更具体地了解“连接”和“聚合”这两个词的含义。这些让人想到的功能是3：list、c和paste。该代码的功能是什么。我们是从数据框中提取列还是处理 data.tables？ 'block.read.parent.cigar' 和其他输入变量的结构是什么.....更好地解释了这个问题！（显然其他人同意。这不是我的反对意见。）
@Dwin，谢谢！我不确定我在 Q 中是否澄清得足够清楚，但基本的 Q 是如何加快对上例中的大型数据表的聚合操作。还要记住，我可以使用多个内核，因此可能会有一些智能并行化想法可以大大加快此类操作。希望这会有所帮助，我添加了一个示例
我没有投反对票。但我会这样做的原因是您没有提供有关数据的任何信息。如果read.index 是行索引，那么将每一行单独分组到一行当然会很慢。您将致电paste 数百万次。你用Rprof了吗？你用verbose=TRUE了吗？而且您使用诸如“太慢”之类的词而没有给出数字。事实上，我现在已经说服自己反对它。如果你改进了问题，它可以逆转。
@Dnaiel 现在这是一个很好的问题。 +1。我会试着看看。我猜有些回答者只是有新的问题提要，所以为了获得更多关注，提供赏金可能是一个想法。
@MattDowle 非常感谢，我很高兴我改进了这种令人困惑的 Q :-) 不确定它有多棒，但这就是我正在处理的问题。我正在学习如何提出更好的问题，这对我有好处。

标签： r data.table mclapply

【解决方案1】：

如果您有多个可用的内核，为什么不利用您可以使用其键快速过滤和分组 data.table 中的行的事实：

library(doMC)
registerDoMC(cores=4)


setkey(dt, "a")

finalRowOrderMatters = FALSE # FALSE can be faster
foreach(x=unique(dt[["a"]]), .combine="rbind", .inorder=finalRowOrderMatters) %dopar% 
     dt[.(x) ,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[[1]])]

请注意，如果唯一组的数量（即 length(unique(a)) ）相对较少，则删除 .combine 参数会更快，将结果返回到列表中，然后在结果上调用 rbindlist。在我对两个内核和 8GB RAM 的测试中，阈值约为 9,000 个唯一值。这是我用来进行基准测试的内容：

# (otion a)
round(rowMeans(replicate(3, system.time({
# ------- #
  foreach(x=unique(dt[["a"]]), .combine="rbind", .inorder=FALSE) %dopar% 
     dt[.(x) ,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[[1]])]
# ------- #
}))), 3) 
# [1]  1.243 elapsed for N ==  1,000
# [1] 11.540 elapsed for N == 10,000, length(unique(dt[["a"]])) == 8617
# [1] 57.404 elapsed for N == 50,000



# (otion b)
round(rowMeans(replicate(3, system.time({
# ------- #
    results <- 
      foreach(x=unique(dt[["a"]])) %dopar% 
         dt[.(x) ,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[[1]])]
    rbindlist(results)
# ------- #
}))), 3)
# [1]  1.117 elapsed for N ==  1,000
# [1] 10.567 elapsed for N == 10,000, length(unique(dt[["a"]])) == 8617
# [1] 76.613 elapsed for N == 50,000


## And used the following to create the dt
N <- 5e4
set.seed(1)
a = sample(1:N, N*2, replace = TRUE)
b = sample(c("3m","2m2d2m","3m2d1i3s2d","5m","4m","9m","1m"), N*2, replace = TRUE)
d = sample(c("3m","2m2d2m","3m2d1i3s2d","5m","4m","9m","1m"), N*2, replace = TRUE)
e = a
dt = data.table(a = a, b = b, d = d, e = e, key="a")

【讨论】：

每个子进程是否需要复制完整的data.table，还是都访问“主”data.table对象？

【解决方案2】：

你能用data.table并行聚合吗？是的。

值得吗？否。这是上一个答案未能突出的关键点。

正如Matt Dowle 在data.table and parallel computing 中解释的那样，在并行运行操作时需要在分发之前制作副本（“块”）。这会减慢速度。在某些情况下，当您不能使用data.table（例如运行许多线性回归）时，值得在内核之间拆分任务。但不是聚合——至少在涉及data.table 时。

简而言之（除非另有证明），使用 data.table 进行聚合，并不要担心使用 doMC 可能会提高速度。 data.table 在聚合方面已经比其他任何可用的东西都快——即使它不是多核的！

这里有一些您可以自己运行的基准测试，比较使用by 与foreach 和mclapply 的data.table 内部聚合。结果列在最前面。

#-----------------------------------------------

# TL;DR FINAL RESULTS (Best to Worst)
# 3 replications, N = 10000:
# (1)  0.007 -- data.table using `by`
# (2)  3.548 -- mclapply with rbindlist
# (3)  5.557 -- foreach with rbindlist
# (4)  5.959 -- foreach with .combine = "rbind"
# (5) 14.029 -- lapply

# ----------------------------------------------

library(data.table)

## And used the following to create the dt
N <- 1e4
set.seed(1)
a = sample(1:N, N*2, replace = TRUE)
b = sample(c("3m","2m2d2m","3m2d1i3s2d","5m","4m","9m","1m"), N*2, replace = TRUE)
d = sample(c("3m","2m2d2m","3m2d1i3s2d","5m","4m","9m","1m"), N*2, replace = TRUE)
e = a
dt = data.table(a = a, b = b, d = d, e = e, key="a")
setkey(dt, "a")

# TEST AGGREGATION WITHOUT PARALLELIZATION ---------------------------
## using data.tables `by` to aggregate
round(rowMeans(replicate(3, system.time({
    dt[,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[1], by=a)]
}))), 3)
# [1] 0.007 elapsed for N == 10,000, length(unique(dt[["a"]])) == 8617

## using `lapply`
round(rowMeans(replicate(3, system.time({
    results <- lapply(unique(dt[["a"]]), function(x) {
        dt[.(x), list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[1])]
    })
    rbindlist(results)
}))), 3)
# [1] 14.029 elapsed for N == 10,000

# USING `mclapply` FORKING ---------------------------------
## use mclapply
round(rowMeans(replicate(3, system.time({
    results <- mclapply(unique(dt[["a"]]),
    function(x) {
        dt[.(x), list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[[1]])]
    }, mc.cores=4)
    rbindlist(results)
}))), 3)
# [1] 3.548 elapsed for N == 10,000


# PARALLELIZATION USING `doMC` PACKAGE ---------------------------------
library(doMC)
mc = 4
registerDoMC(cores=mc)
getDoParWorkers()
# [1] 4

## (option a) by Ricardo Saporta
round(rowMeans(replicate(3, system.time({
    foreach(x=unique(dt[["a"]]), .combine="rbind", .inorder=FALSE) %dopar%
    dt[.(x) ,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[[1]])]
}))), 3)
# [1] 5.959 elapsed for N == 10,000

## (option b) by Ricardo Saporta
round(rowMeans(replicate(3, system.time({
    results <-
      foreach(x=unique(dt[["a"]])) %dopar%
        dt[.(x) ,list(b = paste(b, collapse=""), d = paste(d, collapse=""), e = e[[1]])]
    rbindlist(results)
}))), 3)
# [1] 5.557 elapsed for N == 10,000

registerDoSEQ()
getDoParWorkers()
# [1] 1

【讨论】：