r - 将 data.tables 的不同子集导出到集群中的每个节点

Question

我正在编写一个函数来处理几个非常大的 data.tables，我想在 Windows 机器上并行化这个函数。

我可以通过使用 clusterExport 的 snow 包来为集群中的每个节点创建每个 data.tables 的副本。但是，这不起作用，因为它使用了太多内存。

我想通过将 data.tables 的不同子集导出到每个节点来解决这个问题，但是，我看不到如何在 snow 包中执行此操作。

这是一个可以运行但内存效率低下的玩具代码示例:

library(snow)
dd <- data.frame(a = rep(1:5, each = 2), b = 11:20)
cl <- makeCluster(2, type = "SOCK")
clusterExport(cl = cl, "dd")
clusterApply(cl, x = c(2,7),  function(thresh) colMeans(dd[dd$a < thresh,]))
stopCluster(cl)

这是一个代码示例，它不起作用但解释了我们希望如何将 dd 的子集分发到节点:

library(snow)
dd <- data.frame(a = rep(1:5, each = 2), b = 11:20)
cl <- makeCluster(2, type = "SOCK")

dd_exports <- lapply(c(2,7), function(thresh) dd[dd$a < thresh])
#Now we export the ith element of dd_exports to the ith node:
clusterExport(cl = cl, dd_exports) 
clusterApply(cl, x = c(2,7),  function(x) colMeans(dd))
stopCluster(cl)

Answer 1

因为 cl 是一个列表，所以在 clusterExport 调用期间简单地对其进行子集化:

library(data.table)
library(parallel)

dt <- data.table(a = rep(1:5, each = 2), b = 11:20)
cl <- makeCluster(2)
idx <- c(2, 7)

for (i in seq_along(cl)) {
  dt2 <- dt[a < idx[i]]
  clusterExport(cl[i], "dt2")
}
rm(dt2)
clusterEvalQ(cl, colMeans(dt2))
#> [[1]]
#>    a    b 
#>  1.0 11.5 
#> 
#> [[2]]
#>    a    b 
#>  3.0 15.5